"""Three offline continuations of the September 5 snapshot. No collection."""
import hashlib
from common import *

d,t=load()
recent=d[d.paid&d.year.isin([2025,2026])&d.month.le(8)].copy()
F=['deck','action','auto','tower']
recent['signature']=recent[F].apply(lambda r:'+'.join(k for k in F if r[k]) or 'none_of_four',axis=1)
growth=[]
for year in [2025,2026]:
    yearpool=recent[recent.year.eq(year)]
    for name,mask in [('all',pd.Series(True,index=yearpool.index))]+[(f,yearpool[f]) for f in F]+[('none_of_four',yearpool.signature.eq('none_of_four'))]:
        x=yearpool[mask]; b=x[x.strict]
        growth.append(dict(year=year,family=name,eligible=len(x),strict=len(b),inclusive=int(x.inclusive.sum()),
            build_share=float(x.strict.mean()),strict_median_words=float(b.text_words.median()),
            all_median_words=float(x.text_words.median()),english_support_share=float(b.english_support.mean()),
            unique_developers=int(b.developer_key.nunique()),median_price=float(b.usd_list_price.median())))
pd.DataFrame(growth).to_csv(R/'growth_families.csv',index=False)

# Exact disjoint signatures provide denominators for an additive decomposition.
sig=recent.groupby(['signature','year']).agg(eligible=('appid','size'),builds=('strict','sum')).unstack(fill_value=0)
parts=[]
for signature,row in sig.iterrows():
    n0,n1=[float(row['eligible',y]) for y in [2025,2026]]
    b0,b1=[float(row['builds',y]) for y in [2025,2026]]
    # A zero-sized signature is assigned the other period's rate; its entire
    # change then belongs to appearance/disappearance of the catalog group.
    p0=b0/n0 if n0 else (b1/n1 if n1 else 0)
    p1=b1/n1 if n1 else p0
    parts.append(dict(signature=signature,n2025=n0,n2026=n1,b2025=b0,b2026=b1,
        share2025=p0,share2026=p1,delta=b1-b0,
        catalog_volume_composition=(n1-n0)*(p0+p1)/2,
        within_signature_build_share=(p1-p0)*(n0+n1)/2))
parts=pd.DataFrame(parts)
parts.to_csv(R/'growth_decomposition.csv',index=False)
assert np.allclose(parts.delta,parts.catalog_volume_composition+parts.within_signature_build_share)

# Allocate each multi-tag game equally across its memberships, so totals add.
frac=[]
for year in [2025,2026]:
    x=recent[recent.year.eq(year)&recent.strict]
    k=x[F].sum(axis=1)
    for f in F:
        frac.append(dict(year=year,family=f,fractional_games=float((x[f]/k.where(k.gt(0))).sum())))
    frac.append(dict(year=year,family='none_of_four',fractional_games=int(k.eq(0).sum())))
pd.DataFrame(frac).to_csv(R/'growth_fractional_genres.csv',index=False)

objects=['card_deck','party_composition','equipment_skills','machine_assembly']
objectrows=[];signals=[]
for year in [2025,2026]:
    x=recent[recent.year.eq(year)&recent.strict]
    k=x[objects].sum(axis=1)
    for f in objects:
        objectrows.append(dict(year=year,object=f,n=int(x[f].sum()),share=float(x[f].mean()),fractional_games=float((x[f]/k.where(k.gt(0))).sum())))
    objectrows.append(dict(year=year,object='unassigned',n=int(k.eq(0).sum()),share=float(k.eq(0).mean()),fractional_games=int(k.eq(0).sum())))
    for f in ['signal_explicit_build','signal_combination','signal_deck','signal_playstyle','signal_quantity','signal_prerun','signal_position']:
        signals.append(dict(year=year,signal=f,n=int(x[f].sum()),share=float(x[f].mean())))
pd.DataFrame(objectrows).to_csv(R/'growth_objects.csv',index=False)
pd.DataFrame(signals).to_csv(R/'growth_wording.csv',index=False)

# List every tag's growth within the eligible catalog, retaining small cells.
tagrows=[]
for tag,ids in t[t['rank'].le(20)].groupby('tag').appid:
    for year in [2025,2026]:
        x=recent[recent.year.eq(year)&recent.index.isin(ids)]
        tagrows.append(dict(tag=tag,year=year,eligible=len(x),builds=int(x.strict.sum())))
