import pandas as pd,numpy as np,sys,glob,os,json,warnings; warnings.filterwarnings('ignore')
sys.path.insert(0,'scripts'); from load import *
from scipy.stats import norm; from scipy.optimize import brentq
R='results'; W=pd.read_csv('data/implied_weights.csv',index_col=0); C=W.pts_per_rupee
gaps=[]
# ---------- 1. daily index / futures ----------
nd=day('NIFTY'); bd=day('BANKNIFTY'); vd=day('INDIAVIX')
N1=m1('NIFTY'); B1=m1('BANKNIFTY')
allsess=sorted(nd.index); prev={d:allsess[allsess.index(d)-1] for d in SESS}
def fobhav(d):
    return pd.read_csv(f'data/nse/fo_bhav_{d.replace("-","")}.csv.zip')
rows=[]
for d in SESS:
    p=prev[d]; n=nd.loc[d]; b=bd.loc[d]; x=N1[N1.date==d].set_index('hm')
    fb=fobhav(d); fp=fobhav(p) if os.path.exists(f'data/nse/fo_bhav_{p.replace("-","")}.csv.zip') else None
    r=dict(date=d,n_open=n.open,n_high=n.high,n_low=n.low,n_close_official=n.close,n_prev_close=nd.loc[p].close,
           n_chg=n.close-nd.loc[p].close,n_chg_pct=100*(n.close/nd.loc[p].close-1),n_range=n.high-n.low,
           n_gap=n.open-nd.loc[p].close,n_1514=x.loc['15:14','close'],auction_jump=n.close-x.loc['15:14','close'],
           n_hi_time=x.high.idxmax(),n_lo_time=x.low.idxmin(),
           bn_close=b.close,bn_chg=b.close-bd.loc[p].close,bn_chg_pct=100*(b.close/bd.loc[p].close-1),bn_range=b.high-b.low,vix=vd.loc[d].close,vix_chg=vd.loc[d].close-vd.loc[p].close)
    for sym,pre in [('NIFTY','nf'),('BANKNIFTY','bnf')]:
        f=fb[(fb.TckrSymb==sym)&(fb.FinInstrmTp=='IDF')].sort_values('XpryDt'); lot=f.NewBrdLotQty.iloc[0]
        r[f'{pre}_near_settle']=f.SttlmPric.iloc[0]; r[f'{pre}_near_close']=f.ClsPric.iloc[0]
        r[f'{pre}_oi_contracts']=f.OpnIntrst.sum()/lot; r[f'{pre}_oi_chg_contracts']=f.ChngInOpnIntrst.sum()/lot
        r[f'{pre}_vol_contracts']=f.TtlTradgVol.sum(); r[f'{pre}_basis_near']=f.ClsPric.iloc[0]-(n.close if sym=='NIFTY' else b.close)
        pc=f.ClsPric.iloc[0]-f.PrvsClsgPric.iloc[0]; r[f'{pre}_px_chg']=pc
        oc=r[f'{pre}_oi_chg_contracts']
        r[f'{pre}_buildup']=('Long buildup' if pc>0 and oc>0 else 'Short buildup' if pc<0 and oc>0 else 'Short covering' if pc>0 else 'Long unwinding')
    rows.append(r)
daily=pd.DataFrame(rows); 
# check bhavcopy units: OpnIntrst may already be contracts? detect
# ---------- 2. contributions ----------
sd={s:day(s) for s in STK}
contrib=pd.DataFrame({d:{s:C[s]*(sd[s].loc[d].close-sd[s].loc[prev[d]].close) for s in STK} for d in SESS})
contrib.to_csv(f'{R}/contrib_daily_pts.csv')
daily['sum_contrib']=contrib.sum().values; daily['recon_resid']=daily.n_chg-daily.sum_contrib
pct=pd.DataFrame({d:{s:100*(sd[s].loc[d].close/sd[s].loc[prev[d]].close-1) for s in STK} for d in SESS}); pct.to_csv(f'{R}/stock_pct_daily.csv')
# windows
M={s:m1(s) for s in STK}
def px(df,d,hm,col='close'):
    x=df[(df.date==d)&(df.hm==hm)]; return x[col].iloc[0] if len(x) and pd.notna(x[col].iloc[0]) else np.nan
