from pathlib import Path
import pandas as pd,numpy as np,json,math
R=Path(__file__).resolve().parent;s=pd.read_parquet(R/'sample.parquet');end=pd.Timestamp('2026-09-01',tz='UTC');recent_start=pd.Timestamp('2025-09-01',tz='UTC')
status=json.loads((R/'status.json').read_text());done={x['appid']:x for x in status};rows=[];monthlyrows=[];bursts=[]
for game in s.itertuples():
    f=R/'completed'/f'{game.appid}.parquet'
    if not f.exists():continue
    r=pd.read_parquet(f);dates=pd.to_datetime(r.timestamp_created,unit='s',utc=True);n=len(r)
    offset=(dates.min()-game.date).total_seconds()/86400 if n else np.nan
    valid=not(n and offset < -7)
    row=dict(appid=game.appid,name=game.name,date=game.date,period=game.period,review_band=game.review_band,stratum=game.stratum,order=game.order,population_n=game.population_n,weight=game.weight,price=game.price,primary=game.primary,known_ea=game.known_ea,snapshot_reviews=game.reviews,collected=n,summary_total=done[game.appid]['summary_total'],reconciliation_difference=n-done[game.appid]['summary_total'],first_review=dates.min() if n else None,first_review_offset_days=offset,release_clock_valid=valid,through_august=int(dates.lt(end).sum()))
    for clock,origin in [('release',game.date),('firstreview',dates.min() if n else pd.NaT)]:
        for days in [7,30,90,365,730]:
            row[f'{clock}_d{days}']=int(dates.lt(origin+pd.Timedelta(days=days)).sum()) if pd.notna(origin) and origin+pd.Timedelta(days=days)<=end and (clock!='release' or valid) else np.nan
    if valid:
        n90=row['release_d90'];n365=row['release_d365'];n730=row['release_d730']
        row['post90_year1']=n365-n90;row['year1_after90_share']=(n365-n90)/n365 if n365 else np.nan
        row['year1_to90_ratio']=n365/n90 if n90 else np.nan;row['year2_count']=n730-n365 if np.isfinite(n730) else np.nan;row['year2_to_year1']=row['year2_count']/n365 if n365 and np.isfinite(n730) else np.nan
        row['firstyear_share_of_current']=n365/row['through_august'] if row['through_august'] else np.nan
        row['postyear1_count']=row['through_august']-n365
    # Full calendar months are kept, including zero months.
    start=min(game.date,dates.min()) if n else game.date
    months=pd.date_range(start=start.normalize().replace(day=1),end=end,freq='MS',inclusive='left')
    counts=dates.dt.strftime('%Y-%m').value_counts();series=np.array([counts.get(m.strftime('%Y-%m'),0) for m in months],dtype=int)
    rec=series[months>=recent_start];row['last12_count']=int(rec.sum());row['last12_active_months']=int((rec>0).sum());row['last12_months_ge5']=int((rec>=5).sum());row['last12_peak_month']=int(rec.max()) if len(rec) else 0
    for m,k in zip(months,series):monthlyrows.append(dict(appid=game.appid,name=game.name,month=m.strftime('%Y-%m'),count=int(k),months_from_recorded_release=(m.year-game.date.year)*12+m.month-game.date.month))
    # Late bursts are descriptive screens, not diagnoses of updates or virality.
    for min_count,ratio in [(15,3),(20,4),(50,6)]:
        starts=[]
        for j in range(6,len(series)-2):
            if months[j]<game.date+pd.Timedelta(days=365) or not valid:continue
            before=int(series[j-6:j].sum());after=int(series[j:j+3].sum());expected=before/2
            if after>=min_count and after>=ratio*max(1,expected):starts.append((j,before,after))
        episodes=[]
        for j,before,after in starts:
            if not episodes or j-episodes[-1][0]>6:episodes.append((j,before,after))
        row[f'bursts_{min_count}_{ratio}']=len(episodes)
        for j,before,after in episodes:bursts.append(dict(appid=game.appid,name=game.name,threshold=min_count,ratio=ratio,start_month=months[j].strftime('%Y-%m'),prior6=before,next3=after,age_years=(months[j]-game.date).total_seconds()/86400/365.25))
    # A matched 2-year observation window for comparing burst prevalence by era.
    row['burst_in_first2years']=any(x['appid']==game.appid and x['threshold']==20 and 1<=x['age_years'] and x['age_years']+3/12<=2 for x in bursts) if valid and game.date+pd.Timedelta(days=730)<=end else np.nan
    rows.append(row)
