# Descriptive (hindsight) comparison at the actual day low: sharp-fall days that recovered >=200 same day vs those that did not (<100).
import pandas as pd,numpy as np
from scipy.stats import mannwhitneyu,fisher_exact
B='/workspace/research/flow-forensics-2026-10'; R=f'{B}/recov'
E=pd.read_csv(f'{R}/results/day_events.csv',parse_dates=['date']); E=E[(E.date>='2024-10-21')&E.sharp_fall]
OPT=pd.read_pickle(f'{R}/data/opt_panels.pkl'); exps=sorted({e for _,e in OPT})
P=pd.read_parquet(f'{B}/lows/data/minute_panel_v.parquet',columns=['close','low','fut','foi_front','date','hm']); P=P[P.date>='2024-10-21']
def rd(f):
    d=pd.read_csv(f); d['date']=pd.to_datetime(d.ts.str[:10]); d['hm']=d.ts.str[11:16]; return d
tn=rd(f'{R}/data/today/NIFTY.csv'); tf=rd(f'{R}/data/today/NIFTYFUT.csv').set_index('hm'); tn['fut']=tn.hm.map(tf.close); tn['foi_front']=tn.hm.map(tf.oi)
P=pd.concat([P,tn[P.columns]]); PG={d:g.set_index('hm') for d,g in P.groupby('date')}
def ldx(nm,fn):
    d=pd.read_parquet(f'{R}/data/{nm}_1m.parquet'); t=pd.read_csv(f'{R}/data/today/{fn}.csv'); d=pd.concat([d,t]); d['date']=pd.to_datetime(d.ts.str[:10]); d['hm']=d.ts.str[11:16]
    return {k:g.set_index('hm') for k,g in d.groupby('date')}
VX=ldx('VIX','INDIAVIX'); BN=ldx('BANKNIFTY','BANKNIFTY')
def sh(h,m): x=int(h[:2])*60+int(h[3:])+m; x=min(max(x,555),914); return f'{x//60:02d}:{x%60:02d}'
rows=[]
for r in E.itertuples():
    d=r.date; g=PG.get(d); tl=r.t_low; L=r.low
    if g is None: continue
    x=dict(date=d,group='recovered>=200' if r.rec>=200 else ('no_recovery<100' if r.rec<100 else 'partial100-200'),rec=r.rec,t_low=tl,
           low_min=(int(tl[:2])*60+int(tl[3:])-555),depth_pc=r.fall_pc,gap=r.gap,gap_down=r.gap<=-50,round_dist=abs(L-round(L/100)*100))
    try:
        b=lambda h:g.fut.get(h)-g.close.get(h); x['basis_chg_low_to+10']=b(sh(tl,10))-b(tl); x['basis_chg_-10_to_low']=b(tl)-b(sh(tl,-10))
        x['futoi_chg_low_to+15_pct']=(g.foi_front.get(sh(tl,15))/g.foi_front.get(tl)-1)*100
    except Exception: pass
    v=VX.get(d)
    if v is not None: x['vix_chg_low_to+15']=v.close.get(sh(tl,15),np.nan)-v.close.get(tl,np.nan); x['vix_lvl']=v.close.get(tl,np.nan)
    bn=BN.get(d)
    if bn is not None:
        w=bn.loc[:'15:14']; x['bn_low_minus_nifty_low_min']=(int(w.low.idxmin()[:2])*60+int(w.low.idxmin()[3:]))-(int(tl[:2])*60+int(tl[3:]))
    e=[e for e in exps if e>=d]; e=e[0] if e else None; x['dte']=(e-d).days if e is not None else np.nan
    O=OPT.get((d,e))
    if O is not None:
        oi=O['oi']; kb=np.floor(L/50)*50
        def s(tp,ks,h):
            v=[oi[(tp,k)].get(h,np.nan) for k in ks if (tp,k) in oi.columns]; v=[z for z in v if z==z]; return np.sum(v) if v else np.nan
        pk=[kb-50,kb,kb+50,kb+100]; ck=[kb+100,kb+150,kb+200,kb+250]
        x['put_oi_chg_-10_to_+15_pct']=(s('PE',pk,sh(tl,15))/s('PE',pk,sh(tl,-10))-1)*100
        x['call_oi_chg_low_to+30_pct']=(s('CE',ck,sh(tl,30))/s('CE',ck,tl)-1)*100
        x['call_oi_chg_-10_to_low_pct']=(s('CE',ck,tl)/s('CE',ck,sh(tl,-10))-1)*100
        pv={k:oi[('PE',k)].get(tl,np.nan) for (tp,k) in oi.columns if tp=='PE' and L-200<=k<=L+200}; pv={k:v for k,v in pv.items() if v==v}
        if pv: ks=max(pv,key=pv.get); x['heavy_put_dist']=L-ks
    rows.append(x)
X=pd.DataFrame(rows); X.to_csv(f'{R}/results/day_low_profiles.csv',index=False)
a=X[X.group=='recovered>=200']; b=X[X.group=='no_recovery<100']
num=[c for c in X.columns if c not in('date','group','t_low','gap_down','rec')]
out=[]
for c in num:
    u,v=a[c].dropna(),b[c].dropna()
    if len(u)>3 and len(v)>3: out.append(dict(metric=c,recovered_median=u.median(),norecovery_median=v.median(),n_rec=len(u),n_norec=len(v),p_mwu=mannwhitneyu(u,v).pvalue))
O_=pd.DataFrame(out)
def bh(p):
    p=np.asarray(p,float); n=len(p); o=np.argsort(p); q=np.empty(n); m=1
    for r,i in enumerate(o[::-1]): m=min(m,p[i]*n/(n-r)); q[i]=m
    return q
for flag,fn in [('gap_down',lambda s:s.gap_down==True),('dte0',lambda s:s.dte==0),('low_before_1000',lambda s:s.low_min<=45),('heavy_put_within40',lambda s:s.heavy_put_dist.abs()<=40),('bn_first',lambda s:s.bn_low_minus_nifty_low_min<0)]:
    t=[[fn(a).sum(),len(a)-fn(a).sum()],[fn(b).sum(),len(b)-fn(b).sum()]]
    O_=pd.concat([O_,pd.DataFrame([dict(metric='share_'+flag,recovered_median=fn(a).mean()*100,norecovery_median=fn(b).mean()*100,n_rec=len(a),n_norec=len(b),p_mwu=fisher_exact(t)[1])])])
O_['q_bh']=bh(O_.p_mwu.values)
O_.round(4).to_csv(f'{R}/results/day_low_compare.csv',index=False)
print(X.group.value_counts()); print(O_.round(3).to_string())