"""Offline portfolio study. All outcomes are September 5 snapshot totals."""
from pathlib import Path
import json,re,unicodedata,hashlib
import numpy as np
import pandas as pd
from source_data import load_games
R=Path(__file__).resolve().parent;ROOT=R.parent
def norm(s):return re.sub(r'\s+',' ',unicodedata.normalize('NFKC',str(s)).casefold()).strip()
def save(name,x):
    if isinstance(x,pd.DataFrame):x.to_csv(R/name,index=False)
    else:(R/name).write_text(json.dumps(x,indent=2,default=str,ensure_ascii=False))

SPEC=dict(snapshot='2026-09-05',network='offline',
 identity='Full developer credits; normalized credit NAME is the primary identity. Steam creator-page IDs can be shared by many different developers, so never merge names using those IDs. Main sample has exactly one distinct nonempty normalized developer name; names pointing to multiple creator pages are conservatively excluded. This splits unverified name changes and may conflate unrecognized homonyms.',
 population='Currently paid, non-explicit, valid dated released games. No genre exclusions. Titles explicitly identifying demos/prologues/playtests/previews excluded in main, retained in accounting. This is an observed paid catalog, not all work or a verified indie/solo census.',
 portfolios='At least 2 and at most 10 included games through 2025, first observed included release 2010-2023, latest 2020-2025. Main shape tables one developer one row. Single-game identities also counted separately. Strong catalogs defined explicitly by current peak >=1000 reviews; descriptive selection only.',
 transitions='Consecutive eligible paid games per developer; 30-3650 day gaps. Latest qualifying transition per developer per period. Predecessor strength is its review count NOW, not at follow-up launch. Distinct release dates required; simultaneous launches excluded.',
 periods=['2019_2022','2023_2025','2026_JanAug'],
 outcomes='10/100/1000 filtered reviews, >=100 and >=80% positive, median reviews, release-quarter percentile. No sales or value judgment.',
 controls='Descriptive expected counts using current release quarter and current price, then predecessor current-review band for gap/similarity comparisons. Cells >=30, fallback by successively dropping price then quarter; report support. Reference is full transition pool. No causal interpretation.',
 similarity='Jaccard of top-10 tag names after excluding a fixed administrative set; <=.25 distant, >.25 to <.5 middle, >=.5 close. Store positioning, not verified mechanical similarity or artistic novelty.',
 gap='30-180d,181-365d,1-2y,2-4y,4-10y. Release intervals are not development times.',
 sensitivity='Unshared creator-page subset, prior/current recorded EA removed, one game per developer weighting, 2-5 vs 6-10 title catalogs, date-cohort review percentiles, shared-franchise removal, price adjustment. Repeat credits may still be studio groups or changes of staff.',
 limitations='Missing delisted/renamed/other-platform games; migration/EA dates; survivor selection for developers who returned; current tags/prices/text; regression to mean comparing with largest game; no causal audience carryover, follower counts, player overlap, or release-time reputation.')
save('specification.json',SPEC)

