# Live-causal candidates: after a sharp fall, first bounce of >=30 pts off the running low within 15 min.
# Features use only data up to the trigger minute. Outcome: +200 from that low before breaking it (by 15:14).
import pandas as pd,numpy as np,glob,os,pyarrow.parquet as pq
B='/workspace/research/flow-forensics-2026-10'; R=f'{B}/recov'
OD='/workspace/nifty-options-s3/data/options/1m/by_expiry/'
P=pd.read_parquet(f'{B}/lows/data/minute_panel_v.parquet',columns=['timestamp','open','high','low','close','fut','foi_front','date','hm'])
P=P[P.date>='2024-10-21']
T=f'{R}/data/today/'
def rd(f):
    d=pd.read_csv(f); d['timestamp']=pd.to_datetime(d.ts.str[:19]); d['date']=d.timestamp.dt.normalize(); d['hm']=d.timestamp.dt.strftime('%H:%M'); return d
tn=rd(T+'NIFTY.csv'); tf=rd(T+'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]],ignore_index=True)
def ld(nm):
    d=pd.read_parquet(f'{R}/data/{nm}_1m.parquet'); d['timestamp']=pd.to_datetime(d.ts.str[:19]); t=rd(T+f'{ {"VIX":"INDIAVIX","BANKNIFTY":"BANKNIFTY"}[nm]}.csv')
    d=pd.concat([d,t]); d['date']=d.timestamp.dt.normalize(); d['hm']=d.timestamp.dt.strftime('%H:%M'); return d.set_index(['date','hm'])
VIX=ld('VIX'); BN=ld('BANKNIFTY')
D=pd.read_parquet('/workspace/research/common/data/nifty_daily_hist.parquet')[['date','low','close']]
u=pd.read_csv(f'{B}/data/1m/NIFTY_day.csv'); u['date']=pd.to_datetime(u.ts.str[:10])
D=pd.concat([D,u[['date','low','close']],pd.DataFrame({'date':[pd.Timestamp('2026-10-26')]})]).drop_duplicates('date',keep='last').sort_values('date').reset_index(drop=True)
D['pc']=D.close.shift(1); D['pdl']=D.low.shift(1); DI=D.set_index('date')
# ---- options
EXPS=sorted(pd.Timestamp(os.path.basename(f)[:10]) for f in glob.glob(OD+'*.parquet'))
OCT=[pd.Timestamp(x) for x in ['2026-10-06','2026-10-13','2026-10-19','2026-10-27','2026-11-03']]
ALLE=sorted(set(EXPS)|set(OCT))
def nexp(d,skip0=False):
    c=[e for e in ALLE if e>=d and not(skip0 and e==d)]; return c[0] if c else None
days=sorted(P.date.unique())
OPT={}   # (date,expiry) -> DataFrame index hm, columns MultiIndex (type,strike) for 'oi' and 'close','open','low','high'
need={}
for d in days:
    for e in {nexp(d),nexp(d,True)}:
        if e is not None: need.setdefault(e,[]).append(d)
cols=['timestamp','option_type','strike','open','high','low','close','open_interest']
for e,ds in sorted(need.items()):
    if e in EXPS:
        t=pq.read_table(OD+e.strftime('%Y-%m-%d')+'.parquet',columns=cols).to_pandas()
        t['date']=t.timestamp.dt.normalize(); t=t[t.date.isin(ds)]
        t=t.rename(columns={'option_type':'type','open_interest':'oi'})
    else:
        fs=[]
        for d in ds:
            if d==pd.Timestamp('2026-10-26'):
                for f in glob.glob(T+'OPT_*.csv'):
                    x=rd(f); n=os.path.basename(f)[4:-4]; x['strike']=float(n[:-2]); x['type']=n[-2:]; x['oi']=x.oi; fs.append(x)
            else:
                for f in glob.glob(f'{R}/data/opt_oct/{d:%Y-%m-%d}_{e:%Y-%m-%d}_*.parquet'):
                    x=pd.read_parquet(f); x['timestamp']=x.ts; x['date']=d; fs.append(x)
        if not fs: continue
        t=pd.concat(fs)
        if e!=pd.Timestamp('2026-10-27') and pd.Timestamp('2026-10-26') in ds: pass
    t['hm']=t.timestamp.dt.strftime('%H:%M')
    for d,g in t.groupby('date'):
        g=g.drop_duplicates(['hm','type','strike'])
        OPT[(d,e)]=g.pivot(index='hm',columns=['type','strike'],values=['oi','open','high','low','close']).sort_index()
print('option day-panels',len(OPT),flush=True)
pd.to_pickle(OPT,f'{R}/data/opt_panels.pkl')
def hmm(h): return int(h[:2])*60+int(h[3:])-555
def mh(m): m+=555; return f'{m//60:02d}:{m%60:02d}'
rows=[]
for d in days:
    g=P[(P.date==d)&(P.hm<='15:14')].set_index('hm').sort_index()
    if g.close.isna().all() or len(g)<100: continue
    pc=DI.pc.get(d); pdl=DI.pdl.get(d); op=g.open.iloc[0]
    e0=nexp(d); O=OPT.get((d,e0))
    hi=g.high.values; lo=g.low.values; cl=g.close.values; H=list(g.index)
    RL=np.inf; tRL=None; trig_for=None; nday=0
    for i,h in enumerate(H):
        if np.isnan(lo[i]): continue
        if lo[i]<RL-0.001:
            RL=lo[i]; tRL=i
        if h>'14:14' or i==0 or nday>=3: continue
        hb=max(pc if pc==pc else -np.inf, np.nanmax(hi[:tRL+1]))
        fall=hb-RL
        if trig_for is not None and RL>=trig_for-5: continue   # one trigger per low episode (new low must be >=5 pts lower)
        if fall>=120 and cl[i]>=RL+30 and i-tRL<=15:
            trig_for=RL; nday+=1
            # outcome
            fut_lo=lo[i+1:]; fut_hi=hi[i+1:]
            brk=np.where(fut_lo<RL-5)[0]; b=brk[0] if len(brk) else len(fut_lo)
            mx=np.nanmax(np.r_[hi[i],fut_hi[:b]]) if b>0 else hi[i]
            r=dict(date=d,hm=h,t_low=H[tRL],RL=RL,entry=cl[i],pc=pc,pdl=pdl,open=op,fall=fall,depth_pc=(pc-RL) if pc==pc else np.nan,
                   min_since_open=tRL,bounce_min=i-tRL,below_pdl=RL<pdl if pdl==pdl else np.nan,gap_down=(op<=pc-50) if pc==pc else np.nan,
                   round_dist=abs(RL-round(RL/100)*100),maxrun=mx-RL,day_low=np.nanmin(lo),
                   y150=mx>=RL+150,y200=mx>=RL+200,y250=mx>=RL+250,final_low=np.nanmin(lo[i:])>=RL-0.001)
            # fall speed
            j0=max(0,tRL-15); r['fall_speed']=(np.nanmax(hi[j0:tRL+1])-RL)/max(tRL-j0,1)
            # futures basis & OI
            k10=max(0,tRL-10); f=g.fut.values; oi=g.foi_front.values
            r['basis_chg']=(f[i]-cl[i])-(f[k10]-cl[k10]); r['basis_now']=f[i]-cl[i]
            r['futoi_chg_pct']=(oi[i]/oi[tRL]-1)*100 if oi[tRL]>0 else np.nan
            r['futoi_fall_pct']=(oi[tRL]/oi[k10]-1)*100 if oi[k10]>0 else np.nan
            # BN lead
            try:
                bn=BN.loc[d].sort_index(); bnw=bn.loc[:h]; tb=bnw.low.idxmin()
                r['bn_lead_min']=hmm(H[tRL])-hmm(tb); r['bn_bounce_pct']=(bnw.close.iloc[-1]/bnw.low.min()-1)*100
                r['nf_bounce_pct']=(cl[i]/RL-1)*100
            except Exception: pass
            try:
                vx=VIX.loc[d].sort_index(); r['vix_chg']=vx.close.get(h)-vx.close.get(H[tRL]); r['vix_day']=vx.close.get(h)-vx.open.iloc[0]; r['vix_lvl']=vx.close.get(h)
            except Exception: pass
            r['dte']=(e0-d).days if e0 is not None else np.nan; r['weekday']=d.dayofweek
            if O is not None:
                oi_=O['oi']; c_=O['close']
                def osum(tp,ks,hh):
                    v=[oi_[(tp,k)].get(hh,np.nan) for k in ks if (tp,k) in oi_.columns]; v=[x for x in v if x==x]; return np.sum(v) if v else np.nan
                kb=np.floor(RL/50)*50
                pk=[kb-50,kb,kb+50,kb+100]; ck=[kb+100,kb+150,kb+200,kb+250]
                a=osum('PE',pk,H[k10]); b2=osum('PE',pk,h); r['put_oi_chg_pct']=(b2/a-1)*100 if a and a==a and a>0 else np.nan
                a=osum('CE',ck,H[k10]); b2=osum('CE',ck,h); r['call_oi_chg_pct']=(b2/a-1)*100 if a and a==a and a>0 else np.nan
                # heavy put strike near low
                win=[k for (tp,k) in oi_.columns if tp=='PE' and RL-200<=k<=RL+200]
                pv={k:oi_[('PE',k)].get(H[tRL],np.nan) for k in win}; pv={k:v for k,v in pv.items() if v==v}
                if pv:
                    ks=max(pv,key=pv.get); r['heavy_put_strike']=ks; r['heavy_put_dist']=RL-ks
                    r['heavy_put_ratio']=pv[ks]/np.median(list(pv.values()))
                K=round(cl[i]/50)*50
                try:
                    s1=c_[('CE',K)].get(h)+c_[('PE',K)].get(h); s0=c_[('CE',K)].get(H[tRL])+c_[('PE',K)].get(H[tRL]); r['straddle_chg_pct']=(s1/s0-1)*100
                except Exception: pass
                # PCR change (near strikes)
            rows.append(r)
C=pd.DataFrame(rows); C.to_csv(f'{R}/results/candidates.csv',index=False)
print(len(C),'candidates on',C.date.nunique(),'days; y200 rate',C.y200.mean().round(3))
print(C.tail(4).T)