"""Stricter common-support comparisons and independent-credit diagnostics."""
import json
import numpy as np
from design import *
g=pd.read_parquet(R/'classified.parquet')
results=pd.read_csv(R/'host_results.csv');sens=pd.read_csv(R/'sensitivity.csv')
base=pd.read_parquet(ROOT/'analysis-2026-09-05/games.parquet',columns=['appid','first_date','valid_released','reviews'])
for appid,a in json.loads((ROOT/'portfolios-2026-09-06/date_overrides.json').read_text()).items():base.loc[base.appid.eq(int(appid)),'first_date']=pd.Timestamp(a['date'],tz='UTC')
c=pd.read_parquet(ROOT/'2026-09-05/exports/creators.parquet');c=c[c.role.eq('developers')].copy();c['developer_id']='name:'+c.name.map(norm)
history=c[['appid','developer_id']].drop_duplicates().merge(base[base.valid_released],on='appid')
first=history.groupby('developer_id').first_date.min()
strong=history[history.reviews.ge(1000)].groupby('developer_id').first_date.min()
g['has_prior']=g.developer_id.map(first).lt(g.first_date-pd.Timedelta(days=30))
g['has_prior_current1000']=g.developer_id.map(strong).lt(g.first_date-pd.Timedelta(days=30))
g['period_day']=g.first_date.map(lambda d:d.timestamp()/86400)

matches=[];rows=[]
for period in ['2019_2022','2023_2025','2026_JanAug']:
 base=g[g.period.eq(period)];max_days=30 if period=='2026_JanAug' else 90
 for host in HOSTS:
  root=base[base['host10_'+host]&base['text_'+host]]
  for variant,pool in [('all',root),('no_core',root[~root.core]),('no_rogue',root[~root.rogue]),('no_prior_current1000',root[~root.has_prior_current1000]),('no_prior_credit',root[~root.has_prior])]:
   b=pool[pool.strict];controls=pool[~pool.strict];supported=[]
   for a in b.itertuples():
    peers=controls[controls.price_bucket.fillna('missing').eq(a.price_bucket if pd.notna(a.price_bucket) else 'missing')&controls.developer_id.ne(a.developer_id)].copy()
    peers['distance']=(peers.period_day-a.period_day).abs()
    peers=peers[peers.distance.le(max_days)].sort_values(['distance','appid']).head(5)
    if len(peers)<3:continue
    record=dict(period=period,host=host,variant=variant,appid=a.appid,game=a.name,developer_id=a.developer_id,reviews=a.reviews,
      peer_n=len(peers),peer_ids='|'.join(peers.appid.astype(str)),max_gap_days=float(peers.distance.max()),
      **{f'expected{k}':float(peers.reviews.ge(k).mean()) for k in [50,100,200,1000]})
    supported.append(record);matches.append(record)
   n=len(supported);s=pd.DataFrame(supported)
   rr=dict(period=period,host=host,variant=variant,pool_n=len(pool),build_n=len(b),supported_n=n,support=n/len(b) if len(b) else None,
           build_ge100=int(b.reviews.ge(100).sum()),build_developers=int(b.developer_id.nunique()))
   for k in [50,100,200,1000]:
    count=int(s.reviews.ge(k).sum()) if n else 0;expected=float(s[f'expected{k}'].sum()) if n else 0
    rr[f'observed{k}']=count;rr[f'expected{k}']=expected;rr[f'ratio{k}']=count/expected if expected else None
   rows.append(rr)
pd.DataFrame(rows).to_csv(R/'matched_results.csv',index=False)
pd.DataFrame(matches).to_csv(R/'matched_games.csv',index=False)

# Candidate table keeps the original screen and exposes every further diagnostic.
screen=pd.read_csv(R/'sparse_screen.csv');main=pd.DataFrame(rows).query("period == '2023_2025' and variant == 'all'")
screen=screen.merge(main[['host','supported_n','support','observed100','expected100','ratio100']].rename(columns={'expected100':'matched_expected100','ratio100':'matched_ratio100'}),on='host',validate='one_to_one')
for variant in ['no_core','no_rogue','no_prior_current1000','no_prior_credit']:
 s=pd.DataFrame(rows).query("period == '2023_2025' and variant == @variant")
 screen=screen.merge(s[['host','build_n','build_ge100','build_developers','supported_n','ratio100']].rename(columns={k:variant+'_'+k for k in ['build_n','build_ge100','build_developers','supported_n','ratio100']}),on='host',validate='one_to_one')
ext=sens.query("period == '2023_2025' and variant == 'extended'")[['host','b_n','share','b_ge100','ratio100']]
screen=screen.merge(ext.rename(columns={k:'extended_'+k for k in ['b_n','share','b_ge100','ratio100']}),on='host',validate='one_to_one')
screen.to_csv(R/'candidate_diagnostics.csv',index=False)

# Every count-qualified case in each passing group is retained for checking that
# the feature and the host are both actually advertised, not inferred from reviews.
m=pd.read_parquet(R/'build_members.parquet')
passing=set(screen.loc[screen.screen_pass,'host'])
short=m[m.period.eq('2023_2025')&m.host.isin(passing)&m.reviews.ge(100)].copy()
short.to_csv(R/'candidate_successes.csv',index=False)
g[['appid','developer_id','has_prior','has_prior_current1000']].to_parquet(R/'creator_history_flags.parquet',index=False)
print('MATCHED MAIN\n',main[['host','build_n','supported_n','observed100','expected100','ratio100']].round(3).to_string(index=False))
print('CANDIDATES\n',screen[screen.screen_pass][['host','b_n','b_ge100','supported_n','support','matched_ratio100','no_core_build_n','no_core_build_ge100','no_rogue_build_n','no_rogue_build_ge100','no_prior_current1000_build_n','no_prior_current1000_build_ge100']].round(3).to_string(index=False))