G=load_games()
C=pd.read_parquet(ROOT/'2026-09-05/exports/creators.parquet')
T=pd.read_parquet(ROOT/'analysis-2026-09-05/tags.parquet')
c=C[C.role.eq('developers')].copy();c['norm']=c.name.map(norm);c=c[c.norm.ne('')]
c['raw_key']=['clan:'+str(int(cid)) if pd.notna(cid) and cid>0 else 'name:'+n for cid,n in zip(c.clan_id,c.norm)]
nm=c[c.raw_key.str.startswith('clan:')].groupby('norm').raw_key.agg(lambda x:set(x))
mapping={n:next(iter(s)) for n,s in nm.items() if len(s)==1}
ambiguous={n for n,s in nm.items() if len(s)>1}
c['key']=[mapping.get(n,k) if k.startswith('name:') else k for n,k in zip(c.norm,c.raw_key)]
c['page_names']=c.groupby('key').norm.transform('nunique')
counts=c.groupby('appid').norm.nunique();sole=set(counts[counts.eq(1)].index)
c=c.sort_values('ordinal').drop_duplicates(['appid','norm'])
first=c[c.appid.isin(sole)].drop_duplicates('appid')[['appid','name','norm','key','page_names']]
first=first.rename(columns={'name':'credit','norm':'credit_norm','key':'creator_page_key'})
first['dev']='name:'+first.credit_norm
d=G.merge(first,on='appid',how='left',validate='one_to_one')
d['ambiguous_name']=d.credit_norm.isin(ambiguous)
# A bounded explicit-title exclusion; ordinary free demos are already outside paid.
promo=re.compile(r'\b(?:prologue|playtest|demo|preview)\b',re.I)
d['promotional_title']=d.name.fillna('').str.contains(promo)
d['included']=d.main&d.dev.notna()&~d.ambiguous_name&~d.promotional_title
x=d[d.included].copy().sort_values(['dev','first_date','appid'])
save('accounting.json',dict(source_games=len(d),valid_released=int(d.valid_released.sum()),paid_nonexplicit=int(d.main.sum()),
 paid_multicredit_or_missing=int((d.main&d.dev.isna()).sum()),paid_ambiguous_name=int((d.main&d.dev.notna()&d.ambiguous_name).sum()),
 paid_promotional_excluded=int((d.main&d.dev.notna()&~d.ambiguous_name&d.promotional_title).sum()),included=len(x),developers=x.dev.nunique(),ambiguous_normalized_names=len(ambiguous)))
save('ambiguous_names.csv',pd.DataFrame([dict(name=n,clans='|'.join(sorted(nm[n]))) for n in sorted(ambiguous)]))

# Date-only reference ranks include all paid/non-explicit records, regardless credits.
ref=G[G.main].copy();ref['cohort_pct']=ref.groupby('quarter').reviews.rank(pct=True,method='average')
x=x.merge(ref[['appid','cohort_pct']],on='appid',validate='one_to_one').sort_values(['dev','first_date','appid'])
excluded_tags={'Indie','Singleplayer','Early Access','Free to Play','Great Soundtrack','Controller','Steam Machine','Software','Utilities'}
ts=T[T['rank'].le(10)&~T.tag.isin(excluded_tags)].groupby('appid').tag.agg(set).to_dict()
fr=C[C.role.eq('franchises')].groupby('appid').name.agg(lambda z:{norm(v) for v in z if str(v).strip()}).to_dict()
x['ordinal']=x.groupby('dev').cumcount()+1
for col in ['appid','name','first_date','reviews','positive_pct','usd_list_price','cohort_pct','known_ea_history','publisher_key','publisher','credit_norm','page_names']:
    x['prev_'+col]=x.groupby('dev')[col].shift()
x['gap_days']=(x.first_date-x.prev_first_date).dt.total_seconds()/86400
x['previous_max_now']=x.groupby('dev').reviews.transform(lambda z:z.cummax().shift())
x['similarity']=[len(ts.get(a,set())&ts.get(b,set()))/len(ts.get(a,set())|ts.get(b,set())) if ts.get(a,set())|ts.get(b,set()) else np.nan for a,b in zip(x.appid,x.prev_appid)]
x['shared_franchise']=[bool(fr.get(a,set())&fr.get(b,set())) for a,b in zip(x.appid,x.prev_appid)]
x['same_publisher']=x.publisher.map(norm).eq(x.prev_publisher.map(norm))
x['prior_band']=pd.cut(x.prev_reviews,[-1,9,99,999,9999,np.inf],labels=['<10','10-99','100-999','1000-9999','10000+']).astype(str)
x['gap_band']=pd.cut(x.gap_days,[29.999,180,365,730,1461,3650],labels=['30-180d','181-365d','1-2y','2-4y','4-10y']).astype(str)
x['similarity_band']=pd.cut(x.similarity,[-.001,.25,.4999999,1],labels=['distant','middle','close']).astype(str)
x['period']='other'
for name,lo,hi in [('2019_2022',2019,2022),('2023_2025',2023,2025)]:x.loc[x.year.between(lo,hi),'period']=name
x.loc[x.year.eq(2026)&x.month.le(8),'period']='2026_JanAug'
x.to_parquet(R/'games.parquet',index=False)

