"""QQQ/SPY/GLD/BTC + interest-bearing USD cash. Standalone, no old-project writes.
Run python3 backtest.py (uses cached public data; fetch_data.py refreshes downloads).
All decisions use completed data only. Main specifications fixed before viewing results.
"""
from pathlib import Path
import json,hashlib,ast
import numpy as np
import pandas as pd
P=Path(__file__).resolve().parent;D=P/'data';OUT=P/'results';OUT.mkdir(exist_ok=True)
N,M,Q,MINH,VT,VW=25,20,.975,250,.15,40
# Data-quality boundary, NOT return optimization: prior actual trade minute 2017-03-23 had no prints.
STRICT_START='2017-03-24'
POOL=['QQQ','SPY','GLD','BTC']; COST={'QQQ':.0003,'SPY':.0003,'GLD':.0003,'BTC':.001}
TZ='America/New_York'

def momentum(C,n=N):
 x=np.arange(n,dtype=float);x-=x.mean();ss=(x*x).sum()
 def score(a):
  y=np.log(a);ym=y.mean();b=(x*(y-ym)).sum()/ss;sst=((y-ym)**2).sum()
  r2=1-((y-ym-b*x)**2).sum()/sst if sst>0 else 0
  return np.expm1(250*b)*r2
 return C.rolling(n).apply(score,raw=True)

def ny_time(d,h,m=0):return (pd.Timestamp(d)+pd.Timedelta(hours=h,minutes=m)).tz_localize(TZ)

def load(start_override=None,save_audit=True):
 raw={};adj={};ops={};acts={}
 for c in POOL[:-1]:
  x=pd.read_csv(D/f'{c}.csv',index_col=0,parse_dates=True).dropna(subset=['open','close','adjclose']);raw[c]=x
  fac=x.adjclose/x.close;adj[c]=x.adjclose;ops[c]=x.open*fac
  z=json.loads((D/f'{c}.json').read_text())['chart']['result'][0];acts[c]=z.get('events',{})
  assert not acts[c].get('splits'), 'New split event: review corporate-action conventions.'
 sessions=raw['SPY'].index
 assert all(raw[c].index.equals(sessions) for c in raw)
 btc=pd.read_csv(D/'BTCUSD_30min.csv',index_col='time',parse_dates=True)
 assert not btc.index.has_duplicates
 assert (btc[['open','high','low','close']]>0).all().all()
 # Previous bar ends precisely at 16:00 NY: this is known at decision time.
 sidx=pd.DatetimeIndex([ny_time(d,15,30).tz_convert('UTC') for d in sessions])
 bs=btc.reindex(sidx)
 assert not bs.close.isna().any()
 C=pd.DataFrame(adj,index=sessions);O=pd.DataFrame(ops,index=sessions)
 C['BTC']=bs.close.to_numpy()
 S=momentum(C);BIAS=C/C.rolling(M).mean()-1
 THR=BIAS.expanding(min_periods=MINH).quantile(Q);MED=BIAS.expanding(min_periods=MINH).quantile(.5)
 FL=(BIAS>THR).fillna(False);VOL=C.pct_change().rolling(VW).std()*np.sqrt(252)
 eligible=THR.notna().all(axis=1)&VOL.notna().all(axis=1)
 first_signal=sessions[np.flatnonzero(eligible)[0]]
 eligible_start=sessions[sessions.get_loc(first_signal)+1]
 start=max(eligible_start,pd.Timestamp(start_override or STRICT_START))
 start=sessions[sessions>=start][0]
 first_signal=sessions[sessions<start][-1]
 dates=pd.date_range(start,sessions[-1],freq='D')
 # Both valuation and execution prices are true historical Bitstamp USD prices,
 # never USDT, never UTC daily close relabelled as NY close.
 close_ts=pd.DatetimeIndex([ny_time(d,15,30).tz_convert('UTC') for d in dates])
 open_ts=pd.DatetimeIndex([ny_time(d,9,30).tz_convert('UTC') for d in dates])
 DC=C.reindex(C.index.union(dates)).ffill().reindex(dates);DC['BTC']=btc.reindex(close_ts).close.to_numpy()
 DO=O.reindex(dates);DO.loc[~dates.isin(sessions),POOL[:-1]]=DC[POOL[:-1]].shift(1).loc[~dates.isin(sessions)]
 DO['BTC']=btc.reindex(open_ts).open.to_numpy()
 volume=pd.Series(btc.reindex(open_ts-pd.Timedelta(minutes=30)).volume.to_numpy(),index=dates)
 exec_volume=pd.Series(btc.reindex(open_ts).volume.to_numpy(),index=dates)
 assert not DO.isna().any().any() and not DC.isna().any().any()
 # FRED discount-basis 91-day bill -> approximate annual investment yield.
 # Quote deliberately held back FIVE calendar days, preventing publication lookahead.
 rf=pd.read_csv(D/'DTB3.csv');rf.columns=['date','rate'];rf['rate']=pd.to_numeric(rf.rate,errors='coerce');rf=rf.dropna()
 available=pd.DatetimeIndex(pd.to_datetime(rf.date)+pd.Timedelta(days=5)).tz_localize(TZ)
 dr=rf.rate.to_numpy()/100; iy=dr*(365/360)/(1-dr*91/360)
 def known_rate(ts):
  k=available.searchsorted(ts,side='right')-1
  assert k>=0
  return iy[k]
 prev=ny_time(start-pd.Timedelta(days=1),16)
 rf1=[];rf2=[]
 for d in dates:
  op=ny_time(d,9,30);cl=ny_time(d,16)
  rf1.append(np.expm1(known_rate(prev)*(op-prev).total_seconds()/(365*86400)))
  rf2.append(np.expm1(known_rate(op)*(cl-op).total_seconds()/(365*86400)))
  prev=cl
 ctx=dict(C=C,O=O,S=S,BIAS=BIAS,THR=THR,MED=MED,FL=FL,VOL=VOL,sessions=sessions,dates=dates,DC=DC,DO=DO,volume=volume,
          exec_volume=exec_volume,rf1=np.array(rf1),rf2=np.array(rf2),start=start,first_signal=first_signal,acts=acts)
 for k in (['C','O','S','BIAS','THR','MED','FL','VOL','DC','DO'] if save_audit else []):ctx[k].to_csv(OUT/f'audit_{k}.csv')
 return ctx