WIN=[('gap','prev','0915o'),('09:15-10:15','0915o','10:14'),('10:15-13:00','10:14','12:59'),('13:00-15:14','12:59','15:14'),('auction 15:15-close','15:14','official')]
wrows=[]
for d in SESS:
    for s in STK:
        pts={'prev':sd[s].loc[prev[d]].close,'official':sd[s].loc[d].close,'0915o':px(M[s],d,'09:15','open')}
        for h in ['10:14','12:59','15:14']: pts[h]=px(M[s],d,h)
        for name,a,b in WIN: wrows.append(dict(date=d,sym=s,window=name,pts=C[s]*(pts[b]-pts[a])))
    nn={'prev':nd.loc[prev[d]].close,'official':nd.loc[d].close,'0915o':px(N1,d,'09:15','open'),'10:14':px(N1,d,'10:14'),'12:59':px(N1,d,'12:59'),'15:14':px(N1,d,'15:14')}
    for name,a,b in WIN: wrows.append(dict(date=d,sym='NIFTY_ACTUAL',window=name,pts=nn[b]-nn[a]))
win=pd.DataFrame(wrows); win.to_csv(f'{R}/contrib_windows_long.csv',index=False)
winp=win.pivot_table(index=['date','window'],columns='sym',values='pts')
wsum=pd.DataFrame({'nifty_actual':winp['NIFTY_ACTUAL'],'sum_components':winp[STK].sum(1)}); wsum.to_csv(f'{R}/window_recon.csv')
# ---------- 3. volume forensics ----------
def addratio(df):
    df=df.copy(); df=df[df.hm<='15:14']
    piv=df.pivot_table(index='date',columns='hm',values='volume')
    base=piv.shift(1).rolling(20,min_periods=15).mean()   # previous 20 sessions, same minute
    ratio=(piv/base)
    return piv,base,ratio
VR={}; dayflow=[]
for s in STK+['NIFTYFUT','BANKNIFTYFUT']:
    df=m1(s) if s in STK else m1(s,clean=False)
    piv,base,ratio=addratio(df); VR[s]=ratio
    if s not in STK: continue
    for d in SESS:
        x=df[(df.date==d)&(df.hm<='15:14')].copy(); x['ret']=x.close.diff(); x.loc[x.index[0],'ret']=x.close.iloc[0]-x.open.iloc[0]
        up=x.volume[x.ret>0].sum(); dn=x.volume[x.ret<0].sum(); vwap=(x.volume*(x.high+x.low+x.close)/3).sum()/x.volume.sum()
        tot=x.volume.sum(); b20=base.loc[d].sum()
        dayflow.append(dict(date=d,sym=s,vol_0915_1514=tot,vol_ratio_20d=tot/b20,upvol_share=up/(up+dn),vwap=vwap,close_1514=x.close.iloc[-1],
            close_vs_vwap_pct=100*(x.close.iloc[-1]/vwap-1),spike_minutes=int((ratio.loc[d]>=4).sum()),max_ratio=ratio.loc[d].max(),max_ratio_hm=ratio.loc[d].idxmax()))
flow=pd.DataFrame(dayflow)
# delivery
dl=[]
for f in sorted(glob.glob('data/nse/sec_bhavdata_full_*.csv')):
    t=pd.read_csv(f); t.columns=[c.strip() for c in t.columns]; t=t[t.SERIES.str.strip()=='EQ']; t['SYMBOL']=t.SYMBOL.str.strip()
    t['date']=pd.to_datetime(t.DATE1.str.strip(),format='%d-%b-%Y').dt.strftime('%Y-%m-%d'); dl.append(t[t.SYMBOL.isin(STK)])
dl=pd.concat(dl).drop_duplicates(['SYMBOL','date']).sort_values('date')
for c in ['DELIV_PER','DELIV_QTY','TTL_TRD_QNTY','AVG_PRICE','TURNOVER_LACS']: dl[c]=pd.to_numeric(dl[c],errors='coerce')
dl['deliv_val_cr']=dl.DELIV_QTY*dl.AVG_PRICE/1e7; dl['turnover_cr']=dl.TURNOVER_LACS/100
dl=dl.sort_values(['SYMBOL','date'])
for c in ['DELIV_PER','deliv_val_cr','turnover_cr']: dl[c+'_20d']=dl.groupby('SYMBOL')[c].transform(lambda s:s.shift(1).rolling(20,min_periods=15).mean())
dl=dl.rename(columns={'SYMBOL':'sym'})
flow=flow.merge(dl[['sym','date','DELIV_PER','DELIV_PER_20d','deliv_val_cr','deliv_val_cr_20d','turnover_cr','turnover_cr_20d']],on=['sym','date'],how='left')
flow=flow.merge(contrib.stack().rename('contrib_pts').reset_index().rename(columns={'level_0':'sym','level_1':'date'}),on=['sym','date'])
flow=flow.merge(pct.stack().rename('chg_pct').reset_index().rename(columns={'level_0':'sym','level_1':'date'}),on=['sym','date'])
flow.to_csv(f'{R}/stock_day_flow.csv',index=False)
for d in SESS:
    if flow[(flow.date==d)].DELIV_PER.isna().any(): gaps.append(f'delivery missing {d}: '+','.join(flow[(flow.date==d)&flow.DELIV_PER.isna()].sym))
