"""Frozen spot stop-loss experiment. Standard library only. Reproduce: python stop_loss_study.py INPUT_DIRECTORY OUTPUT.json Input directory contains BTCUSDT.csv, ETHUSDT.csv, SOLUSDT.csv, manifest.json. No network, optimizers, synthetic bars, leverage or executed-fill claims. """ import csv import json import random import statistics import sys from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path VERSION = 'spot-stop-paths-1.0' SYMBOLS = ('BTCUSDT', 'ETHUSDT', 'SOLUSDT') POLICIES = ('fixed_1', 'fixed_2', 'atr_2', 'structure') COST = .0015 SEED = 20260919 def iso(t): return datetime.fromtimestamp(t, timezone.utc).isoformat().replace('+00:00', 'Z') def quantile(xs, p): xs = sorted(xs) if not xs: return None z = (len(xs)-1)*p i = int(z) return xs[i] + (xs[min(i+1, len(xs)-1)]-xs[i])*(z-i) def load(directory, symbol): with (Path(directory)/(symbol+'.csv')).open() as f: return {int(r['time']): {k: float(v) for k, v in r.items() if k != 'time'} for r in csv.DictReader(f)} def hourly(bars): result = {} for t in sorted(bars): if t % 3600: continue chunk = [bars.get(t+j*300) for j in range(12)] if any(x is None for x in chunk): continue result[t] = dict(o=chunk[0]['o'], c=chunk[-1]['c'], h=max(x['h'] for x in chunk), l=min(x['l'] for x in chunk)) return result def net(entry, exit_price, cost=COST): return exit_price/entry*(1-cost)/(1+cost)-1 def followup(bars, trigger, entry, stop): times = [trigger+j*300 for j in range(1, 289)] recover = next((t for t in times if bars[t]['c'] >= entry), None) adverse_level = stop-(entry-stop) adverse = next((t for t in times if bars[t]['l'] <= adverse_level), None) if recover is not None and adverse is not None: category = 'both' order = 'same_bar_unknown' if recover == adverse else ('recovery_first' if recover < adverse else 'adverse_first') elif recover is not None: category, order = 'recovery_only', 'recovery_only' elif adverse is not None: category, order = 'adverse_only', 'adverse_only' else: category, order = 'neither', 'neither' return dict(recovered_1h=recover is not None and recover <= trigger+3600, recovered_4h=recover is not None and recover <= trigger+14400, recovered_24h=recover is not None, adverse_24h=adverse is not None, recovery_time=iso(recover+300) if recover is not None else None, recovery_hours=(recover+300-(trigger+300))/3600 if recover is not None else None, adverse_bar_time=iso(adverse) if adverse is not None else None, adverse_level=adverse_level, path_category=category, first_order=order, post_trigger_low=min(bars[t]['l'] for t in times), trigger_bar_recovery_ambiguous=bars[trigger]['h'] >= entry) def simulate(bars, entry_time, stop, cost=COST): entry = bars[entry_time]['o'] trigger = next((t for t in range(entry_time, entry_time+86400, 300) if bars[t]['l'] <= stop), None) fill = min(bars[trigger]['o'], stop) if trigger is not None else bars[entry_time+86400]['o'] planned_loss = -net(entry, stop, cost) weight = min(1., .01/planned_loss) r = dict(stop_price=stop, stop_distance=1-stop/entry, triggered=trigger is not None, trigger_time=iso(trigger) if trigger is not None else None, exit_price=fill, exit_time=iso(trigger if trigger is not None else entry_time+86400), net_return=net(entry, fill, cost), allocation=weight, risk_budget_return=weight*net(entry, fill, cost), modeled_stop_loss=planned_loss, gap_below_stop=trigger is not None and bars[trigger]['o'] < stop) if trigger is not None: r.update(followup(bars, trigger, entry, stop)) return r def candidates(bars, symbol, rule, start, end): hours = hourly(bars) result = [] next_allowed = start for t in range(start, end, 3600): signal = t-3600 prior = [hours.get(signal-j*3600) for j in range(1, 21)] h = hours.get(signal) if h is None or any(x is None for x in prior): continue if rule == 'breakout': qualifies = h['c'] > max(x['h'] for x in prior) else: dt = datetime.fromtimestamp(t, timezone.utc) qualifies = dt.weekday() == 0 and dt.hour == 0 if not qualifies: continue atr = statistics.mean(max(hours[signal-j*3600]['h']-hours[signal-j*3600]['l'], abs(hours[signal-j*3600]['h']-hours[signal-(j+1)*3600]['c']), abs(hours[signal-j*3600]['l']-hours[signal-(j+1)*3600]['c'])) for j in range(14)) row = dict(id=f'{rule}:{symbol}:{t}', symbol=symbol, rule=rule, entry_time=t, entry_utc=iso(t), decision_time=iso(t), atr14=atr, atr_fraction=atr/h['c'], trend='above_sma20' if h['c'] > statistics.mean([h['c']]+[x['c'] for x in prior[:19]]) else 'at_or_below_sma20', volatility='high' if atr/h['c'] >= .0075 else 'ordinary', period='first_half' if t < (start+end)/2 else 'second_half', signal_close=h['c'], preceding_high=max(x['h'] for x in prior), preceding_low=min(x['l'] for x in prior)) if t < next_allowed: row['status'] = 'cooldown' else: next_allowed = t+172800 if t+172800 >= end: row['status'] = 'pending' elif any(j not in bars for j in range(t, t+172800+300, 300)): row['status'] = 'missing_path' else: entry = bars[t]['o'] stops = dict(fixed_1=entry*.99, fixed_2=entry*.98, atr_2=entry-2*atr, structure=row['preceding_low']-.25*atr) row.update(entry_price=entry, stops=stops) if any(not .0025 <= 1-p/entry <= .15 for p in stops.values()): row['status'] = 'invalid_common_distance' else: row['status'] = 'complete' row['hold_exit_price'] = bars[t+86400]['o'] row['hold_return'] = net(entry, row['hold_exit_price']) row['policies'] = {k: simulate(bars, t, p) for k,p in stops.items()} result.append(row) return result def cluster_ci(rows, value): if len(rows) < 20: return None clusters = defaultdict(list) for row in rows: dt = datetime.fromtimestamp(row['entry_time'], timezone.utc).isocalendar() clusters[f'{dt.year}-{dt.week}'].append(value(row)) if len(clusters) < 8: return None sums = [(sum(x), len(x)) for x in clusters.values()] rng = random.Random(SEED) estimates = [] for _ in range(1000): sample = [rng.choice(sums) for _ in sums] estimates.append(sum(x[0] for x in sample)/sum(x[1] for x in sample)) return [quantile(estimates, .025), quantile(estimates, .975)] def summarize(rows, intervals=False, cost=COST): result = {} for policy in POLICIES: pp = [r['policies'][policy] for r in rows] triggered = [p for p in pp if p['triggered']] returns = [net(r['entry_price'], r['policies'][policy]['exit_price'], cost) for r in rows] risks = [min(1, .01/-net(r['entry_price'], r['policies'][policy]['stop_price'], cost))*ret for r,ret in zip(rows,returns)] diffs = [ret-net(r['entry_price'], r['hold_exit_price'], cost) for r,ret in zip(rows,returns)] result[policy] = dict(n=len(rows), stopped_n=len(triggered), recovered_1h_n=sum(p['recovered_1h'] for p in triggered), recovered_4h_n=sum(p['recovered_4h'] for p in triggered), recovered_24h_n=sum(p['recovered_24h'] for p in triggered), adverse_24h_n=sum(p['adverse_24h'] for p in triggered), paths=dict(Counter(p['path_category'] for p in triggered)), trigger_bar_ambiguity_n=sum(p['trigger_bar_recovery_ambiguous'] for p in triggered), mean=statistics.mean(returns) if returns else None, median=quantile(returns,.5), p10=quantile(returns,.1), risk_mean=statistics.mean(risks) if risks else None, risk_p10=quantile(risks,.1), mean_difference=statistics.mean(diffs) if diffs else None, positive_n=sum(r>0 for r in returns), mean_difference_ci95=cluster_ci(rows,lambda r:net(r['entry_price'],r['policies'][policy]['exit_price'],cost)-net(r['entry_price'],r['hold_exit_price'],cost)) if intervals else None, recovery_ci95=cluster_ci([r for r in rows if r['policies'][policy]['triggered']],lambda r: int(r['policies'][policy]['recovered_24h'])) if intervals else None, median_recovery_hours=quantile([p['recovery_hours'] for p in triggered if p['recovered_24h']],.5)) return result def run(directory): manifest = json.loads((Path(directory)/'manifest.json').read_text()) all_rows=[] for symbol in SYMBOLS: bars=load(directory,symbol) for rule in ('breakout','weekly'): all_rows.extend(candidates(bars,symbol,rule,manifest['start'],manifest['end_exclusive'])) result=dict(version=VERSION, input_start=iso(manifest['start']), input_end_exclusive=iso(manifest['end_exclusive']), input_observed_at=manifest['captured_at'], cost_per_side=COST, seed=SEED, candidates=all_rows, rules={}) for rule in ('breakout','weekly'): rows=[r for r in all_rows if r['rule']==rule and r['status']=='complete'] result['rules'][rule]=dict(n=len(rows),status_counts=dict(Counter(r['status'] for r in all_rows if r['rule']==rule)), summary=summarize(rows,True), hold_mean=statistics.mean(r['hold_return'] for r in rows), hold_median=quantile([r['hold_return'] for r in rows],.5), by_symbol={s:summarize([r for r in rows if r['symbol']==s]) for s in SYMBOLS}, groups={k:summarize([r for r in rows if r[field]==k]) for field,keys in [('volatility',['high','ordinary']),('period',['first_half','second_half']),('trend',['above_sma20','at_or_below_sma20'])] for k in keys}, cost_sensitivity={str(c):summarize(rows,cost=c) for c in (.0005,.003)}) return result if __name__ == '__main__': result=run(sys.argv[1]) Path(sys.argv[2]).write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n') print(json.dumps({k:{'n':v['n'],'summary':v['summary']} for k,v in result['rules'].items()},indent=2))