def simulate(ctx,name,bias=False,cool=False,vt=False,zero=False,high=False,pool=None,buyhold=None,start_override=None):
 pool=pool or POOL;S,B,F,MED,V=(ctx[k] for k in ['S','BIAS','FL','MED','VOL'])
 dates=ctx['dates'];sessions=ctx['sessions'];DC=ctx['DC'];DO=ctx['DO'];cols=POOL
 w=np.zeros(4);cash=1.;nav=1.;H=None;intent=None;pending=None;blocked=set();events=[];signals=[];records=[];blocked_events=[]
 costs=np.array([COST[c] for c in cols])*(1 if not high else 1) 
 if high:costs=np.array([.001,.001,.001,.005])
 start=start_override or ctx['start']
 def decide(d):
  nonlocal H,intent,pending
  prevH=H
  if cool:
   for c in list(blocked):
    if B.loc[d,c]<MED.loc[d,c]:blocked.remove(c);blocked_events.append((str(d.date()),c,'unblock'))
   if H is not None and F.loc[d,H]:blocked.add(H);blocked_events.append((str(d.date()),H,'block'))
  ok=[c for c in pool if pd.notna(S.loc[d,c]) and S.loc[d,c]>0 and (not bias or not F.loc[d,c]) and c not in blocked]
  ok.sort(key=lambda c:-S.loc[d,c]);X=ok[0] if ok else None
  if buyhold:X=buyhold
  H=X
  if X!=intent or (intent is None and not records):
   k=min(1.,VT/V.loc[d,X]) if vt and X and V.loc[d,X]>0 else (1. if X else 0.)
   tgt=np.zeros(4)
   if X:tgt[cols.index(X)]=k
   reason='momentum'
   if prevH and bias and F.loc[d,prevH]:reason='BIAS_cooldown' if cool else 'BIAS_filter'
   elif X is None:reason='no_positive_eligible_asset'
   pending=(tgt,d,reason,X)
   intent=X
  signals.append(dict(date=str(d.date()),selected=X or 'CASH',blocked=','.join(sorted(blocked))))
 first=sessions[sessions<start][-1];decide(first)
 # Previous calendar-day ETF marks from last complete session; BTC from last bar before 16:00.
 prevclose=ctx['C'].loc[first,cols].to_numpy().copy()
 # Start is always the next session after first, but there can be a weekend gap;
 # initial portfolio is all cash so the initial BTC mark has no economic effect.
 for i,d in enumerate(dates):
  if d<start:continue
  op=DO.loc[d,cols].to_numpy();cl=DC.loc[d,cols].to_numpy()
  rf1=0. if zero else ctx['rf1'][i];rf2=0. if zero else ctx['rf2'][i]
  part1=w*(op/prevclose-1);cashpart1=cash*rf1
  g1=1+part1.sum()+cashpart1
  w=w*(op/prevclose)/g1;cash=cash*(1+rf1)/g1
  logs=np.zeros(6)
  if abs(g1-1)>1e-14:logs[:4]+=part1*np.log(g1)/(g1-1);logs[4]+=cashpart1*np.log(g1)/(g1-1)
  tr=0.;fee=0.;count=0
  if pending is not None and d in sessions:
   target,sd,why,X=pending
   # The COMPLETED 09:00-09:30 candle gives a historical, already-known liquidity check.
   # Defer the whole switch when this prior candle has no prints.
   # Current-candle volume never determines the decision; assert ex post that every
   # actual BTC fill's 09:30 candle does contain trades.
   needsbtc=(w[3]>1e-12 or target[3]>1e-12)
   if not needsbtc or ctx['volume'].loc[d]>0:
    if needsbtc:assert ctx['exec_volume'].loc[d]>0, ('No actual BTC print at fill',str(d))
    delta=target-w;tr=float(np.abs(delta).sum());fee=float((np.abs(delta)*costs).sum());count=int((np.abs(delta)>1e-10).sum())
    events.append(dict(signal_date=str(sd.date()),execution_date=str(d.date()),execution_time_PT='06:30',selected=X or 'CASH',reason=why,
      turnover_gross=tr,cost_fraction=fee,legs=count,**{f'target_{c}':target[j] for j,c in enumerate(cols)},
      **{f'before_{c}':w[j] for j,c in enumerate(cols)},**{f'price_{c}':op[j] for j,c in enumerate(cols)}))
    w=target.copy();cash=1-w.sum();pending=None
   else:events.append(dict(signal_date=str(sd.date()),execution_date=str(d.date()),selected=X or 'CASH',reason='DEFERRED_BTC_NO_TRADES',turnover_gross=0,cost_fraction=0,legs=0))
  assert cash>=-1e-10 and w.min()>=-1e-10 and w.sum()<=1+1e-10
  gcost=1-fee;logs[5]+=np.log(gcost)
  part2=w*(cl/op-1);cashpart2=cash*rf2;g2=1+part2.sum()+cashpart2
  if abs(g2-1)>1e-14:logs[:4]+=part2*np.log(g2)/(g2-1);logs[4]+=cashpart2*np.log(g2)/(g2-1)
  w=w*(cl/op)/g2;cash=cash*(1+rf2)/g2
  ret=g1*gcost*g2-1;nav*=1+ret
  assert np.isclose(logs.sum(),np.log1p(ret),atol=1e-10)
  records.append(dict(date=d,r=ret,nav=nav,rf=(1+ctx['rf1'][i])*(1+ctx['rf2'][i])-1,turnover=tr,legs=count,fee=fee,cash=cash,
    **{c:w[j] for j,c in enumerate(cols)},**{f'log_{c}':logs[j] for j,c in enumerate(cols+['cash','cost'])}))
  if d in sessions:decide(d)
  prevclose=cl.copy()
 df=pd.DataFrame(records).set_index('date');ev=pd.DataFrame(events);sg=pd.DataFrame(signals)
 return dict(daily=df,events=ev,signals=sg,blocked=sorted(blocked),pending=pending,block_events=blocked_events)