# index-weighted aggregates per day
wt=W.implied_weight_pct/100
agg=[]
for d in SESS:
    x=flow[flow.date==d].set_index('sym')
    agg.append(dict(date=d,w_vol_ratio=(x.vol_ratio_20d*wt).sum(),w_upvol_share=(x.upvol_share*wt).sum(),w_close_vs_vwap=(x.close_vs_vwap_pct*wt).sum(),
        w_deliv_pct=(x.DELIV_PER*wt).sum(),w_deliv_pct_20d=(x.DELIV_PER_20d*wt).sum(),deliv_val_cr=x.deliv_val_cr.sum(),deliv_val_cr_20d=x.deliv_val_cr_20d.sum(),
        turnover_cr=x.turnover_cr.sum(),turnover_cr_20d=x.turnover_cr_20d.sum(),n_adv=(x.chg_pct>0).sum(),n_dec=(x.chg_pct<0).sum()))
daily=daily.merge(pd.DataFrame(agg),on='date')
# index-weighted minute volume ratio + futures ratio -> timeline
tl=[]
for d in SESS:
    wr=sum(VR[s].loc[d].fillna(0)*wt[s] for s in STK)
    n=N1[N1.date==d].set_index('hm'); f=m1('NIFTYFUT',clean=False); f=f[f.date==d].set_index('hm')
    for hm in wr.index:
        tl.append(dict(date=d,hm=hm,nifty=n.close.get(hm),fut=f.close.get(hm),fut_vol=f.volume.get(hm),fut_oi=f.oi.get(hm),cash_wvol_ratio=wr[hm],fut_vol_ratio=VR['NIFTYFUT'].loc[d].get(hm)))
tl=pd.DataFrame(tl); tl.to_csv(f'{R}/minute_timeline.csv',index=False)
# bursts: 5-minute windows on Nifty futures
br=[]
for d in SESS:
    f=m1('NIFTYFUT',clean=False); f=f[(f.date==d)].reset_index(drop=True); vr=VR['NIFTYFUT'].loc[d]
    nv=m1('NIFTYFUT_NOV',clean=False); nv=nv[nv.date==d].set_index('hm').oi; f['oi']=f.oi+f.hm.map(nv).ffill()  # Oct+Nov OI (rollover-neutral); Upstox intraday OI is lagged/partial
    f['vr']=f.hm.map(vr); f['r5']=f.close-f.open.shift(4); f['v5']=f.vr.rolling(5).mean(); f['oi5']=f.oi-f.oi.shift(5)
    cand=f[(f.r5.abs()>=40)&(f.v5>=2)&(f.hm<='15:14')].copy()
    used=set()
    for i in cand.reindex(cand.r5.abs().sort_values(ascending=False).index).index:
        if any(abs(i-j)<10 for j in used): continue
        used.add(i); a=i-4; s=f.loc[a,'hm']; e=f.loc[i,'hm']
        # stock contributions inside burst, using minute closes (cash) prev-minute close -> end
        cs={}
        for sym in STK:
            m=M[sym]; m=m[m.date==d].reset_index(drop=True)
            if a-1<0: p0=sd[sym].loc[prev[d]].close
            else: p0=m.loc[a-1,'close']
            cs[sym]=C[sym]*(m.loc[i,'close']-p0) if i<len(m) else np.nan
        cs=pd.Series(cs).dropna(); vrs=pd.Series({sym:VR[sym].loc[d,s:e].mean() for sym in STK})
        top=cs.sort_values(ascending=f.loc[i,'r5']<0).head(5)
        oi_chg=f.loc[i,'oi']-(f.loc[a-1,'oi'] if a>0 else np.nan)
        side='BUY burst' if f.loc[i,'r5']>0 else 'SELL burst'
        kind=('short covering' if oi_chg<0 else 'fresh longs') if side=='BUY burst' else ('long unwinding' if oi_chg<0 else 'fresh shorts')
        br.append(dict(date=d,start=s,end=e,side=side,fut_pts=round(f.loc[i,'r5'],1),fut_vol_ratio=round(f.loc[i,'v5'],2),fut_oi_chg_units=oi_chg,classify=kind,
            cash_pts_sum=round(cs.sum(),1),top5=', '.join(f'{k} {v:+.1f}' for k,v in top.items()),
            top5_volratio=', '.join(f'{k} {vrs[k]:.1f}x' for k in top.index),n_stocks_vol_ge3x=int((vrs>=3).sum())))