def stats(z):
    n=len(z);rev=z.reviews
    return dict(n=n,developers=z.dev.nunique(),under10=int(rev.lt(10).sum()),ge100=int(rev.ge(100).sum()),ge1000=int(rev.ge(1000).sum()),
        liked100=int((rev.ge(100)&z.positive_pct.ge(80)).sum()),median_reviews=float(rev.median()) if n else None,
        rate100=float(rev.ge(100).mean()) if n else None,rate1000=float(rev.ge(1000).mean()) if n else None,
        liked100_rate=float((rev.ge(100)&z.positive_pct.ge(80)).mean()) if n else None,
        median_cohort_pct=float(z.cohort_pct.median()) if n else None,
        median_gap=float(z.gap_days.median()) if n and 'gap_days' in z else None,
        median_price=float(z.usd_list_price.median()) if z.usd_list_price.notna().any() else None)

z=x[x.year.le(2025)].copy();z['ge100']=z.reviews.ge(100);z['ge1000']=z.reviews.ge(1000)
p=z.groupby('dev').agg(credit=('credit','last'),n=('appid','size'),first_year=('year','first'),last_year=('year','last'),
 total_reviews=('reviews','sum'),median_reviews=('reviews','median'),first_reviews=('reviews','first'),last_reviews=('reviews','last'),
 count100=('ge100','sum'),count1000=('ge1000','sum'),first_cohort_pct=('cohort_pct','first'),last_cohort_pct=('cohort_pct','last'))
rank=z.sort_values('reviews',ascending=False,kind='stable').copy();rank['review_rank']=rank.groupby('dev').cumcount()
peak=rank[rank.review_rank.eq(0)].set_index('dev')
p['peak_reviews']=peak.reviews;p['peak_name']=peak.name;p['peak_appid']=peak.appid;p['peak_ordinal']=peak.ordinal
p['peak_share']=p.peak_reviews/p.total_reviews.where(p.total_reviews.gt(0))
p['second_reviews']=rank[rank.review_rank.eq(1)].set_index('dev').reviews.reindex(p.index).fillna(0).astype(int)
p['nonpeak_ge100']=p.count100-p.peak_reviews.ge(100).astype(int);p['nonpeak_ge1000']=p.count1000-p.peak_reviews.ge(1000).astype(int)
p['peak_cohort_ordinal']=z.loc[z.groupby('dev').cohort_pct.idxmax()].set_index('dev').ordinal
p['peak_first']=p.peak_ordinal.eq(1)
p=p.reset_index();p['eligible_shape']=p.n.between(2,10)&p.first_year.between(2010,2023)&p.last_year.between(2020,2025)
save('portfolios.csv',p)
shape=[]
for label,mask in [('all_eligible',p.eligible_shape),('peak1000',p.eligible_shape&p.peak_reviews.ge(1000)),('peak10000',p.eligible_shape&p.peak_reviews.ge(10000))]:
    z=p[mask]
    for size,sm in [('2-10',z.n.between(2,10)),('2-5',z.n.between(2,5)),('6-10',z.n.between(6,10))]:
        a=z[sm];shape.append(dict(group=label,size=size,n=len(a),median_peak_share=float(a.peak_share.median()),peak_half=int(a.peak_share.ge(.5).sum()),peak80=int(a.peak_share.ge(.8).sum()),
           other100=int(a.nonpeak_ge100.ge(1).sum()),other1000=int(a.nonpeak_ge1000.ge(1).sum()),median_second=float(a.second_reviews.median()),first_is_peak=int(a.peak_first.sum()),
           first_is_cohort_peak=int(a.peak_cohort_ordinal.eq(1).sum()),median_games=float(a.n.median())))
save('portfolio_shapes.csv',pd.DataFrame(shape))

