"""Frozen, reproducible research. No trading. python SCRIPT INPUT_DIR OUTPUT_DIR. Inputs: spot.json.gz (daily OHLCV and universe), derivatives.json (hourly/settlements). """ import csv,gzip,json,statistics,sys from collections import Counter,defaultdict from datetime import datetime,timezone from pathlib import Path DAY=86400;HOUR=3600 VERSION='four-investor-decisions-1.0' def iso(t):return datetime.fromtimestamp(t,timezone.utc).isoformat().replace('+00:00','Z') def median(v):return statistics.median(v) if v else None def daily(d):return {int(t):b for t,b in d.items()} def complete(d,t,end,step):return all(u in d for u in range(t,end+1,step)) def maxdd(values): peak=values[0];dd=0 for v in values:peak=max(peak,v);dd=min(dd,v/peak-1) return dd def summary(rows,key): values=[r[key] for r in rows if r.get(key) is not None] return dict(n=len(values),median=median(values),mean=statistics.mean(values) if values else None, min=min(values) if values else None,max=max(values) if values else None, positive_n=sum(v>0 for v in values),negative_n=sum(v<0 for v in values)) def write(out,name,data): (out/(name+'.json')).write_text(json.dumps(data,separators=(',',':'),allow_nan=False)+'\n') def csvout(out,name,rows): keys=list(dict.fromkeys(k for r in rows for k in r)) with (out/(name+'.csv')).open('w',newline='') as f: w=csv.DictWriter(f,fieldnames=keys);w.writeheader() for r in rows:w.writerow({k:json.dumps(v,separators=(',',':')) if isinstance(v,(list,dict)) else v for k,v in r.items()}) ALLOC={'BTC':{'BTCUSDT':1},'ETH':{'ETHUSDT':1},'SOL':{'SOLUSDT':1},'BTC50_ETH50':{'BTCUSDT':.5,'ETHUSDT':.5},'BTC60_ETH30_SOL10':{'BTCUSDT':.6,'ETHUSDT':.3,'SOLUSDT':.1}} def portfolio(series,t,days,weights,cost=.0015,points=False): q={s:w/(series[s][t]['o']*(1+cost)) for s,w in weights.items()} path=[dict(time=t,equity=sum(q[s]*series[s][t]['o']*(1-cost) for s in q))] for u in range(t,t+days*DAY,DAY):path.append(dict(time=u+DAY,equity=sum(q[s]*series[s][u]['c']*(1-cost) for s in q))) terminal=sum(q[s]*series[s][t+days*DAY]['o']*(1-cost) for s in q) path[-1]['equity']=terminal r=dict(net_return=terminal-1,drawdown=maxdd([1]+[p['equity'] for p in path])) if points:r['points']=path return r def allocation(series): symbols=['BTCUSDT','ETHUSDT','SOLUSDT'];start=max(min(series[s]) for s in symbols);end=min(max(series[s]) for s in symbols) rows=[];candidates=[] for t in range(start,end+DAY,DAY): if datetime.fromtimestamp(t,timezone.utc).day!=1:continue for days in (90,180,365): status='pending' if t+days*DAY>end else ('complete' if all(complete(series[s],t,t+days*DAY,DAY) for s in symbols) else 'missing') candidates.append(dict(entry=t,entry_utc=iso(t),days=days,status=status)) if status!='complete':continue for policy,w in ALLOC.items(): for cost in (.0005,.0015,.003): rows.append(dict(id=f'{t}:{days}:{policy}:{cost}',entry=t,entry_utc=iso(t),days=days,policy=policy,cost=cost,**portfolio(series,t,days,w,cost))) main=[r for r in rows if r['cost']==.0015];summ=[] for days in (90,180,365): for p in ALLOC: rs=[r for r in main if r['days']==days and r['policy']==p] summ.append(dict(days=days,policy=p,**summary(rs,'net_return'),median_drawdown=median([r['drawdown'] for r in rs]))) anchor=[];previous=None for t in sorted({r['entry'] for r in main if r['days']==365}): if previous is None or t>=previous+365*DAY:anchor.append(t);previous=t return dict(version=VERSION,start=iso(start),last_open=iso(end),candidates=candidates,observations=rows,summary=summ, nonoverlap_365=[dict(policy=p,**summary([r for r in main if r['days']==365 and r['policy']==p and r['entry'] in anchor],'net_return')) for p in ALLOC]) def rotation_path(alt,btc,t,days,cost=.0015,points=False): a0=alt[t]['o'];b0=btc[t]['o'];qbtc=(1-cost)/(b0*(1+cost));path=[] for u in range(t,t+days*DAY+1,DAY): # Entry and terminal at open; intermediate values at previous UTC close. ap=alt[u]['o'] if u in (t,t+days*DAY) else alt[u-DAY]['c'] bp=btc[u]['o'] if u in (t,t+days*DAY) else btc[u-DAY]['c'] path.append(dict(time=u,hold=ap/a0*(1-cost),rotate=qbtc*bp*(1-cost),cash=1-cost)) last=path[-1];r=dict(hold_return=last['hold']-1,rotate_return=last['rotate']-1,cash_return=-cost, rotate_minus_hold=last['rotate']-last['hold'],hold_drawdown=maxdd([1]+[p['hold'] for p in path]),rotate_drawdown=maxdd([1]+[p['rotate'] for p in path])) if points:r['points']=path return r def rotation(series,universe): btc=series['BTCUSDT'];rows=[];candidates=[];cutoff=max(btc) for record in universe: if record.get('excluded_reason') is not None:continue symbol=record['symbol'];bars=series.get(symbol,{}) if len(bars)<182:continue last={h:None for h in (.3,.5,.7)} for t in sorted(bars): if t-DAY not in bars:continue if not complete(bars,t-180*DAY,t,DAY):continue priorpeak=max(bars[u]['c'] for u in range(t-180*DAY,t,DAY)) avgq=statistics.mean(bars[u]['qv'] for u in range(t-30*DAY,t,DAY)) if avgq<1e6:continue before=bars[t-DAY]['c']/priorpeak-1;now=bars[t]['c']/priorpeak-1 for threshold in (.3,.5,.7): if not before>-threshold>=now:continue if last[threshold] is not None and t-last[threshold]<90*DAY:continue last[threshold]=t;entry=t+DAY for days in (30,90,180): end=entry+days*DAY status='pending' if end>cutoff else ('complete' if complete(bars,entry,end,DAY) and complete(btc,entry,end,DAY) else 'missing_path') r=dict(id=f'{symbol}:{t}:{threshold}:{days}',symbol=symbol,signal=t,signal_utc=iso(t+DAY),entry=entry,entry_utc=iso(entry),days=days,threshold=threshold,prior_peak=priorpeak,signal_close=bars[t]['c'],trailing30_quote_volume=avgq,status=status) r['prior60_relative_return']=((bars[t]['c']/bars[t-60*DAY]['c'])/(btc[t]['c']/btc[t-60*DAY]['c'])-1) if t in btc and t-60*DAY in btc else None candidates.append(r) if status=='complete': for cost in (.0005,.0015,.003):rows.append(r|dict(id=r['id']+':'+str(cost),candidate_id=r['id'],cost=cost,**rotation_path(bars,btc,entry,days,cost))) main=[r for r in rows if r['cost']==.0015];summ=[] for h in (.3,.5,.7): for days in (30,90,180): rs=[r for r in main if r['threshold']==h and r['days']==days] summ.append(dict(threshold=h,days=days,**summary(rs,'rotate_minus_hold'),hold_median=median([r['hold_return'] for r in rs]),rotate_median=median([r['rotate_return'] for r in rs]),hold_recovery_n=sum((1+r['hold_return'])*series[r['symbol']][r['entry']]['o']>=r['prior_peak'] for r in rs),symbols=len({r['symbol'] for r in rs}))) return dict(version=VERSION,candidates=candidates,observations=rows,summary=summ,status_counts=dict(Counter(r['status'] for r in candidates)),symbols_with_candidates=len({r['symbol'] for r in candidates})) def funding_valid(rates,start,end): """Check actual archive-declared intervals, not an assumed fixed 8h schedule.""" times=sorted(rates) inside=[t for t in times if start<=t<=end] if not inside:return False idx={t:i for i,t in enumerate(times)} previous=max((t for t in times if t<=start),default=None) following=min((t for t in times if t>=end),default=None) if previous is None or following is None:return False for t in times[idx[previous]+1:idx[following]+1]: p=times[idx[t]-1] if abs(t-p-rates[t]['interval_hours']*HOUR)>1:return False return True def boundary(q,entry,wallet,side,mmr=.005,fee=.0005): return (q*entry-wallet)/(q*(1-mmr-fee)) if side==1 else (wallet+q*entry)/(q*(1+mmr+fee)) def leverage_path(marks,trades,rates,t,days,L,side,mmr=.005,points=False): end=t+days*DAY;entry=trades[t]['o'];q=1/entry;wallet=1/L-.0005;funding=0.;breach=None;path=[] for u in range(t,end,HOUR): if t=level):breach=u if points and (u-t)%HOUR==0:path.append(dict(time=u,price=marks[u]['c']/entry*100,boundary=level/entry*100,equity=(wallet+side*q*(marks[u]['c']-entry))*L*100,funding=funding*L*100)) r=dict(breach_time=breach,breach_utc=iso(breach) if breach is not None else None,endpoint_direction_correct=side*(trades[end]['o']/entry-1)>0,endpoint_price_return=trades[end]['o']/entry-1,counterfactual_net_return=(wallet+side*q*(trades[end]['o']-entry)-q*trades[end]['o']*.0005)*L-1,funding_paid_per_initial_notional=funding) if points:r['points']=path return r def arbitrage_path(spot,marks,trades,rates,t,days,slip=.0002,points=False): end=t+days*DAY;s0=spot[t]['o'];f0=trades[t]['o'];q=1/s0;collateral=q*f0 sf=.001+slip;ff=.0005+slip;entrycost=sf+q*f0*ff;capital=1+collateral+entrycost wallet=collateral;income=0.;first=None;breach=None;path=[] for u in range(t,end,HOUR): if t=level:breach=u if (u-t)%DAY==0: sp=spot[u]['o'];fp=trades[u]['o'];cost=entrycost+q*sp*sf+q*fp*ff;basis=q*((sp-s0)+(f0-fp));net=income+basis-cost if first is None and net>=0:first=(u-t)/DAY if points:path.append(dict(time=u,funding=income/capital*100,basis=basis/capital*100,cost=-cost/capital*100,net=net/capital*100,spot=sp/s0*100,future=fp/f0*100,margin_headroom=(wallet+income+q*(f0-marks[u]['o'])-(.005+ff)*q*marks[u]['o'])/capital*100)) sp=spot[end]['o'];fp=trades[end]['o'];cost=entrycost+q*sp*sf+q*fp*ff;basis=q*((sp-s0)+(f0-fp));net=income+basis-cost if first is None and net>=0:first=days if points:path.append(dict(time=end,funding=income/capital*100,basis=basis/capital*100,cost=-cost/capital*100,net=net/capital*100,spot=sp/s0*100,future=fp/f0*100,margin_headroom=(wallet+income+q*(f0-marks[end]['o'])-(.005+ff)*q*marks[end]['o'])/capital*100)) r=dict(initial_capital=capital,funding_income=income,basis_pnl=basis,total_cost=cost,net_pnl=net,counterfactual_net_return=net/capital,net_return=net/capital if breach is None else None,breach_time=breach,first_daily_break_even=first,first_break_even_then_loss=first is not None and net<0) if points:r['points']=path return r def derivative_studies(series,deriv): lev=[];arb=[];candidates=[];coverage=[] for symbol in ('BTCUSDT','ETHUSDT','SOLUSDT'): obj=deriv[symbol];marks=daily(obj['markPriceKlines']);trades=daily(obj['klines']);rates=daily(obj['fundingRate']);spot=series[symbol] start=max(min(marks),min(trades));last=min(max(marks),max(trades));end=last//DAY*DAY coverage.append(dict(symbol=symbol,first=iso(start),last=iso(last),mark_rows=len(marks),trade_rows=len(trades),settlements=len(rates),missing_mark_hours=sum(u not in marks for u in range(start,last+HOUR,HOUR)))) for t in range(start,end+DAY,DAY): if datetime.fromtimestamp(t,timezone.utc).weekday()!=0:continue for days in (7,30,90): finish=t+days*DAY;status='pending' if finish>end else ('complete' if complete(marks,t,finish,HOUR) and complete(trades,t,finish,HOUR) and complete(spot,t,finish,DAY) and funding_valid(rates,t,finish) else 'missing') candidates.append(dict(symbol=symbol,entry=t,days=days,status=status)) if status!='complete':continue for slip in (0.,.0002,.001):arb.append(dict(id=f'{symbol}:{t}:{days}:{slip}',symbol=symbol,entry=t,entry_utc=iso(t),days=days,slip=slip,**arbitrage_path(spot,marks,trades,rates,t,days,slip))) if days==90:continue for L in (2,5,10): for side in (1,-1): for mmr in (.004,.005,.01):lev.append(dict(id=f'{symbol}:{t}:{days}:{L}:{side}:{mmr}',symbol=symbol,entry=t,entry_utc=iso(t),days=days,leverage=L,side=side,mmr=mmr,**leverage_path(marks,trades,rates,t,days,L,side,mmr))) ls=[];ars=[] for days in (7,30): for L in (2,5,10): for side in (1,-1): rs=[r for r in lev if r['days']==days and r['leverage']==L and r['side']==side and r['mmr']==.005];correct=[r for r in rs if r['endpoint_direction_correct']] ls.append(dict(days=days,leverage=L,side=side,n=len(rs),breach_n=sum(r['breach_time'] is not None for r in rs),correct_n=len(correct),correct_but_breached_n=sum(r['breach_time'] is not None for r in correct))) for days in (7,30,90): for symbol in ('BTCUSDT','ETHUSDT','SOLUSDT'): rs=[r for r in arb if r['symbol']==symbol and r['days']==days and r['slip']==.0002] ars.append(dict(symbol=symbol,days=days,**summary(rs,'net_return'),formed_n=len(rs),breach_n=sum(r['breach_time'] is not None for r in rs),median_funding=median([r['funding_income']/r['initial_capital'] for r in rs]),median_cost=median([r['total_cost']/r['initial_capital'] for r in rs]),break_even_n=sum(r['first_daily_break_even'] is not None for r in rs),median_first_break_even=median([r['first_daily_break_even'] for r in rs if r['first_daily_break_even'] is not None]),break_even_then_loss_n=sum(r['first_break_even_then_loss'] for r in rs))) common=dict(version=VERSION,coverage=coverage,candidates=candidates,status_counts=dict(Counter(r['status'] for r in candidates))) return common|dict(observations=lev,summary=ls),common|dict(observations=arb,summary=ars) def run(inputs,out): out.mkdir(parents=True,exist_ok=True) raw=json.loads(gzip.decompress((inputs/'spot.json.gz').read_bytes()));series={s:daily(d) for s,d in raw['series'].items()} deriv=json.loads((inputs/'derivatives.json').read_text())['series'] products={'allocation':allocation(series),'rotation':rotation(series,raw['universe'])} products['leverage'],products['arbitrage']=derivative_studies(series,deriv) for name,data in products.items(): write(out,name,data);csvout(out,name+'-observations',data['observations']);csvout(out,name+'-summary',data['summary']);csvout(out,name+'-candidates',data['candidates']) print(name,len(data['observations']),'observations',flush=True) if __name__=='__main__':run(Path(sys.argv[1]),Path(sys.argv[2]))