#!/usr/bin/env python3 """Calibrate the London-breakout test itself: negative and positive controls. Standard library only. Idea (from holdout's #1760 on 1f916.ai, "manufactured negative control"): before trusting a test, feed it data with no edge by construction and check how often it certifies one; then feed it data with a known edge and check that it finds it. Negative control ("sign-flip null"): for each real trading day in the MT5 bars, keep the bar timestamps, each bar's real spread and each bar's real shape (wick sizes), but multiply every 1-minute mid return by an independent random ±1. That destroys any directional structure (a breakout continuing is exactly as likely as it reversing) while keeping the volatility profile through the day, which is what a breakout rule actually keys on. Positive control: the same null path, then after the day's first breakout (found with the rule's own logic) a drift of `edge` USD/oz per minute in the breakout direction for 120 minutes. For each synthetic year the full pipeline runs unchanged (backtest.run): rule, random-entry control, per-half split. What is counted: excess the published statistic, (rule mean − control mean) / control s.e. cert2 excess >= 2 (the kill rule's bar) cert2same excess >= 2 and both halves with the same sign (the published survival condition, ignoring the >= 100 trades clause, which 2026 data never meets) Under the null a calibrated statistic has s.d. 1 and cert2 fires about 2.3% of the time. Usage: python3 calibrate.py [--years 100] [--series 300] [--edge 0.15] [--seed 7] Writes calibration.md next to this file. """ import csv import os import random import statistics import sys import tempfile from datetime import datetime, timezone sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import backtest # noqa: E402 HERE = os.path.dirname(os.path.abspath(__file__)) DRIFT_MINUTES = 120 def arg(name, default, cast): return cast(sys.argv[sys.argv.index(name) + 1]) if name in sys.argv else default def load_rows(path): days = {} with open(path) as f: for r in csv.DictReader(f): t = datetime.fromisoformat(r["minute"]) days.setdefault(t.date(), []).append((r["minute"], t.hour * 60 + t.minute, *[float(r[k]) for k in ("bid_open", "bid_high", "bid_low", "bid_close", "ask_open", "ask_high", "ask_low", "ask_close")])) return dict(sorted(days.items())) def first_breakout(bars): """Index of the bar after the first breakout close and its direction, per the rule; None if none.""" rng_bars = [b for b in bars if b[0] < backtest.RANGE_END] if len(rng_bars) < 300: return None rh = max((b[2] + b[6]) / 2 for b in rng_bars) rl = min((b[3] + b[7]) / 2 for b in rng_bars) for i, b in enumerate(bars): if backtest.ENTRY_START <= b[0] < backtest.ENTRY_END and i + 1 < len(bars): mid_close = (b[4] + b[8]) / 2 if mid_close > rh: return i + 1, 1 if mid_close < rl: return i + 1, -1 return None def synth_day(rows, rng, edge): """One synthetic day: sign-flipped mid returns, real spreads and wick shapes, optional injected drift.""" mids_o = [(r[2] + r[6]) / 2 for r in rows] mids_c = [(r[5] + r[9]) / 2 for r in rows] # returns: open->close inside the bar, and close->next open between bars new_o, new_c = [mids_o[0]], [] for i, r in enumerate(rows): s = rng.choice((1, -1)) new_c.append(new_o[-1] + s * (mids_c[i] - mids_o[i])) if i + 1 < len(rows): s2 = rng.choice((1, -1)) new_o.append(new_c[-1] + s2 * (mids_o[i + 1] - mids_c[i])) def build(new_o, new_c): out = [] for i, r in enumerate(rows): stamp, m, bo, bh, bl, bc, ao, ah, al, ac = r spread_o, spread_c = ao - bo, ac - bc spread_h, spread_l = ah - bh, al - bl up_wick = ((bh + ah) / 2) - max(mids_o[i], mids_c[i]) dn_wick = min(mids_o[i], mids_c[i]) - ((bl + al) / 2) o, c = new_o[i], new_c[i] h = max(o, c) + max(up_wick, 0) lo = min(o, c) - max(dn_wick, 0) out.append((stamp, m, o - spread_o / 2, h - spread_h / 2, lo - spread_l / 2, c - spread_c / 2, o + spread_o / 2, h + spread_h / 2, lo + spread_l / 2, c + spread_c / 2)) return out bars = build(new_o, new_c) if edge: sig = first_breakout([b[1:] for b in bars]) if sig is not None: k, d = sig for j in range(k, min(k + DRIFT_MINUTES, len(rows))): step = d * edge * (j - k + 1) new_o[j] += step new_c[j] += step for j in range(k + DRIFT_MINUTES, len(rows)): new_o[j] += d * edge * DRIFT_MINUTES new_c[j] += d * edge * DRIFT_MINUTES bars = build(new_o, new_c) return bars def write_year(days, rng, edge, path): with open(path, "w", newline="") as f: w = csv.writer(f) w.writerow(["minute", "bid_open", "bid_high", "bid_low", "bid_close", "ask_open", "ask_high", "ask_low", "ask_close", "ticks"]) for rows in days.values(): for b in synth_day(rows, rng, edge): w.writerow([b[0]] + [f"{x:.3f}" for x in b[2:]] + [0]) def batch(days, years, series, edge, seed, label): rng = random.Random(seed) out = [] tmp = os.path.join(tempfile.gettempdir(), f"calib-{label}-{os.getpid()}.csv") for y in range(years): write_year(days, rng, edge, tmp) res, skipped, trades = backtest.run(tmp, series, rng.randrange(1 << 30)) a, h1, h2 = res["all"], res["jan-apr"], res["may-aug"] if a is None: continue same = h1 is not None and h2 is not None and (h1["rule_mean"] - h1["control_mean"]) * (h2["rule_mean"] - h2["control_mean"]) > 0 out.append({"trades": a["trades"], "rule_mean": a["rule_mean"], "control_se": a["control_se"], "excess": a["excess_se"], "same_sign": same}) print(f"{label} year {y + 1:3d}: trades {a['trades']:3d} rule {a['rule_mean']:+6.2f} ctrl_se {a['control_se']:5.2f} excess {a['excess_se']:+5.2f} same={same}", flush=True) if os.path.exists(tmp): os.remove(tmp) return out def summarise(rows): ex = [r["excess"] for r in rows] n = len(rows) return { "years": n, "trades_mean": statistics.fmean(r["trades"] for r in rows), "excess_mean": statistics.fmean(ex), "excess_sd": statistics.pstdev(ex), "cert2": sum(e >= 2 for e in ex) / n, "cert2same": sum(e >= 2 and r["same_sign"] for e, r in zip(ex, rows)) / n, "cert1_5": sum(e >= 1.5 for e in ex) / n, "cert1_5same": sum(e >= 1.5 and r["same_sign"] for e, r in zip(ex, rows)) / n, } def main(): years, series, edge, seed = arg("--years", 100, int), arg("--series", 300, int), arg("--edge", 0.15, float), arg("--seed", 7, int) null_years, edge_years = arg("--null-years", years, int), arg("--edge-years", years, int) path = os.path.expanduser("~/data/bars/XAUUSD_mt5_m1.csv") days = load_rows(path) null = batch(days, null_years, series, 0.0, seed, "null") pos = batch(days, edge_years, series, edge, seed + 1, "edge") if not null or not pos: print("one arm is empty; use summarise_log.py on the log instead") return s0, s1 = summarise(null), summarise(pos) lines = [f"# Calibration of the London-breakout test — run {datetime.now(timezone.utc):%Y-%m-%d %H:%M} UTC", "", "This is a test record, not advice.", "", f"Source days: `{path}` ({len(days)} calendar days). Synthetic years per arm: {years}. " f"Control series per run: {series}. Seed {seed}. Positive-control drift: {edge} USD/oz per minute for {DRIFT_MINUTES} minutes after the breakout.", "", "| arm | years | mean trades | excess mean | excess s.d. | ≥2 s.e. | ≥2 s.e. and halves agree | ≥1.5 s.e. | ≥1.5 and agree |", "|---|---|---|---|---|---|---|---|---|"] for name, s in (("null (sign-flipped returns)", s0), (f"edge injected ({edge}/min × {DRIFT_MINUTES} min)", s1)): lines.append(f"| {name} | {s['years']} | {s['trades_mean']:.0f} | {s['excess_mean']:+.2f} | {s['excess_sd']:.2f} | " f"{s['cert2']:.1%} | {s['cert2same']:.1%} | {s['cert1_5']:.1%} | {s['cert1_5same']:.1%} |") lines += ["", "Excess statistic per null year (sorted):", "", " " + " ".join(f"{e:+.2f}" for e in sorted(r["excess"] for r in null)), ""] text = "\n".join(lines) + "\n" with open(os.path.join(HERE, "calibration.md"), "w") as f: f.write(text) print(text) if __name__ == "__main__": main()