"""Reproduce: python altcoin_catchup_study.py candle-inputs.json research.json. Retrospective Binance spot study. Python standard library only. """ import json import random import statistics as st import sys from collections import defaultdict,Counter from datetime import datetime,timezone DAY=86400 VERSION='altcoin-catchup-1.0' SEED=18092026 def iso(t):return datetime.fromtimestamp(t,timezone.utc).isoformat().replace('+00:00','Z') def quarter(t): d=datetime.fromtimestamp(t,timezone.utc);return f'{d.year}-Q{(d.month-1)//3+1}' def quantile(xs,p): a=sorted(xs) if not a:return None u=(len(a)-1)*p;i=int(u);return a[i]+(a[min(i+1,len(a)-1)]-a[i])*(u-i) def interval(rows,key): # Rows here are one observation per retained BTC event, not per coin. groups=defaultdict(list) for r in rows:groups[quarter(r['event_time'])].append(r[key]) if len(rows)<20 or len(groups)<8:return None names=sorted(groups);rng=random.Random(SEED);rep=[] for _ in range(1000): v=[x for _ in names for x in groups[names[rng.randrange(len(names))]]];rep.append(st.mean(v)) return [quantile(rep,.025),quantile(rep,.975)] def contiguous(data,start,end):return all(t in data for t in range(start,end+DAY,DAY)) def features(data,btc,t): closes=[data[t-i*DAY]['c'] for i in range(20)] changes=[data[t-i*DAY]['c']-data[t-(i+1)*DAY]['c'] for i in range(14)] up=st.mean(max(x,0) for x in changes);down=st.mean(max(-x,0) for x in changes) rsi=50. if up+down==0 else 100*up/(up+down) ratio=(data[t]['c']/btc[t]['c'])/(data[t-3*DAY]['c']/btc[t-3*DAY]['c'])-1 v=data[t]['qv']/st.median(data[t-i*DAY]['qv'] for i in range(1,21)) sma=st.mean(closes) return dict(sma20=sma,above_sma20=data[t]['c']>sma,cutler_rsi14=rsi, volume_ratio20=v,relative_momentum3=ratio, combined_confirmation=data[t]['c']>sma and v>=1.5 and ratio>0 and 50<=rsi<=70) def outcomes(data,btc,t,fee): entry=t+DAY;baseline=data[t-7*DAY]['c']/btc[t-7*DAY]['c'];out={} for h in [7,14,30]: end=entry+h*DAY alt=data[end]['o']*(1-fee)/(data[entry]['o']*(1+fee))-1 br=btc[end]['o']*(1-fee)/(btc[entry]['o']*(1+fee))-1 q=data[end]['o']/btc[end]['o'];recovered=q>=baseline touches=[u for u in range(entry,end,DAY) if data[u]['c']/btc[u]['c']>=baseline] out[str(h)]=dict(entry_time=entry,exit_time=end,alt_entry=data[entry]['o'],btc_entry=btc[entry]['o'], alt_exit=data[end]['o'],btc_exit=btc[end]['o'],alt_net=alt,btc_net=br,excess_net=alt-br, reference_ratio=baseline,exit_ratio=q,relative_gap_remaining=q/baseline-1, positive=alt>0,outperformed=alt>br,recovered=recovered,positive_and_recovered=alt>0 and recovered, touched_recovery=bool(touches),first_recovery_close_day=touches[0] if touches else None) return out def summarize(rows,with_ci=False): ans={};n=len(rows) for h in ['7','14','30']: by=defaultdict(list) for r in rows:by[r['event_time']].append(r['outcomes'][h]) eventrows=[] for t,rr in sorted(by.items()): eventrows.append(dict(event_time=t,alt_net=st.mean(x['alt_net'] for x in rr), excess_net=st.mean(x['excess_net'] for x in rr),**{k:st.mean(float(x[k]) for x in rr) for k in ['positive','outperformed','recovered','positive_and_recovered']})) values=[r['outcomes'][h]['alt_net'] for r in rows];excess=[r['outcomes'][h]['excess_net'] for r in rows] ans[h]=dict(n=n,btc_events=len(by),quarters=len({quarter(t) for t in by}), median_net=st.median(values) if n else None,mean_net=st.mean(values) if n else None, median_excess=st.median(excess) if n else None, counts={k:sum(r['outcomes'][h][k] for r in rows) for k in ['positive','outperformed','recovered','positive_and_recovered','touched_recovery']}, event_balanced={k:st.mean(x[k] for x in eventrows) if eventrows else None for k in ['alt_net','excess_net','positive','outperformed','recovered','positive_and_recovered']}, event_balanced_ci95={k:interval(eventrows,k) if with_ci else None for k in ['alt_net','excess_net','positive','outperformed','recovered','positive_and_recovered']}, status='exploratory' if n<20 or len(by)<20 else 'descriptive') return ans def experiment(series,cutoff,btc_threshold=.10,lag_pp=.05,fee=.002): btc=series['BTCUSDT'];last=None;events=[];ledger=[] for t in sorted(btc): if not contiguous(btc,t-7*DAY,t):continue br=btc[t]['c']/btc[t-7*DAY]['c']-1 if br=lag_pp, median_quote30=st.median(data[t-i*DAY]['qv'] for i in range(30)),**features(data,btc,t)) if t+31*DAY>=cutoff:r['status']='pending_30d' elif not contiguous(data,t+DAY,t+31*DAY) or not contiguous(btc,t+DAY,t+31*DAY):r['status']='missing_future_path' else:r.update(status='complete',outcomes=outcomes(data,btc,t,fee)) ledger.append(r) complete=[r for r in ledger if r['status']=='complete'];lags=[r for r in complete if r['lagger']];others=[r for r in complete if not r['lagger']] baskets=[] for event in events: if event['status']!='retained':continue a=[r for r in lags if r['btc_event']==event['id']];b=[r for r in others if r['btc_event']==event['id']] x=dict(btc_event=event['id'],event_time=event['known_at'],lagger_n=len(a),comparison_n=len(b)) if not a or not b:x['status']='no_complete_laggers' if not a else 'no_complete_comparison' else: x.update(status='paired',outcomes={}) for h in ['7','14','30']: la=st.mean(r['outcomes'][h]['alt_net'] for r in a);co=st.mean(r['outcomes'][h]['alt_net'] for r in b) x['outcomes'][h]=dict(lagger_mean=la,comparison_mean=co,difference=la-co,btc_net=a[0]['outcomes'][h]['btc_net']) baskets.append(x) paired=[x for x in baskets if x['status']=='paired'];pairsum={} for h in ['7','14','30']: rr=[dict(event_time=r['event_time'],delta=r['outcomes'][h]['difference']) for r in paired] pairsum[h]=dict(n=len(rr),mean_difference=st.mean(x['delta'] for x in rr) if rr else None, median_difference=st.median(x['delta'] for x in rr) if rr else None, positive_n=sum(x['delta']>0 for x in rr),ci95=interval(rr,'delta')) groups=defaultdict(list) for r in lags: y=datetime.fromtimestamp(r['event_time'],timezone.utc).year for key in ['symbol:'+r['symbol'],'period:'+('2020–2022' if y<=2022 else '2023–2024' if y<=2024 else '2025–cutoff'), 'sma20:'+str(r['above_sma20']),'volume:'+str(r['volume_ratio20']>=1.5), 'relative3:'+str(r['relative_momentum3']>0),'rsi50to70:'+str(50<=r['cutler_rsi14']<=70), 'combined:'+str(r['combined_confirmation']), 'future_btc:'+('up' if r['outcomes']['30']['btc_net']>0 else 'not_up')]:groups[key].append(r) return dict(parameters=dict(btc_return7_min=btc_threshold,lag_pp_min=lag_pp,fee_per_side=fee),events=events, ledger=ledger,baskets=baskets,paired_baskets=pairsum, status_counts=dict(Counter(r['status'] for r in ledger)), laggers=summarize(lags,True),comparison=summarize(others),groups={k:summarize(v) for k,v in groups.items()}) def run(raw): series={s:{int(t):r for t,r in d.items()} for s,d in raw['series'].items()} main=experiment(series,raw['cutoff']);sensitivity={} for name,b,l in [('btc5',.05,.05),('btc15',.15,.05),('lag2.5',.10,.025),('lag10',.10,.10)]: sensitivity[name]=experiment(series,raw['cutoff'],b,l) lags=[r for r in main['ledger'] if r.get('lagger') and r['status']=='complete'];cost={} for fee in [.001,.003,.005]: rows=[dict(r,outcomes=outcomes(series[r['symbol']],series['BTCUSDT'],r['signal_time'],fee)) for r in lags] cost[str(fee)]=summarize(rows) return dict(version=VERSION,cutoff=iso(raw['cutoff']),observed_at=raw['observed_at'], symbols=sorted(series),input_rows=sum(len(d) for d in series.values()), temporal_coverage=min(iso(t) for d in series.values() for t in d)+'/'+iso(raw['cutoff']), main=main,threshold_sensitivity=sensitivity,cost_sensitivity=cost) if __name__=='__main__': with open(sys.argv[1]) as f:raw=json.load(f) with open(sys.argv[2],'w') as f:json.dump(run(raw),f,indent=2,ensure_ascii=False,allow_nan=False)