pd.DataFrame(tagrows).to_csv(R/'growth_all_tags.csv',index=False)

HYBRIDS={
    'party_tactics':['Party-Based RPG','Tactical RPG','Strategy RPG','Turn-Based Tactics','Grid-Based Movement','Hex Grid'],
    'economy_building':['Resource Management','Automation','Base Building','City Builder','Colony Sim'],
    'inventory':['Inventory Management'],
    'auto_battle':['Auto Battler'],
    'tower_defense':['Tower Defense'],
    'action_roguelike':['Action Roguelike'],
}
for rank in [10,20]:
    for label,tags in HYBRIDS.items():
        d[f'h_{label}_{rank}']=d.index.isin(t.loc[t.tag.isin(tags)&t['rank'].le(rank),'appid'])
    hs=[f'h_{f}_{rank}' for f in HYBRIDS]
    d[f'h_any_{rank}']=d[hs].any(axis=1)

rows=[];thresholds=[];conditional=[];members=[]
for period in ['2025','2026_H1','2026_JulAug']:
    base=d[d.build&d.period.eq(period)]
    y100=base.reviews.ge(100);yliked=y100&base.positive_pct.ge(80)
    ex100=expected(base,y100);exliked=expected(base,yliked)
    extext=expected(base,yliked,('date_bin','price_bucket','text_bucket'))
    groups={'all_builds':base,'deck_strict':base[base.deck],
            'other_builds':base[~base.deck],
            'deck_all_eligible':d[d.paid&d.period.eq(period)&d.deck],
            'deck_missed_by_strict':d[d.paid&d.period.eq(period)&d.deck&~d.strict]}
    for f in ['action','auto','tower']:
        groups[f+'_strict']=base[base[f]]
    for rank in [10,20]:
        for label in HYBRIDS:
            groups[f'deck_{label}_tag{rank}']=base[base.deck&base[f'h_{label}_{rank}']]
        groups[f'deck_any_hybrid_tag{rank}']=base[base.deck&base[f'h_any_{rank}']]
        groups[f'deck_no_selected_hybrid_tag{rank}']=base[base.deck&~base[f'h_any_{rank}']]
    for label,x in groups.items():
        row=dict(period=period,group=label,**stats(x))
        if x.index.isin(base.index).all():
            for col,e,y in [('volume',ex100,y100),('joint',exliked,yliked),('joint_text',extext,yliked)]:
                denom=e.reindex(x.index).sum()
                row[col+'_expected']=float(denom)
                row[col+'_ratio']=float(y.reindex(x.index).sum()/denom) if denom else None
        rows.append(row)
        if label in ['deck_strict','other_builds','deck_all_eligible','deck_missed_by_strict']:
            for nr in [50,100,200]:
                for positivity in [75,80,85]:
                    thresholds.append(dict(period=period,group=label,review_threshold=nr,positivity=positivity,n=len(x),qualifying=int((x.reviews.ge(nr)&x.positive_pct.ge(positivity)).sum())))
        if label in ['deck_strict','deck_missed_by_strict']:
            z=x[['appid','name','first_date','reviews','positive_pct','usd_list_price','developer','publisher','short_description','description','store_url']].copy()
            z['period']=period;z['group']=label
            for h in HYBRIDS: z[h]=x[f'h_{h}_20']
            members.append(z)
    # Conditional positivity comparisons use only reviewed games as the base.
    reviewed=base[base.reviews.ge(100)]
    exsat=expected(reviewed,reviewed.positive_pct.ge(80))
    for label,mask in [('deck',reviewed.deck),('other_builds',~reviewed.deck)]:
        x=reviewed[mask]; e=exsat.reindex(x.index).sum()
        conditional.append(dict(period=period,group=label,n=len(x),positive80=int(x.positive_pct.ge(80).sum()),expected=float(e),ratio=float(x.positive_pct.ge(80).sum()/e) if e else None))
pd.DataFrame(rows).to_csv(R/'reception_groups.csv',index=False)
pd.DataFrame(thresholds).to_csv(R/'reception_thresholds.csv',index=False)
pd.DataFrame(conditional).to_csv(R/'conditional_positivity.csv',index=False)
pd.concat(members).to_parquet(R/'deck_members.parquet',index=False)