def stats(res,start=None):
 d=res['daily'];d=d.loc[start:] if start else d
 r=d.r;rf=d.rf;nav=(1+r).cumprod();nav0=np.r_[1.,nav.to_numpy()];dd=nav0/np.maximum.accumulate(nav0)-1
 yrs=len(d)/365.25
 return {'CAGR':float(nav.iloc[-1]**(1/yrs)-1),'MDD':float(dd.min()),'Sharpe_excess_cash':float((r-rf).mean()/(r-rf).std(ddof=1)*np.sqrt(365.25)),
 'Volatility':float(r.std(ddof=1)*np.sqrt(365.25)),'Turnover_oneway_per_year':float(d.turnover.sum()/yrs),'Turnover_roundtrip_per_year':float(d.turnover.sum()/2/yrs),
 'Trade_legs':int(d.legs.sum()),'Switches':int((d.turnover>1e-10).sum()),'BTC_average_weight':float(d.BTC.mean()),'Cash_average_weight':float(d.cash.mean()),'Terminal_NAV':float(nav.iloc[-1]),'Years':yrs}

def main():
 ctx=load();print('SAMPLE',ctx['start'],ctx['dates'][-1],len(ctx['dates']),flush=True)
 specs={'Momentum':{},'Momentum_VT':dict(vt=True),'BIAS_filter_only':dict(bias=True),'C3_cooldown':dict(bias=True,cool=True),'C4_VT_cooldown':dict(bias=True,cool=True,vt=True),
 'SPY_BH':dict(buyhold='SPY'),'QQQ_BH':dict(buyhold='QQQ'),'BTC_BH':dict(buyhold='BTC')}
 allres={};summary={};sub={};yearly={};sensitivity={}
 for name,kw in specs.items():
  z=simulate(ctx,name,**kw);allres[name]=z;summary[name]=stats(z)
  z['daily'].to_csv(OUT/f'{name}_daily.csv');z['events'].to_csv(OUT/f'{name}_trades.csv',index=False);z['signals'].to_csv(OUT/f'{name}_signals.csv',index=False)
  yearly[name]=z['daily'].r.groupby(z['daily'].index.year).apply(lambda a:(1+a).prod()-1)
  for st in ['2019-01-01','2020-01-01']:sub[(name,st)]=stats(z,st)
  print(name,summary[name],flush=True)
  if name not in ['SPY_BH','QQQ_BH','BTC_BH']:
   for label,extra in [('zero_cash',dict(zero=True)),('high_cost',dict(high=True)),('no_BTC',dict(pool=POOL[:-1]))]:
    zz=simulate(ctx,name,**kw,**extra);sensitivity[(name,label)]=stats(zz)
    zz['daily'].to_csv(OUT/f'{name}_{label}_daily.csv')
 pd.DataFrame(summary).T.to_csv(OUT/'summary.csv');pd.DataFrame(sub).T.to_csv(OUT/'subperiods.csv');pd.DataFrame(yearly).to_csv(OUT/'yearly_returns.csv');pd.DataFrame(sensitivity).T.to_csv(OUT/'sensitivity.csv')
 attrib=pd.DataFrame({n:{c:z['daily'][f'log_{c}'].sum() for c in POOL+['cash','cost']} for n,z in allres.items()}).T;attrib.to_csv(OUT/'log_return_attribution.csv')
 latest={n:{'signal':z['signals'].iloc[-1].to_dict(),'weights':z['daily'][POOL+['cash']].iloc[-1].to_dict(),'blocked':z['blocked'],'pending':None if z['pending'] is None else {'weights':z['pending'][0].tolist(),'signal_date':str(z['pending'][1].date()),'selected':z['pending'][3] or 'CASH'},'block_events_count':len(z['block_events'])} for n,z in allres.items()}
 (OUT/'latest.json').write_text(json.dumps(latest,indent=2))
 plot(allres,summary,ctx)
 metadata=dict(start=str(ctx['start'].date()),end=str(ctx['dates'][-1].date()),first_signal=str(ctx['first_signal'].date()),parameters=dict(N=N,M=M,Q=Q,MINH=MINH,VT=VT,VW=VW,cost=COST),
  source_urls=['https://query1.finance.yahoo.com/v8/finance/chart/QQQ','https://query1.finance.yahoo.com/v8/finance/chart/SPY','https://query1.finance.yahoo.com/v8/finance/chart/GLD','https://www.bitstamp.net/api/v2/ohlc/btcusd/','https://www.bitstamp.net/api/','https://fred.stlouisfed.org/series/DTB3','https://fred.stlouisfed.org/graph/fredgraph.csv?id=DTB3','https://www.federalreserve.gov/releases/h15/'])
 (OUT/'metadata.json').write_text(json.dumps(metadata,indent=2));print('DONE',OUT,flush=True)

