# Compact per-day panel for the call-wall study: nearest weekly expiry only (0DTE included), 1m, 09:15-15:29.
# Sources: recov/data/opt_panels.pkl (2024-10-21..2026-10-23), by_expiry parquet for 2024-10-01..10-18, callwall/data/today (26-Oct full day).
# No filling: minutes absent in a source stay NaN. Spot 15:15-15:29 since 2026-08-03 is already NaN in minute_panel_v (frozen feed).
import pandas as pd,numpy as np,glob,os,pyarrow.parquet as pq,pickle
B='/workspace/research/flow-forensics-2026-10'; R=f'{B}/recov'; C=f'{B}/callwall'
OD='/workspace/nifty-options-s3/data/options/1m/by_expiry/'
HM=[f'{(555+m)//60:02d}:{(555+m)%60:02d}' for m in range(375)]
P=pd.read_parquet(f'{B}/lows/data/minute_panel_v.parquet',columns=['open','high','low','close','fut','date','hm'])
P=P[P.date>='2024-10-01']
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
T=f'{C}/data/today/'
tn=rd(T+'NIFTY.csv'); tf=rd(T+'NIFTYFUT.csv').set_index('hm'); tn['fut']=tn.hm.map(tf.close)
# rule: spot 15:15-15:29 frozen since 2026-08-03 -> missing
tn.loc[tn.hm>='15:15',['open','high','low','close']]=np.nan
P=pd.concat([P,tn[P.columns]],ignore_index=True)
OPT=pd.read_pickle(f'{R}/data/opt_panels.pkl')
# 2024-10-01..10-18 from by_expiry files
ext={}
for e in ['2024-10-03','2024-10-10','2024-10-17','2024-10-24']:
    t=pq.read_table(OD+e+'.parquet',columns=['timestamp','option_type','strike','open','high','low','close','open_interest']).to_pandas()
    t['date']=t.timestamp.dt.normalize(); t=t[(t.date>='2024-10-01')&(t.date<'2024-10-21')]
    t=t.rename(columns={'option_type':'type','open_interest':'oi'}); t['hm']=t.timestamp.dt.strftime('%H:%M')
    for d,g in t.groupby('date'):
        g=g.drop_duplicates(['hm','type','strike'])
        ext[(d,pd.Timestamp(e))]=g.pivot(index='hm',columns=['type','strike'],values=['oi','open','high','low','close']).sort_index()
OPT.update(ext)
# today full day (exp 2026-10-27)
fs=[]
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:]; fs.append(x)
t=pd.concat(fs).drop_duplicates(['hm','type','strike'])
OPT[(pd.Timestamp('2026-10-26'),pd.Timestamp('2026-10-27'))]=t.pivot(index='hm',columns=['type','strike'],values=['oi','open','high','low','close']).sort_index()
byd={}
for (d,e) in OPT: byd.setdefault(d,[]).append(e)
out={}; gaps=[]
for d,g in P.groupby('date'):
    g=g.drop_duplicates('hm').set_index('hm').reindex(HM)
    if d not in byd: gaps.append((d,'no option panel')); continue
    e=min(x for x in byd[d] if x>=d); O=OPT[(d,e)]
    ks=sorted({k for (tp,k) in O['oi'].columns})
    mid=np.nanmedian(g.close.values)
    ks=[k for k in ks if abs(k-mid)<=1500 and k%50==0]
    if not ks or mid!=mid: gaps.append((d,f'no strikes within 1500 of spot (mid={mid})')); continue
    M={}
    for f in ['oi','open','high','low','close']:
        for tp in ['CE','PE']:
            M[f+'_'+tp]=np.column_stack([O[f][(tp,k)].reindex(HM).values.astype(float) if (tp,k) in O[f].columns else np.full(375,np.nan) for k in ks])
    out[d]=dict(exp=e,K=np.array(ks,float),o=g.open.values.astype(float),h=g.high.values.astype(float),l=g.low.values.astype(float),c=g.close.values.astype(float),fut=g.fut.values.astype(float),**M)
    nsp=np.isnan(out[d]['c'][:360]).sum()
    if nsp>5: gaps.append((d,f'spot missing {nsp} min before 15:15'))
pickle.dump(out,open(f'{C}/data/cw_days.pkl','wb'))
pd.DataFrame(gaps,columns=['date','gap']).to_csv(f'{C}/results/data_gaps.csv',index=False)
print('days',len(out),min(out),max(out)); print('gaps',len(gaps)); print(pd.DataFrame(gaps).to_string())