from common import *
g=pd.read_parquet(R/'games.parquet');m=g[g.mature].copy();m['mature_n']=m.groupby('developer_id').appid.transform('size');m['strong']=m.reviews.ge(1000);m['price_band']=m.price_band.fillna('missing').astype(str)
rows=[]
for dev,a in m[['developer_id','appid','reviews']].groupby('developer_id',sort=False):
    past=[]
    for row in a.itertuples():
        streak=0
        for v in reversed(past):
            if v>=100:break
            streak+=1
        h=[i for i,v in enumerate(past) if v>=1000]
        rows.append(dict(appid=row.appid,quiet_streak=streak,releases_since_hit=len(past)-1-h[-1] if h else np.nan,first_hit_index=h[0]+1 if h else np.nan,prior_moderates=sum(100<=v<1000 for v in past)))
        past.append(row.reviews)
m=m.merge(pd.DataFrame(rows),on='appid',validate='one_to_one')
x=m[m['index'].ge(2)&m.date.ge(pd.Timestamp('2019-01-01',tz='UTC'))&m.first_date.ge(pd.Timestamp('2010-01-01',tz='UTC'))&m.gap_days.ge(30)].copy()
x['prior_hits_bin']=pd.cut(x.prior_hits_1000,[-1,0,1,2,np.inf],labels=['none','one','two','three_plus']).astype(str)
x['prior_peak_band']=pd.cut(x.prior_best,[-1,999,2999,9999,29999,np.inf],labels=['<1000','1000-2999','3000-9999','10000-29999','30000+']).astype(str)
x['prior_n_band']=pd.cut(x['index']-1,[0,2,4,9,np.inf],labels=['1-2','3-4','5-9','10+']).astype(str)
x['quiet_streak_bin']=pd.cut(x.quiet_streak,[-1,0,1,2,np.inf],labels=['none','one','two','three_plus']).astype(str)
all_latest=x.sort_values('date').drop_duplicates('developer_id',keep='last');small=x[x.mature_n.le(10)].sort_values('date').drop_duplicates('developer_id',keep='last')
small.to_parquet(R/'latest_transitions.parquet',index=False);all_latest.to_parquet(R/'all_latest_transitions.parquet',index=False)
results=[];quiet=[];contrast=[]
for scope,pool in [('catalog_le10',small),('all_catalog_sizes',all_latest)]:
    for group,a in pool.groupby('prior_hits_bin'):
        results.append(dict(scope=scope,prior_hits=group,**stats(a),**compare(a,pool[pool.prior_hits_bin.ne(group)],['year','price_band','prior_peak_band','prior_n_band'])))
    for peak,b in pool[pool.prior_best.ge(1000)].groupby('prior_peak_band'):
        for group,a in b.groupby('prior_hits_bin'):contrast.append(dict(scope=scope,prior_peak=peak,prior_hits=group,**stats(a)))
    for history,mask in [('no_earlier_hit',pool.prior_hits_1000.eq(0)),('one_earlier_hit',pool.prior_hits_1000.eq(1)),('two_or_more_hits',pool.prior_hits_1000.ge(2))]:
        base=pool[mask]
        for streak,a in base.groupby('quiet_streak_bin'):
            quiet.append(dict(scope=scope,history=history,quiet_streak=streak,**stats(a),**compare(a,base[base.quiet_streak_bin.ne(streak)],['year','price_band','prior_peak_band'])))
save(results,'repeat_hit_information.csv');save(contrast,'repeat_hits_by_peak.csv');save(quiet,'quiet_streak_outcomes.csv')
# Explicit contrast: two modest hits vs one much larger peak, descriptive only.
p=small
groups={'two_or_more_1000_4999_peaks':p.prior_hits_1000.ge(2)&p.prior_best.between(1000,4999),'one_10000_19999_peak':p.prior_hits_1000.eq(1)&p.prior_best.between(10000,19999),'one_1000_4999_peak':p.prior_hits_1000.eq(1)&p.prior_best.between(1000,4999)}
save([dict(group=k,**stats(p[mask])) for k,mask in groups.items()],'two_modest_vs_one_large.csv')
# Next observed release vs no next release after quiet states; fixed three-year window.
presence=[]
for dev,a in m.groupby('developer_id',sort=False):
    if a.first_date.iloc[0]<pd.Timestamp('2010-01-01',tz='UTC'):continue
    qualified=a[a.date.between(pd.Timestamp('2015-01-01',tz='UTC'),pd.Timestamp('2022-08-31',tz='UTC'))&a.reviews.lt(100)&a['index'].le(10)]
    if qualified.empty:continue
    state=qualified.iloc[-1];prior=a[a['index'].le(state['index'])];h=int(prior.reviews.ge(1000).sum());after=a[a.date.gt(state.date)&a.date.le(state.date+pd.Timedelta(days=365.25*3))]
    presence.append(dict(developer_id=dev,credit=state.credit,state_appid=state.appid,date=state.date,earlier_hits='none' if h==0 else 'one' if h==1 else 'two_plus',returned=len(after)>0,next_reviews=after.reviews.iloc[0] if len(after) else np.nan,next1000=bool(len(after) and after.reviews.iloc[0]>=1000),any1000=after.reviews.ge(1000).any()))
pr=pd.DataFrame(presence);pr.to_csv(R/'quiet_state_presence_members.csv',index=False)
save([dict(earlier_hits=k,n=len(a),returned=int(a.returned.sum()),return_rate=a.returned.mean(),any1000_all=a.any1000.mean(),next1000_returners=a.loc[a.returned,'next1000'].mean()) for k,a in pr.groupby('earlier_hits')],'quiet_state_presence.csv')
print('Latest transitions small/all',len(small),len(all_latest),flush=True)