bursts=pd.DataFrame(br).sort_values(['date','start']); bursts.to_csv(f'{R}/bursts.csv',index=False)
# spike list per stock (ratio>=5 and abs return)
sp=[]
for s in STK:
    m=M[s]
    for d in SESS:
        x=m[(m.date==d)&(m.hm<='15:14')].set_index('hm'); r=VR[s].loc[d]
        for hm in r[r>=5].index:
            sp.append(dict(date=d,hm=hm,sym=s,vol_ratio=r[hm],ret_pct=100*(x.close[hm]/x.open[hm]-1),pts=C[s]*(x.close[hm]-x.open[hm])))
spikes=pd.DataFrame(sp); spikes.to_csv(f'{R}/stock_volume_spikes.csv',index=False)
# ---------- 4. players ----------
fd=pd.read_csv('/workspace/market-live-stack/feeds/macro/history_daily.csv').set_index('date')
daily['fii_cash_cr']=daily.date.map(fd.fii_net_cr); daily['dii_cash_cr']=daily.date.map(fd.dii_net_cr)
po=[]
for d in [prev[SESS[0]]]+SESS:
    t=pd.read_csv(f'data/nse/fao_participant_oi_{pd.Timestamp(d).strftime("%d%m%Y")}.csv',skiprows=1); t.columns=[c.strip() for c in t.columns]
    for _,r in t.iterrows():
        if r['Client Type']=='TOTAL': continue
        po.append(dict(date=d,cat=r['Client Type'].strip(),fut_idx_long=r['Future Index Long'],fut_idx_short=r['Future Index Short'],
           call_long=r['Option Index Call Long'],call_short=r['Option Index Call Short'],put_long=r['Option Index Put Long'],put_short=r['Option Index Put Short'],
           fut_stk_long=r['Future Stock Long'],fut_stk_short=r['Future Stock Short']))
po=pd.DataFrame(po); po['fut_idx_net']=po.fut_idx_long-po.fut_idx_short; po['fut_idx_long_pct']=100*po.fut_idx_long/(po.fut_idx_long+po.fut_idx_short)
po['opt_net_call']=po.call_long-po.call_short; po['opt_net_put']=po.put_long-po.put_short
po=po.sort_values(['cat','date'])
for c in ['fut_idx_long','fut_idx_short','fut_idx_net','opt_net_call','opt_net_put']: po[c+'_chg']=po.groupby('cat')[c].diff()
po.to_csv(f'{R}/participant_oi.csv',index=False)
fs=[]
for d in SESS:
    x=pd.read_excel(f'data/nse/fii_stats_{d.replace("-","")}.xls',header=None)
    def row(lbl):
        r=x[x[0].astype(str).str.strip()==lbl].iloc[0]; return float(r[2])-float(r[4]),float(r[5])
    for lbl,k in [('INDEX FUTURES','fii_idxfut'),('NIFTY FUTURES','fii_niftyfut'),('BANKNIFTY FUTURES','fii_bnfut'),('INDEX OPTIONS','fii_idxopt'),('STOCK FUTURES','fii_stkfut')]:
        try: n_,oi=row(lbl); fs.append(dict(date=d,k=k,net_cr=n_,oi=oi))
        except Exception as e: gaps.append(f'fii_stats {d} {lbl} missing')
fs=pd.DataFrame(fs).pivot(index='date',columns='k',values='net_cr').add_suffix('_net_cr'); fs.to_csv(f'{R}/fii_deriv_stats.csv')
daily=daily.merge(fs.reset_index(),on='date',how='left')
fii=po[po.cat=='FII'].set_index('date')
daily['fii_idxfut_long_pct']=daily.date.map(fii.fut_idx_long_pct); daily['fii_idxfut_net_chg']=daily.date.map(fii.fut_idx_net_chg)
daily['fii_idxfut_long_chg']=daily.date.map(fii.fut_idx_long_chg); daily['fii_idxfut_short_chg']=daily.date.map(fii.fut_idx_short_chg)
for cat in ['Client','Pro','DII']:
    q=po[po.cat==cat].set_index('date'); daily[f'{cat.lower()}_idxfut_net_chg']=daily.date.map(q.fut_idx_net_chg)