# Combined periods for small hybrid groups; standardization retains period/date.
pool=d[d.build&d.period.ne('other')]
deck=pool[pool.deck]
e=expected(deck,deck.reviews.ge(100)&deck.positive_pct.ge(80))
hybridrows=[]
for rank in [10,20]:
    for label in list(HYBRIDS)+['any']:
        mask=deck[f'h_{label}_{rank}']
        for present in [True,False]:
            x=deck[mask.eq(present)]
            den=e.reindex(x.index).sum()
            hybridrows.append(dict(rank=rank,hybrid=label,present=present,**stats(x),
                expected_joint=float(den),within_deck_joint_ratio=float((x.reviews.ge(100)&x.positive_pct.ge(80)).sum()/den) if den else None))
pd.DataFrame(hybridrows).to_csv(R/'hybrid_combined.csv',index=False)

# Creator clustering check on the focal deck finding; remove most prolific as
# well as most reviewed creator. Report outcomes and expected counts separately.
sens=[]
for period in ['2025','2026_H1','2026_JulAug']:
    b=d[d.build&d.period.eq(period)];x=b[b.deck]
    ex=expected(b,b.reviews.ge(100)&b.positive_pct.ge(80))
    for field in ['developer_key','publisher_key']:
        for rule in ['most_reviews','most_games']:
            counts=x.groupby(field).reviews.agg(['sum','size'])
            key=counts['sum' if rule=='most_reviews' else 'size'].idxmax()
            z=x[x[field].ne(key)];den=ex.reindex(z.index).sum()
            sens.append(dict(period=period,field=field,rule=rule,removed_key=key,removed_name=x.loc[x[field].eq(key),field.replace('_key','')].iloc[0],removed_n=len(x)-len(z),**stats(z),joint_ratio=float((z.reviews.ge(100)&z.positive_pct.ge(80)).sum()/den) if den else None))
pd.DataFrame(sens).to_csv(R/'creator_sensitivity.csv',index=False)

# Outcomes are joined only after manual labels have been frozen.
frozen=json.loads((R/'pilot_labels_frozen.json').read_text())
assert hashlib.sha256((R/'pilot_labels.json').read_bytes()).hexdigest()==frozen['sha256']
labels=pd.DataFrame(json.loads((R/'pilot_labels.json').read_text()))
pilot=labels.merge(d[['appid','reviews','positive_pct','usd_list_price','text_words','quantity_max','store_url','developer','publisher']].reset_index(drop=True),on='appid',validate='one_to_one')
pilot.to_csv(R/'pilot_with_outcomes.csv',index=False)
pilots=[]
for axis in ['label','explicit_interaction','family','period']:
    for value,x in pilot.groupby(axis):
        pilots.append(dict(axis=axis,value=value,**stats(x.assign(developer_key=x.developer,publisher_key=x.publisher)),median_words=float(x.text_words.median()),claims100=int(x.quantity_max.ge(100).sum())))
pd.DataFrame(pilots).to_csv(R/'pilot_results.csv',index=False)

save_json('run_summary.json',dict(growth=parts[['delta','catalog_volume_composition','within_signature_build_share']].sum().to_dict(),
    unique_game_count=len(d),pilot_labels_sha256=frozen['sha256'],network_paused=json.loads((ROOT/'network_policy.json').read_text())['network_paused']))
print('GROWTH\n',pd.DataFrame(growth).round(3).to_string(index=False))
print('DECOMPOSITION\n',parts.round(2).to_string(index=False))
print('MAIN RECEPTION\n',pd.DataFrame(rows).query("group in ['all_builds','deck_strict','other_builds','deck_all_eligible','deck_missed_by_strict']")[['period','group','n','under10','ge100','liked100','ge1000','median_reviews','satisfied_among100','volume_ratio','joint_ratio','joint_text_ratio']].round(3).to_string(index=False))
print('HYBRIDS\n',pd.DataFrame(hybridrows).query('rank == 20')[['hybrid','present','n','ge100','liked100','liked100_rate','satisfied_among100','within_deck_joint_ratio']].round(3).to_string(index=False))
print('PILOT\n',pd.DataFrame(pilots).round(3).to_string(index=False))