tr=x[x.gap_days.between(30,3650)&x.period.ne('other')].copy()
all_transitions=tr.copy()
# At most one vote per developer in each period.
tr=tr.sort_values('first_date').groupby(['dev','period'],sort=False).tail(1).copy()
def expectations(pool,y,control_prior=False):
    keys=['prior_band'] if control_prior else []
    base=y.groupby(pool.prior_band).transform('mean') if control_prior else pd.Series(y.mean(),index=pool.index)
    support=pd.Series(False,index=pool.index)
    for cols in [keys+['quarter'],keys+['quarter','price_bucket']]:
        k=pool[cols].fillna('missing').astype(str).agg('|'.join,axis=1)
        enough=y.groupby(k).transform('size').ge(30)
        base=y.groupby(k).transform('mean').where(enough,base)
        support=enough
    # Recalibrate global or prior-band totals after sparse fallback.
    group=pool.prior_band if control_prior else pd.Series('all',index=pool.index)
    target=y.groupby(group).transform('sum')
    for _ in range(40):
        total=base.groupby(group).transform('sum');base=(base*target.div(total.where(total.gt(0))).fillna(0)).clip(upper=1)
        if (base.groupby(group).transform('sum')-target).abs().max()<1e-8:break
    assert abs(base.sum()-y.sum())<1e-5
    return base,support

rows=[];sensitivity=[]
for period,pool in tr.groupby('period'):
    y=pool.reviews.ge(100).astype(float);yl=(pool.reviews.ge(100)&pool.positive_pct.ge(80)).astype(float)
    e,s=expectations(pool,y);el,sl=expectations(pool,yl)
    ep,sp=expectations(pool,y,True)
    for axis in ['prior_band','gap_band','similarity_band']:
        for value,z in pool.groupby(axis):
            pred=ep if axis!='prior_band' else e
            den=pred.reindex(z.index).sum();denl=el.reindex(z.index).sum()
            rows.append(dict(period=period,axis=axis,value=value,**stats(z),expected100=float(den),ratio100=float(z.reviews.ge(100).sum()/den) if den else None,
                expected_liked100_dateprice=float(denl),liked_ratio_dateprice=float((z.reviews.ge(100)&z.positive_pct.ge(80)).sum()/denl) if denl else None,
                full_cell_support=float((sp if axis!='prior_band' else s).reindex(z.index).mean()),median_prior_reviews=float(z.prev_reviews.median())))
    for name,mask in [('all',pd.Series(True,index=pool.index)),('no_recorded_ea',~pool.known_ea_history&~pool.prev_known_ea_history.astype(bool)),
                      ('no_shared_franchise',~pool.shared_franchise),('same_publisher',pool.same_publisher),('different_publisher',~pool.same_publisher),
                      ('unshared_creator_pages',pool.page_names.eq(1)&pool.prev_page_names.eq(1))]:
        for band,z in pool[mask].groupby('prior_band'):
            sensitivity.append(dict(period=period,restriction=name,prior_band=band,**stats(z)))
save('transition_groups.csv',pd.DataFrame(rows));save('transition_sensitivity.csv',pd.DataFrame(sensitivity))
tr.to_parquet(R/'transitions.parquet',index=False)
all_transitions.to_parquet(R/'all_transitions.parquet',index=False)

# First observed paid releases in the same periods, for a clearly labeled reference.
reference=[]
for period in ['2019_2022','2023_2025','2026_JanAug']:
    for name,z in [('first_observed_paid',x[x.period.eq(period)&x.ordinal.eq(1)]),('returning_developer',tr[tr.period.eq(period)])]:
        reference.append(dict(period=period,group=name,**stats(z)))
save('first_vs_returning.csv',pd.DataFrame(reference))
print('ACCOUNTING', (R/'accounting.json').read_text())
print('SHAPES\n',pd.DataFrame(shape).round(3).to_string(index=False))
print('TRANSITIONS\n',pd.DataFrame(rows).query("period == '2023_2025'").round(3).to_string(index=False))
print('REFERENCE\n',pd.DataFrame(reference).round(3).to_string(index=False))