def plot(res,summary,ctx):
 import matplotlib;matplotlib.use('Agg');import matplotlib.pyplot as plt
 fig,axs=plt.subplots(2,1,figsize=(13,8),sharex=True,gridspec_kw={'height_ratios':[2,1]})
 names=['Momentum','Momentum_VT','C3_cooldown','C4_VT_cooldown','SPY_BH','QQQ_BH','BTC_BH']
 colors=['#d97706','#38bdf8','#ef4444','#059669','#475569','#a855f7','#999999']
 for n,c in zip(names,colors):
  d=res[n]['daily'];nav=d.nav;dd=nav/np.maximum.accumulate(np.r_[1,nav.to_numpy()])[1:]-1
  label=f"{n}: CAGR {summary[n]['CAGR']:.1%}, MDD {summary[n]['MDD']:.1%}"
  axs[0].plot(nav.index,nav,label=label,color=c,lw=1.7 if 'C3' in n or 'C4' in n else 1,alpha=.9)
  axs[1].plot(dd.index,100*dd,color=c,lw=.9)
 axs[0].set_yscale('log');axs[0].set_ylabel('Net asset value (log)');axs[1].set_ylabel('Drawdown %');axs[0].legend(fontsize=8,ncol=2)
 axs[0].set_title(f"QQQ / SPY / GLD / BTC + USD cash | {ctx['start'].date()} - {ctx['dates'][-1].date()}\nPT 13:00 signals; next US session PT 06:30 fills | ETF 3bp / BTC 10bp per side")
 for ax in axs:ax.grid(alpha=.2)
 fig.tight_layout();fig.savefig(P/'nav_drawdown.png',dpi=160);plt.close(fig)

if __name__=='__main__':main()