"""Frozen breakeven-stop comparison. Standard library; no network or orders. Reproduce after downloading the prior stop-loss research.json and candle-inputs.zip: python breakeven_stop_study.py PRIOR_RESEARCH_JSON PRIOR_CANDLE_ZIP OUTPUT_JSON """ import csv import hashlib import io import json import random import statistics import sys import zipfile from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path VERSION = "breakeven-stop-paths-1.0" COST = .0015 SEED = 20260925 POLICIES = ("never_move", "move_after_half_r", "move_after_one_r", "move_after_two_r") THRESHOLDS = {"never_move": None, "move_after_half_r": .01, "move_after_one_r": .02, "move_after_two_r": .04} def quantile(values, fraction): values = sorted(values) if not values: return None at = (len(values)-1)*fraction lower = int(at) return values[lower] + (values[min(lower+1, len(values)-1)]-values[lower])*(at-lower) def iso(ts): return datetime.fromtimestamp(ts, timezone.utc).isoformat().replace("+00:00", "Z") def load_inputs(candidates_path, candles_zip): original = json.loads(Path(candidates_path).read_text()) if original.get("version") != "spot-stop-paths-1.0": raise ValueError("Wrong source edition") cases = [row for row in original["candidates"] if row["rule"] == "breakout" and row["status"] == "complete"] if len(cases) != 313: raise ValueError(f"Expected frozen 313-case source cohort, found {len(cases)}") bars = {} with zipfile.ZipFile(candles_zip) as z: for symbol in ("BTCUSDT", "ETHUSDT", "SOLUSDT"): with z.open(f"{symbol}.csv") as raw: reader = csv.DictReader(io.TextIOWrapper(raw, encoding="utf-8")) bars[symbol] = {int(row["time"]): {key: float(value) for key, value in row.items() if key != "time"} for row in reader} return cases, bars def run_case(case, candles, policy): entry_time = case["entry_time"] entry = case["entry_price"] if entry != candles[entry_time]["o"]: raise ValueError("Published entry and frozen candle disagree") stop_initial = entry*.98 stop_breakeven = entry*(1+COST)/(1-COST) target = entry*1.06 threshold = THRESHOLDS[policy] active_stop = stop_initial armed_at = None for step in range(288): stamp = entry_time + step*300 bar = candles.get(stamp) if bar is None: raise ValueError(f"Incomplete 24h path: {case['id']} at {stamp}") if bar["l"] <= active_stop: fill = min(bar["o"], active_stop) reason = "net_breakeven_stop" if armed_at is not None else "initial_stop" break if bar["h"] >= target: fill = target reason = "target" break if threshold is not None and armed_at is None and bar["h"] >= entry*(1+threshold): armed_at = stamp active_stop = stop_breakeven # next bar only; current stop check already occurred. else: stamp = entry_time + 86400 if stamp not in candles: raise ValueError(f"Missing 24h exit open: {case['id']}") fill = candles[stamp]["o"] reason = "time_exit" net = fill/entry*(1-COST)/(1+COST)-1 return {"observation_id": case["id"], "symbol": case["symbol"], "entry_time": entry_time, "entry_utc": case["entry_utc"], "entry_price": entry, "policy": policy, "activation_time": armed_at, "exit_time": stamp, "exit_utc": iso(stamp), "exit_reason": reason, "exit_price": fill, "net_return": net, "gap_below_active_stop": reason.endswith("stop") and fill < active_stop, "active_stop_at_exit": active_stop if reason.endswith("stop") else None, "volatility": case["volatility"], "period": case["period"]} def cluster_interval(rows, policy): selected = [row for row in rows if row["policy"] == policy] by_case = defaultdict(dict) for row in rows: by_case[row["observation_id"]][row["policy"]] = row clusters = defaultdict(list) for row in selected: dt = datetime.fromtimestamp(row["entry_time"], timezone.utc).isocalendar() key = f"{dt.year}-{dt.week}" clusters[key].append(row["net_return"]-by_case[row["observation_id"]]["never_move"]["net_return"]) if len(selected) < 20 or len(clusters) < 8: return None rng = random.Random(SEED) chunks = list(clusters.values()) sample_means = [] for _ in range(1000): picked = [rng.choice(chunks) for _ in chunks] sample_means.append(sum(sum(chunk) for chunk in picked)/sum(len(chunk) for chunk in picked)) return [quantile(sample_means, .025), quantile(sample_means, .975)] def summarize(all_rows, group=None): subset = [r for r in all_rows if group is None or group(r)] base = {r["observation_id"]: r for r in subset if r["policy"] == "never_move"} result = {} for policy in POLICIES: rows = [r for r in subset if r["policy"] == policy] if not rows: result[policy] = {"n": 0, "mean_net_return": None} continue values = [r["net_return"] for r in rows] paired = [r["net_return"]-base[r["observation_id"]]["net_return"] for r in rows] result[policy] = { "n": len(rows), "activated_n": sum(r["activation_time"] is not None for r in rows), "breakeven_exit_n": sum(r["exit_reason"] == "net_breakeven_stop" for r in rows), "net_breakeven_gap_loss_n": sum(r["exit_reason"] == "net_breakeven_stop" and r["net_return"] < -1e-12 for r in rows), "base_target_but_breakeven_exit_n": sum(r["exit_reason"] == "net_breakeven_stop" and base[r["observation_id"]]["exit_reason"] == "target" for r in rows), "positive_n": sum(x > 0 for x in values), "mean_net_return": statistics.mean(values), "median_net_return": quantile(values, .5), "p10_net_return": quantile(values, .1), "mean_paired_diff_vs_never_move": statistics.mean(paired), "paired_diff_ci95": cluster_interval(subset, policy) if policy != "never_move" else None, "exit_reasons": dict(Counter(r["exit_reason"] for r in rows)), } return result def study(candidates_path, candles_zip): cases, candles = load_inputs(candidates_path, candles_zip) observations = [run_case(case, candles[case["symbol"]], policy) for case in cases for policy in POLICIES] return { "version": VERSION, "source_version": "spot-stop-paths-1.0", "source_research_sha256": hashlib.sha256(Path(candidates_path).read_bytes()).hexdigest(), "source_candle_zip_sha256": hashlib.sha256(Path(candles_zip).read_bytes()).hexdigest(), "source_window": ["2025-09-01T00:00:00Z", "2026-09-19T00:00:00Z"], "source_market": "Binance BTCUSDT ETHUSDT SOLUSDT spot; retrospectively recovered 5m candles", "candidate_n": len(cases), "policy_observation_n": len(observations), "assumptions": {"initial_stop_pct": 2, "target_pct": 6, "maximum_hold_hours": 24, "cost_per_side": COST, "activation_after_r": {"move_after_half_r": .5, "move_after_one_r": 1, "move_after_two_r": 2}, "activation_effective": "next complete 5m bar", "same_bar_touch": "stop first", "break_even_exit_multiplier": (1+COST)/(1-COST)}, "summary": summarize(observations), "by_symbol": {symbol: summarize(observations, lambda r, symbol=symbol: r["symbol"] == symbol) for symbol in ("BTCUSDT", "ETHUSDT", "SOLUSDT")}, "by_half": {half: summarize(observations, lambda r, half=half: r["period"] == half) for half in ("first_half", "second_half")}, "by_volatility": {band: summarize(observations, lambda r, band=band: r["volatility"] == band) for band in ("ordinary", "high")}, "observations": observations, "limitations": ["Historical spot OHLC cannot prove tick-order or actual stop fills.", "Net breakeven can still lose on a gap below the active stop.", "The 313 breakout entries are selected, correlated and not a universal entry strategy.", "A one-year cohort does not establish a future best breakeven rule."], } def main(): result = study(sys.argv[1], sys.argv[2]) output = Path(sys.argv[3]) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False)+"\n") print(json.dumps({"candidate_n": result["candidate_n"], "summary": result["summary"]}, indent=2)) if __name__ == "__main__": main()