d=pd.DataFrame(rows);d.to_csv(R/'game_histories_all_completed.csv',index=False)
pd.DataFrame(monthlyrows).to_csv(R/'monthly_arrivals.csv',index=False);pd.DataFrame(bursts).to_csv(R/'late_bursts.csv',index=False)
if len(d)!=60:
    print('Completed histories',len(d),'of60; population aggregates pending');raise SystemExit
d.to_parquet(R/'game_histories.parquet',index=False);d.to_csv(R/'game_histories.csv',index=False)
def ratio_estimate(y,dom):
    y=y.fillna(False).astype(float)*dom.astype(float);z=dom.astype(float);den=(d.weight*z).sum();p=(d.weight*y).sum()/den if den else np.nan;res=y-p*z;v=0
    for _,a in d.groupby('stratum'):
        N=a.population_n.iloc[0];n=len(a);v+=N*N*(1-n/N)*res.loc[a.index].var(ddof=1)/n
    se=math.sqrt(max(0,v))/den if den else np.nan;n=int(dom.sum());k=int(y.sum());boundary=k in [0,n]
    return dict(sample_n=n,k=k,estimate=p,lo95=np.nan if boundary else max(0,p-1.96*se),hi95=np.nan if boundary else min(1,p+1.96*se))
metrics={'any_last12':d.last12_count.gt(0),'atleast12_last12':d.last12_count.ge(12),'active_9of12':d.last12_active_months.ge(9),'active_12of12':d.last12_active_months.eq(12),'late_burst_20_4':d.bursts_20_4.gt(0),'burst_in_first2years':d.burst_in_first2years.eq(True)}
domains=[('all',pd.Series(True,index=d.index)),('50plus',d.snapshot_reviews.ge(50)),('1000plus',d.snapshot_reviews.ge(1000)),('50plus_and_year1_ge10',d.snapshot_reviews.ge(50)&d.release_d365.ge(10)),('50plus_no_known_ea',d.snapshot_reviews.ge(50)&~d.known_ea),('50plus_price_le5',d.snapshot_reviews.ge(50)&d.price.le(5)),('50plus_price_gt5',d.snapshot_reviews.ge(50)&d.price.gt(5))]+[(f'period:{p}',d.period.eq(p)) for p in sorted(d.period.unique())]
stats=[]
for label,dom in domains:
    for key,y in metrics.items():
        valid=dom&d.release_clock_valid if key.startswith('late_burst') else dom
        if key=='burst_in_first2years':valid=dom&d.burst_in_first2years.notna()
        stats.append(dict(domain=label,metric=key,**ratio_estimate(y,valid)))
pd.DataFrame(stats).to_csv(R/'weighted_activity.csv',index=False)
dist=[]
for label,dom in domains:
    for metric in ['release_d30','release_d90','release_d365','year1_after90_share','year1_to90_ratio','year2_count','year2_to_year1','firstyear_share_of_current','postyear1_count','last12_count','last12_active_months']:
        x=d[dom&d[metric].notna()].copy();order=np.argsort(x[metric].to_numpy());values=x[metric].to_numpy()[order];weights=x.weight.to_numpy()[order];cum=np.cumsum(weights)
        def quant(p):return float(values[np.searchsorted(cum,p*cum[-1])]) if len(values) else None
        dist.append(dict(domain=label,metric=metric,n=len(x),weighted_p25=quant(.25),weighted_median=quant(.5),weighted_p75=quant(.75),unweighted_median=x[metric].median()))
pd.DataFrame(dist).to_csv(R/'distributions.csv',index=False)
summary={'sample':len(d),'reviews':int(d.collected.sum()),'valid_release_clocks':int(d.release_clock_valid.sum()),'date_flags':d.loc[~d.release_clock_valid,['appid','name','date','first_review','first_review_offset_days']].astype(str).to_dict('records'),'exact_summary_matches':int(d.reconciliation_difference.eq(0).sum()),'max_abs_summary_difference':int(d.reconciliation_difference.abs().max()),'full_calendar_end':'2026-09-01','primary_population':int(round(d.weight.sum()))}
(R/'summary.json').write_text(json.dumps(summary,indent=2));print(json.dumps(summary,indent=2))