"""Standalone retrospective daily-bar experiments. Python 3.10+, stdlib only. Run: python decision_studies.py candle-inputs.json research.json Prices are next-open execution proxies, not guaranteed fills. No intraday ordering is inferred. All outcomes are independent observations, not portfolio returns. """ import calendar import json import random import statistics as st import sys from collections import defaultdict from datetime import datetime, timezone DAY = 86400 START = 1601510400 # 2020-10-01 UTC, after a 30-day warmup SEED = 18092026 VERSION = 'daily-entry-exit-1.0' def iso(t): return datetime.fromtimestamp(t, timezone.utc).isoformat().replace('+00:00', 'Z') def quantile(xs, q): a = sorted(xs) if not a: return None p = (len(a)-1)*q i = int(p) return a[i] + (a[min(i+1, len(a)-1)]-a[i])*(p-i) def quarter(t): d = datetime.fromtimestamp(t, timezone.utc) return f'{d.year}-Q{(d.month-1)//3+1}' def period(t): y = datetime.fromtimestamp(t, timezone.utc).year return '2020–2022' if y <= 2022 else '2023–2024' if y <= 2024 else '2025–cutoff' def cluster_ci(rows, key): 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) values = [] for _ in range(1000): sample = [x for _ in names for x in groups[names[rng.randrange(len(names))]]] values.append(st.mean(sample)) return [quantile(values, .025), quantile(values, .975)] def summary(rows, strategies): n = len(rows) result = {} for name in strategies: vals = [r[name]['return'] for r in rows] paired = [dict(event_time=r['event_time'], delta=r[name]['return']-r['immediate']['return']) for r in rows] entered = [r for r in rows if r[name]['entered']] result[name] = dict(n=n, entered_n=len(entered), mean=st.mean(vals) if n else None, median=st.median(vals) if n else None, p10=quantile(vals,.1),p90=quantile(vals,.9),worst=min(vals) if n else None, best=max(vals) if n else None, trimmed_mean_10pct=st.mean(sorted(vals)[int(n*.1):n-int(n*.1)]) if n else None, positive_n=sum(x > 0 for x in vals), negative_n=sum(x < 0 for x in vals), cash_n=n-len(entered), filled_median=st.median(r[name]['return'] for r in entered) if entered else None, missed_profitable_immediate_n=sum(not r[name]['entered'] and r['immediate']['return'] > 0 for r in rows), mean_delta=st.mean(r['delta'] for r in paired) if n else None, median_delta=st.median(r['delta'] for r in paired) if n else None, beats_immediate_n=sum(r['delta']>0 for r in paired), mean_delta_ci95=cluster_ci(paired, 'delta'), mean_peak_giveback=st.mean(r[name].get('peak_giveback', 0) for r in rows) if n else None, quarter_clusters=len({quarter(r['event_time']) for r in rows}), status='descriptive' if n >= 20 else 'exploratory') return result def ready(data, start, end): return all(t in data for t in range(start, end+1, DAY)) def pump_outcome(data, t, fee): exit_t = t+31*DAY out = {} for name in ['immediate', 'pullback', 'confirmation']: entry = t+DAY if name == 'immediate' else None if entry is None: for i in range(1, 8): bar = data[t+i*DAY] hit = bar['c'] <= data[t]['c']*.95 if name == 'pullback' else bar['c'] > data[t]['h'] if hit: entry = t+(i+1)*DAY break px = data[entry]['o'] if entry else None proceeds = data[exit_t]['o']*(1-fee)/(px*(1+fee)) if entry else 1.0 out[name] = dict(entered=entry is not None, entry_time=entry, entry_price=px, exit_time=exit_t if entry else None, exit_price=data[exit_t]['o'] if entry else None, return_=proceeds-1) out[name]['return'] = out[name].pop('return_') return out def pumps(series, cutoff, threshold=.10, fee=.002): ledger, rows = [], [] btc = series['BTCUSDT'] for symbol, data in sorted(series.items()): last = None for t in sorted(data): if t < START or t-DAY not in data: continue change = data[t]['c']/data[t-DAY]['c']-1 if change < threshold: continue row = dict(id=f'pump-{threshold:g}-{symbol}-{t}', symbol=symbol, event_time=t+DAY, signal_day=iso(t)[:10], return1=change, signal_close=data[t]['c'], signal_high=data[t]['h'], exit_time=t+31*DAY) if last is not None and t-last < 30*DAY: row.update(status='excluded_cooldown', blocked_by=last) else: last = t if t+31*DAY >= cutoff: row['status'] = 'pending' elif not ready(data, t, t+31*DAY): row['status'] = 'missing_path' else: row['status'] = 'complete' prev = [data[t-i*DAY] for i in range(1, 21) if t-i*DAY in data] bprev = [btc[t-i*DAY]['c'] for i in range(30) if t-i*DAY in btc] row.update(volume_ratio=data[t]['qv']/st.median(b['qv'] for b in prev) if len(prev)==20 else None, extension_sma20=data[t]['c']/st.mean(b['c'] for b in prev)-1 if len(prev)==20 else None, btc_above_sma30=btc[t]['c']>=st.mean(bprev) if len(bprev)==30 else None, **pump_outcome(data, t, fee)) rows.append(row) ledger.append(row) rows.sort(key=lambda r:(r['event_time'],r['symbol'])) return ledger, rows def exit_outcome(data, entry_t, milestone, fee): end = milestone+31*DAY units = 1/(data[entry_t]['o']*(1+fee)) trail = end high = data[milestone]['c'] for t in range(milestone+DAY, end, DAY): high = max(high, data[t]['c']) if data[t]['c'] <= high*.9: trail = t+DAY break trades = {'immediate':[(milestone+DAY,1.)], 'staged':[(milestone+DAY,.5),(end,.5)], 'trailing':[(trail,1.)], 'hold':[(end,1.)]} out = {} for name, sells in trades.items(): proceeds = sum(units*w*data[t]['o']*(1-fee) for t,w in sells) # Mark net liquidation value at each daily close. No fabricated intraday high. marks = [units*data[milestone]['c']*(1-fee)] for t in range(milestone+DAY, end, DAY): cash = sum(units*w*data[e]['o']*(1-fee) for e,w in sells if e<=t) remaining = 1-sum(w for e,w in sells if e<=t) marks.append(cash+units*remaining*data[t]['c']*(1-fee)) out[name] = dict(entered=True, entry_time=entry_t, entry_price=data[entry_t]['o'], return_=proceeds-1, final_proceeds=proceeds, sales=[dict(time=t,weight=w,price=data[t]['o']) for t,w in sells], peak_giveback=max(0,max(marks)-proceeds)) out[name]['return'] = out[name].pop('return_') return out def profits(series, cutoff, threshold=.20, fee=.002): ledger, rows = [], [] for symbol, data in sorted(series.items()): for start in sorted(data): if start < START or datetime.fromtimestamp(start,timezone.utc).day != 1: continue row = dict(id=f'profit-{threshold:g}-{symbol}-{start}', symbol=symbol, entry_time=start, entry_price=data[start]['o'], milestone=None, status=None) for t in range(start, start+90*DAY, DAY): if t>=cutoff: row['status']='pending_milestone'; break if t not in data: row['status']='missing_milestone_path'; break if data[t]['c']/(data[start]['o']*(1+fee))-1 >= threshold: row.update(milestone=t,event_time=t+DAY, milestone_close=data[t]['c'], mark_return=data[t]['c']/(data[start]['o']*(1+fee))-1) if t+31*DAY>=cutoff: row['status']='pending_exit' elif not ready(data,t,t+31*DAY): row['status']='missing_exit_path' else: row.update(status='complete',**exit_outcome(data,start,t,fee));rows.append(row) break if row['status'] is None: row['status']='not_reached_within_90d' ledger.append(row) rows.sort(key=lambda r:(r['event_time'],r['symbol'],r['entry_time'])) return ledger, rows def grouped(rows, strategies): groups = defaultdict(list) for r in rows: groups['symbol:'+r['symbol']].append(r) groups['period:'+period(r['event_time'])].append(r) if 'btc_above_sma30' in r: groups['btc_trend:'+str(r['btc_above_sma30'])].append(r) groups['volume:'+('>=2x' if r['volume_ratio'] is not None and r['volume_ratio']>=2 else '<2x_or_missing')].append(r) # Small descriptive groups do not get deceptively precise intervals. return {k:summary(v,strategies) for k,v in groups.items()} def run(raw): series={s:{int(t):v for t,v in rows.items()} for s,rows in raw['series'].items()} cutoff=raw['cutoff'];result=dict(version=VERSION,cutoff=iso(cutoff),input_observed_at=raw['observed_at'], input_start=min(iso(t) for data in series.values() for t in data), input_end=max(iso(t+DAY) for data in series.values() for t in data), symbols=sorted(series),primary_fee_per_side=.002, ci_contract=dict(method='UTC-calendar-quarter cluster percentile bootstrap of paired mean return differences', iterations=1000,seed=SEED,rng='Python random.Random MT19937; sorted quarter keys; randrange', quantiles='linear interpolation at (n-1)*q',minimum_events=20,minimum_quarters=8),studies={}) for name,fn,threshold,strategies in [('pump',pumps,.10,['immediate','pullback','confirmation']), ('profit',profits,.20,['immediate','staged','trailing','hold'])]: ledger,rows=fn(series,cutoff,threshold) sensitivity={} for level in ([.20] if name=='pump' else [.10,.30]): ll,rr=fn(series,cutoff,level) sensitivity[str(level)]=dict(candidates=len(ll),ledger=ll,summary=summary(rr,strategies),observations=rr) cost={} for fee in [.001,.003,.005]: # Freeze the primary milestone cohort when comparing exit costs. # Otherwise changing costs also changes who/when qualified for exit. rr=[] for r in rows: x=dict(r) outcome=pump_outcome(series[r['symbol']],r['event_time']-DAY,fee) if name=='pump' else exit_outcome(series[r['symbol']],r['entry_time'],r['milestone'],fee) x.update(outcome);rr.append(x) cost[str(fee)]=summary(rr,strategies) result['studies'][name]=dict(ledger=ledger,observations=rows,summary=summary(rows,strategies), groups=grouped(rows,strategies),threshold_sensitivity=sensitivity,cost_sensitivity=cost, unique_symbol_decision_dates=len({(r['symbol'],r['event_time']) for r in rows}), statuses={s:sum(r['status']==s for r in ledger) for s in sorted({r['status'] for r in ledger})}) return result if __name__=='__main__': with open(sys.argv[1]) as f: raw=json.load(f) result=run(raw) with open(sys.argv[2],'w') as f: json.dump(result,f,ensure_ascii=False,indent=2,allow_nan=False)