# ---------- options ----------
def bs(S,K,T,r,sig,cp):
    d1=(np.log(S/K)+(r+sig*sig/2)*T)/(sig*np.sqrt(T)); d2=d1-sig*np.sqrt(T)
    return S*norm.cdf(d1)-K*np.exp(-r*T)*norm.cdf(d2) if cp=='CE' else K*np.exp(-r*T)*norm.cdf(-d2)-S*norm.cdf(-d1)
def iv(p,S,K,T,cp):
    try: return brentq(lambda v:bs(S,K,T,0.065,v,cp)-p,0.01,3)
    except: return np.nan
orow=[];strk=[]
for d in SESS:
    b=fobhav(d); o=b[(b.TckrSymb=='NIFTY')&(b.FinInstrmTp=='IDO')].copy(); lot=65
    exps=sorted(o.XpryDt.unique()); ex=[e for e in exps if e>=d]; e0=ex[0] if ex[0]>d else ex[1]  # skip same-day expiry
    S=o.UndrlygPric.iloc[0]
    def b76iv(e):
        k=o[o.XpryDt==e]; T=(pd.Timestamp(e+' 15:30')-pd.Timestamp(d+' 15:30')).total_seconds()/(365*86400)
        ce=k[k.OptnTp=='CE'].set_index('StrkPric'); pe=k[k.OptnTp=='PE'].set_index('StrkPric'); com=ce.index.intersection(pe.index)
        atm=min(com,key=lambda x:abs(x-S)); F=atm+ce.loc[atm,'ClsPric']-pe.loc[atm,'ClsPric']  # synthetic forward (put-call parity)
        def b76(F,K,T,v,cp):
            d1=(np.log(F/K)+v*v*T/2)/(v*np.sqrt(T)); d2=d1-v*np.sqrt(T)
            return F*norm.cdf(d1)-K*norm.cdf(d2) if cp=='CE' else K*norm.cdf(-d2)-F*norm.cdf(-d1)
        def iv_(p,K,cp):
            try: return brentq(lambda v:b76(F,K,T,v,cp)-p,0.01,3)
            except: return np.nan
        v=np.nanmean([iv_(ce.loc[atm,'ClsPric'],atm,'CE'),iv_(pe.loc[atm,'ClsPric'],atm,'PE')])
        return ce,pe,atm,100*v,F
    ce,pe,atm,ivn,F=b76iv(e0); _,_,_,ivm,_=b76iv('2026-10-27')
    ivc=ivn/100; ivp=ivm/100
    allce=o[o.OptnTp=='CE'].OpnIntrst.sum(); allpe=o[o.OptnTp=='PE'].OpnIntrst.sum()
    orow.append(dict(date=d,expiry=e0,underlying=S,atm=atm,atm_iv_near=100*ivc,atm_iv_oct27=100*ivp,synthetic_fwd=F,pcr_near=pe.OpnIntrst.sum()/ce.OpnIntrst.sum(),pcr_all=allpe/allce,
        max_ce_oi_strike=ce.OpnIntrst.idxmax(),max_ce_oi_lots=ce.OpnIntrst.max()/lot,max_pe_oi_strike=pe.OpnIntrst.idxmax(),max_pe_oi_lots=pe.OpnIntrst.max()/lot,
        top_ce_add=ce.ChngInOpnIntrst.idxmax(),top_ce_add_lots=ce.ChngInOpnIntrst.max()/lot,top_pe_add=pe.ChngInOpnIntrst.idxmax(),top_pe_add_lots=pe.ChngInOpnIntrst.max()/lot,
        ce_oi_chg_lots=ce.ChngInOpnIntrst.sum()/lot,pe_oi_chg_lots=pe.ChngInOpnIntrst.sum()/lot))
    for K in ce.index:
        if abs(K-S)<=1000 and K%100==0: strk.append(dict(date=d,expiry=e0,strike=K,ce_oi_lots=ce.loc[K,'OpnIntrst']/lot,pe_oi_lots=pe.loc[K,'OpnIntrst']/lot if K in pe.index else np.nan,ce_chg_lots=ce.loc[K,'ChngInOpnIntrst']/lot,pe_chg_lots=pe.loc[K,'ChngInOpnIntrst']/lot if K in pe.index else np.nan))
opt=pd.DataFrame(orow); opt.to_csv(f'{R}/nifty_options_daily.csv',index=False); pd.DataFrame(strk).to_csv(f'{R}/nifty_options_strikes.csv',index=False)
daily=daily.merge(opt,on='date',how='left')
daily.to_csv(f'{R}/daily_summary.csv',index=False)
json.dump(gaps,open(f'{R}/gaps.json','w'),indent=1)
pd.set_option('display.width',250)
print(daily.T.to_string()); print(gaps)