"""_ictsb01_silver_bullet_20261020.py - ICTSB01: ICT Silver Bullet on the US100 CFD (Nasdaq-100), 10:00-11:00 New York. DISCOVERY ONLY. Written by: Claude (2026-10-20) for the Aydeus Lab re-test of the ICT Silver Bullet. EMPIRICAL RESEARCH CANDIDATE (NOT LAW / NOT DEPLOYED). Prereg: prereg/ICTSB01_PREREG.md (+ .lock). Replaces the unaudited `_test_ict_silver_bullet_clinical.py` (fixed point thresholds, no draw on liquidity, mid prices + flat cost, limit filled at the gap edge on any touch). Modes --precheck data range, stale quotes, order counts, plan risk and ex-ante cost. No fills, no outcomes. Run before the freeze. --freeze writes prereg/ICTSB01_PREREG.lock (sha256 of the prereg and of this file) and one event in the miner ledger (run_id ICTSB01). --run the one frozen run (refuses to start when the lock does not match): primary rule, diagnostics D1-D5, naive mid version, random-direction placebo (200 seeds), always-long / always-short, bar phase -3..+3, 2x cost, by year, halves, kill gates K1-K8. Primary rule P (all pattern levels on mid prices, all times America/New_York, session day = 18:00 ET -> 17:00 ET): pools PDH / PDL = high / low of the previous session day; ONH / ONL = high / low of the current session day before 09:30. A pool is taken once a mid high exceeds it (a low goes below it) after it formed; status read at 10:00 from minutes before 10:00. draw (DOL) P0 = mid close of the last minute before 10:00. The draw is the nearest untaken pool to P0 (above: PDH / ONH, below: PDL / ONL); no untaken pool on either side = no trade. Direction = towards the draw; the draw level is the target. FVG M5 mid bars (08:45 onwards, 5-minute buckets of the true-clock minutes). Three candles c1 c2 c3 all inside 10:00-11:00 (c1 opens >= 10:00). Bullish: low(c3) - high(c1) >= 0.10 x ATR14(M5); bearish: low(c1) - high(c3) >= 0.10 x ATR14. ATR14 = mean true range of the 14 M5 bars before c1. Only FVGs in the draw's direction count; the first that qualifies is traded. order limit at the FVG's near edge (long: low(c3), short: high(c3)), live from the close of c3 to 10:59 (the last minute before 11:00). stop beyond the FVG's first candle: long low(c1), short high(c1). 1R = |limit - stop|. qualifies room to the draw >= 1R (target - limit, signed); 1R <= 5 ATR14 (data-error guard). The draw must not have been reached between 10:00 and the close of c3 (draw delivered = no trade). One trade a day. cancel before the fill, the target is reached (long: bid high >= target) -> order cancelled, no trade. exit target (limit), stop, or flat at the 16:00 minute open. Execution (G9, pessimistic; hdam_sim helpers `_manage`, `_flat`, `_row` imported unchanged): limit entry long fills when ask low <= limit - 0.25 pt (trade-through), at the limit (no price improvement, no slippage); short on the bid. Entry minute: when the minute opens at / through the limit the whole minute is after the fill (stop, then target, checked); otherwise the stop is checked (price passes the limit before the stop) and the target is not credited. target limit: long needs bid high >= target + 0.25, filled at the target, no slippage; stop first when both fall in one minute. stop / flat long stop on the bid: min(bid open, stop) - 0.5 pt; flat at the bid open - 0.5 (shorts on the ask). Real file spread. R (exit - entry) x side / 1R, net of spread and slippage (hdam `_row`). cost_r = (half spread in + half spread out + 2 x 0.5) / 1R. Diagnostics (reported, never selected): D1 no draw filter (first qualifying FVG of either direction, target = nearest untaken pool on its side), D2 sweep-reversal draw (exactly one side swept before 10:00 -> draw on the other side), D3 the same rules on M1 bars, D4 fixed 2R target with the draw's direction, D5 limit at the FVG midpoint (consequent encroachment). NAIVE = primary on mid prices, touch fills, no cost (what a simple backtest shows). Placebo: per day a coin flip chooses the direction (same engine, same target rule on that side), 200 seeds. """ from __future__ import annotations import hashlib import json import os import sys from types import SimpleNamespace import numpy as np import pandas as pd HERE = os.path.dirname(os.path.abspath(__file__)) LAB = os.path.dirname(HERE) sys.path.insert(0, os.path.join(LAB, "hdam")) sys.path.insert(0, os.path.join(LAB, "prop_gpu_miner")) sys.path.insert(0, LAB) import hdam_sim as H # noqa: E402 NY = "America/New_York" SYM = "nas100usd" THROUGH = 0.25 # trade-through for limit entries and limit targets, points (file tick 0.1) SLIP = 0.5 # points per side on stop and flat exits (hdam cost-screen convention for nas100usd) GAP_MIN_ATR = 0.10 RISK_MAX_ATR = 5.0 ROOM_MIN_R = 1.0 N_PLACEBO = 200 LAB_COST_PTS, LAB_REF_PX = 1.70, 29794.05 # the lab's US100 model: 1.70 pt round trip at the reference price, scaled by price (prop_gpu_miner/PREREG.md) PREREG = os.path.join(LAB, "prereg", "ICTSB01_PREREG.md") LOCK = os.path.join(LAB, "prereg", "ICTSB01_PREREG.lock") OUT_PRE = os.path.join(HERE, "_ictsb01_precheck_20261020.out") OUT_RUN = os.path.join(HERE, "_ictsb01_run_20261020.out") TRADES = os.path.join(HERE, "_ictsb01_trades_20261020.csv") DIAG = os.path.join(HERE, "_ictsb01_diag_trades_20261020.csv") MIN = 60 * 10**9 LINES: list[str] = [] def pr(*a) -> None: s = " ".join(str(x) for x in a) print(s, flush=True) LINES.append(s) def sha256(path: str) -> str: with open(path, "rb") as f: return hashlib.sha256(f.read()).hexdigest() # ----------------------------------------------------------------------------------------------------------------- data def fast_book(book: H.Book, shift_min: int = 0, mid: bool = False) -> SimpleNamespace: """The arrays hdam's helpers read, with the spread computed once (Book.spread recomputes the whole array on every access). mid=True: bid = ask = mid (zero spread), for the gross run under the same fill mechanics.""" mo, mh, ml, mc = (book.bo + book.ao) / 2, (book.bh + book.ah) / 2, (book.bl + book.al) / 2, (book.bc + book.ac) / 2 if mid: return SimpleNamespace(symbol=book.symbol, t=book.t + shift_min * MIN, bo=mo, bh=mh, bl=ml, bc=mc, ao=mo, ah=mh, al=ml, ac=mc, spread=np.zeros_like(mo), mo=mo, mh=mh, ml=ml, mc=mc) return SimpleNamespace(symbol=book.symbol, t=book.t + shift_min * MIN, bo=book.bo, bh=book.bh, bl=book.bl, bc=book.bc, ao=book.ao, ah=book.ah, al=book.al, ac=book.ac, spread=book.ac - book.bc, mo=mo, mh=mh, ml=ml, mc=mc) def ny_ns(day: pd.Timestamp, hm: str) -> int: return pd.Timestamp(f"{day.date()} {hm}", tz=NY).tz_convert("UTC").value def sessions(b: SimpleNamespace) -> list[tuple[pd.Timestamp, int, int]]: loc = pd.to_datetime(b.t, unit="ns", utc=True).tz_convert(NY).tz_localize(None) sd = (loc + pd.Timedelta(hours=6)).normalize() codes = sd.asi8 starts = np.flatnonzero(np.r_[True, codes[1:] != codes[:-1]]) ends = np.r_[starts[1:], len(codes)] return [(pd.Timestamp(codes[s]), int(s), int(e)) for s, e in zip(starts, ends)] def bars(b: SimpleNamespace, day: pd.Timestamp, start_hm: str, n: int, width: int): """n mid bars of `width` minutes from start_hm NY. Returns o, h, l, c (NaN when empty) and the global index of the first minute after each bar.""" t0 = ny_ns(day, start_hm) edges = t0 + np.arange(n + 1, dtype=np.int64) * width * MIN pos = np.searchsorted(b.t, edges, "left") o = np.full(n, np.nan); h = o.copy(); lo = o.copy(); c = o.copy() for i in range(n): a, z = pos[i], pos[i + 1] if z > a: o[i], h[i], lo[i], c[i] = b.mo[a], b.mh[a:z].max(), b.ml[a:z].min(), b.mc[z - 1] return o, h, lo, c, pos[1:] def atr14(h, lo, c) -> np.ndarray: n = len(h) tr = np.full(n, np.nan) tr[1:] = np.maximum(h[1:], c[:-1]) - np.minimum(lo[1:], c[:-1]) out = np.full(n, np.nan) for i in range(15, n): w = tr[i - 14:i] if np.all(np.isfinite(w)): out[i] = w.mean() return out def day_context(b: SimpleNamespace, day: pd.Timestamp, s: int, e: int, ps: int, pe: int) -> dict | None: """Pools, draw and sweep state at 10:00, from minutes strictly before 10:00 NY.""" if pe - ps < 300: return None k930 = int(np.searchsorted(b.t, ny_ns(day, "09:30"), "left")) k1000 = int(np.searchsorted(b.t, ny_ns(day, "10:00"), "left")) if not (s + 60 <= k930 and k930 + 25 <= k1000 <= e): return None if b.t[k1000 - 1] < ny_ns(day, "09:55"): return None pdh, pdl = b.mh[ps:pe].max(), b.ml[ps:pe].min() onh, onl = b.mh[s:k930].max(), b.ml[s:k930].min() hi_pre, lo_pre = b.mh[s:k1000].max(), b.ml[s:k1000].min() hi_rth, lo_rth = b.mh[k930:k1000].max(), b.ml[k930:k1000].min() taken = dict(PDH=hi_pre > pdh, PDL=lo_pre < pdl, ONH=hi_rth > onh, ONL=lo_rth < onl) p0 = b.mc[k1000 - 1] up = [(lv, nm) for lv, nm in ((pdh, "PDH"), (onh, "ONH")) if not taken[nm] and lv > p0] dn = [(lv, nm) for lv, nm in ((pdl, "PDL"), (onl, "ONL")) if not taken[nm] and lv < p0] nu = min(up) if up else None nd = max(dn) if dn else None if nu is None and nd is None: dol = 0 elif nd is None or (nu is not None and nu[0] - p0 < p0 - nd[0]): dol = 1 else: dol = -1 sw_sell, sw_buy = taken["PDL"] or taken["ONL"], taken["PDH"] or taken["ONH"] sweep = 1 if (sw_sell and not sw_buy) else (-1 if (sw_buy and not sw_sell) else 0) return dict(p0=p0, pool={1: nu, -1: nd}, dol=dol, sweep=sweep, k1000=k1000, decided_before=int(b.t[k1000 - 1])) def find_order(o, h, lo, c, end_pos, atr, d: int, i_first: int, i_last: int, pool, entry: str, target: str): """First qualifying FVG in direction d. Bars i_first.. are the 10:00-11:00 window; i_last = last c3 index.""" lvl = pool[0] if pool is not None else None if target == "pool" and lvl is None: return None, "no_pool" ext = -np.inf if d == 1 else np.inf for i3 in range(i_first + 2, i_last + 1): i1 = i3 - 2 # draw delivered between 10:00 and the close of c3 (bars i_first..i3) = no trade for j in range(max(i_first, i3 - 2), i3 + 1): if np.isfinite(h[j]): ext = max(ext, h[j]) if d == 1 else min(ext, lo[j]) if lvl is not None and ((d == 1 and ext >= lvl) or (d == -1 and ext <= lvl)): return None, "draw_delivered" a = atr[i1] if not (np.isfinite(a) and np.all(np.isfinite(h[i1:i3 + 1]))): continue if d == 1: gap = lo[i3] - h[i1] lim = lo[i3] if entry == "edge" else 0.5 * (lo[i3] + h[i1]) stop = lo[i1] else: gap = lo[i1] - h[i3] lim = h[i3] if entry == "edge" else 0.5 * (h[i3] + lo[i1]) stop = h[i1] if gap < GAP_MIN_ATR * a: continue risk = (lim - stop) * d if risk <= 0 or risk > RISK_MAX_ATR * a: continue if target == "pool": if (lvl - lim) * d < ROOM_MIN_R * risk: continue tgt = lvl else: tgt = lim + d * 2.0 * risk return dict(i3=i3, lim=lim, stop=stop, risk=risk, tgt=tgt, atr=a, gap=gap, live_from=int(end_pos[i3]), pool=(pool[1] if pool is not None else "")), "order" return None, "no_fvg" def simulate_limit(b: SimpleNamespace, od: dict, d: int, k_to: int, k_flat: int, k_flat_ok, tag: str, slip: float = SLIP) -> tuple[dict | None, str]: """Pessimistic limit entry + hdam management. Returns (row, status).""" lim, stop, tgt, risk = od["lim"], od["stop"], od["tgt"], od["risk"] for k in range(od["live_from"], min(k_to, k_flat)): if d == 1: touched, delivered = b.al[k] <= lim - THROUGH, b.bh[k] >= tgt else: touched, delivered = b.bh[k] >= lim + THROUGH, b.al[k] <= tgt if not touched: if delivered: return None, "cancel_target" continue gapped = (b.ao[k] <= lim) if d == 1 else (b.bo[k] >= lim) px = why = None k_out = k if d == 1 and b.bl[k] <= stop: px, why = (min(b.bo[k], stop) if gapped else stop) - slip, "sl_entry_minute" elif d == -1 and b.ah[k] >= stop: px, why = (max(b.ao[k], stop) if gapped else stop) + slip, "sl_entry_minute" elif gapped and ((d == 1 and b.bh[k] >= tgt + THROUGH) or (d == -1 and b.al[k] <= tgt - THROUGH)): px, why = tgt, "tp_entry_minute" if px is None: px, k_out, why = H._manage(b, k, k_flat, d, lim, stop, tgt + d * THROUGH, slip, check_entry_minute=False) if why == "tp": px = tgt if px is None: px, k_out, why = H._flat(b, k_flat_ok, k_flat - 1, d, slip) row = H._row(b, tag, d, k, k_out, lim, px, risk, why, H.Costs(slip_pts=slip), k - 1) return row, "filled" return None, "no_fill" def simulate_naive(b: SimpleNamespace, od: dict, d: int, k_to: int, k_flat: int) -> dict | None: """What a simple backtest does: mid prices, fill on touch, stop first from the fill minute, no cost.""" lim, stop, tgt, risk = od["lim"], od["stop"], od["tgt"], od["risk"] for k in range(od["live_from"], min(k_to, k_flat)): if not ((d == 1 and b.ml[k] <= lim) or (d == -1 and b.mh[k] >= lim)): if (d == 1 and b.mh[k] >= tgt) or (d == -1 and b.ml[k] <= tgt): return None continue for j in range(k, k_flat): if (d == 1 and b.ml[j] <= stop) or (d == -1 and b.mh[j] >= stop): return dict(R=-1.0) if (d == 1 and b.mh[j] >= tgt) or (d == -1 and b.ml[j] <= tgt): return dict(R=(tgt - lim) * d / risk) return dict(R=(b.mo[k_flat] - lim) * d / risk if k_flat < len(b.t) else (b.mc[k_flat - 1] - lim) * d / risk) return None SPECS = { # name: (bar width, start, n bars, index of the 10:00 bar, entry, target) "M5": (5, "08:45", 27, 15, "edge", "pool"), "M5_2R": (5, "08:45", 27, 15, "edge", "2R"), "M5_CE": (5, "08:45", 27, 15, "ce", "pool"), "M1": (1, "09:45", 75, 15, "edge", "pool"), } def build_days(b: SimpleNamespace, specs=("M5", "M5_2R", "M5_CE", "M1"), fills: bool = True, naive: bool = False, slip: float = SLIP) -> pd.DataFrame: """One row per (day, spec, direction): the order (if any) and, when fills=True, its simulated trade.""" ss = sessions(b) rows = [] for j in range(1, len(ss)): day, s, e = ss[j] if day.dayofweek >= 5: continue _, ps, pe = ss[j - 1] ctx = day_context(b, day, s, e, ps, pe) if ctx is None: continue k_to = int(np.searchsorted(b.t, ny_ns(day, "11:00"), "left")) tf = ny_ns(day, "16:00") k_flat = int(np.searchsorted(b.t, tf, "left")) k_flat_ok = k_flat if (k_flat < len(b.t) and b.t[k_flat] - tf <= H.FLAT_TOLERANCE_NS) else None for sp in specs: width, st, n, i0, entry, target = SPECS[sp] o, h, lo, c, end_pos = bars(b, day, st, n, width) atr = atr14(h, lo, c) for d in (1, -1): pool = ctx["pool"][d] if target == "2R" and d != ctx["dol"]: continue # D4 keeps the draw's direction only od, why = find_order(o, h, lo, c, end_pos, atr, d, i0, n - 1, pool, entry, target) r = dict(day=day, spec=sp, d=d, dol=ctx["dol"], sweep=ctx["sweep"], status=why) if od is not None: assert ctx["decided_before"] < ny_ns(day, "10:00") <= b.t[od["live_from"]] or od["live_from"] >= k_to r.update(lim=od["lim"], stop=od["stop"], tgt=od["tgt"], risk=od["risk"], atr=od["atr"], gap=od["gap"], pool=od["pool"], i3=od["i3"], live_from=od["live_from"], s_pre=b.spread[od["live_from"] - 1], cost_ex_ante=(b.spread[od["live_from"] - 1] + 2 * slip) / od["risk"], cost_exec_max=(b.spread[od["live_from"] - 1] + slip) / od["risk"], cost_lab=LAB_COST_PTS * od["lim"] / LAB_REF_PX / od["risk"]) if fills: tr, st2 = simulate_limit(b, od, d, k_to, k_flat, k_flat_ok, f"{sp}_{d:+d}", slip) r["status"] = st2 if tr is not None: r.update({f"x_{k}": v for k, v in tr.items()}) if naive and sp == "M5": nv = simulate_naive(b, od, d, k_to, k_flat) r["naive_R"] = nv["R"] if nv is not None else np.nan rows.append(r) return pd.DataFrame(rows) # ----------------------------------------------------------------------------------------------------------------- statistics def stats(x) -> dict: x = np.asarray(x, float) x = x[np.isfinite(x)] if len(x) < 3: return dict(n=len(x), mean=np.nan, t=np.nan, sum=float(x.sum()) if len(x) else 0.0) return dict(n=len(x), mean=float(x.mean()), t=H.tstat(x), sum=float(x.sum())) def fmt(s: dict) -> str: return f"n {s['n']:5d} mean {s['mean']:+.4f}R t {s['t']:+.2f} sum {s['sum']:+.1f}R" def trades_of(D: pd.DataFrame, spec: str, pick: str) -> pd.DataFrame: """pick: 'dol' | 'sweep' | 'first' | 'long' | 'short'. Rows with a fill only.""" X = D[D.spec == spec] if pick == "dol": X = X[X.d == X.dol] elif pick == "sweep": X = X[(X.sweep != 0) & (X.d == X.sweep)] elif pick == "long": X = X[X.d == 1] elif pick == "short": X = X[X.d == -1] elif pick == "first": X = X[X.status.isin(["filled", "no_fill", "cancel_target"])].sort_values(["day", "i3"]).groupby("day", as_index=False).head(1) return X[X.status == "filled"].copy() # ----------------------------------------------------------------------------------------------------------------- modes def precheck() -> None: book = H.load_book(SYM, "discovery") b = fast_book(book) pr(f"ICTSB01 precheck {pd.Timestamp.now('UTC'):%Y-%m-%d %H:%MZ} symbol {SYM} stage discovery hdam_sim sha256 {sha256(os.path.join(LAB, 'hdam', 'hdam_sim.py'))}") pr(f"minutes {len(b.t):,} first {pd.Timestamp(int(b.t[0]), tz='UTC')} last {pd.Timestamp(int(b.t[-1]), tz='UTC')} file price increment 0.1") st = H.stale_share(book, NY, "10:00", "10:59") pr("flat-minute share 10:00-10:59 NY by year (limit 10%):", " ".join(f"{y}:{v:.3f}" for y, v in st.items()), " max", f"{st.max():.3f}") D = build_days(b, specs=("M5", "M1"), fills=False) days = D.day.nunique() pr(f"eligible days {days:,}") P = D[(D.spec == "M5") & (D.d == D.dol) & (D.status == "order")] pr(f"draw direction: long {np.mean(D[D.spec == 'M5'].drop_duplicates('day').dol == 1):.3f}, short {np.mean(D[D.spec == 'M5'].drop_duplicates('day').dol == -1):.3f}, none {np.mean(D[D.spec == 'M5'].drop_duplicates('day').dol == 0):.3f}") pr("primary order status counts:", D[(D.spec == "M5") & (D.d == D.dol)].status.value_counts().to_dict()) pr(f"primary orders {len(P):,} on {days:,} days ({len(P) / days:.1%}); target pool {P.pool.value_counts().to_dict()}") for sp in ("M5", "M1"): Q = D[(D.spec == sp) & (D.d == D.dol) & (D.status == "order")].copy() Q["year"] = Q.day.dt.year pr(f"\n[{sp}] by year: orders, median 1R (pt), 1R / ATR14, spread before the order (pt), median cost per trade in R: hdam ex-ante (spread + 2 x {SLIP})," f" declared execution worst case (spread + 1 x {SLIP}, a stop exit), lab model ({LAB_COST_PTS} pt x price / {LAB_REF_PX})") for y, g in Q.groupby("year"): pr(f" {y} {len(g):4d} 1R {g.risk.median():7.2f} 1R/ATR {np.median(g.risk / g.atr):4.2f} spread {g.s_pre.median():5.2f} " f"hdam {g.cost_ex_ante.median():.3f}R exec {g.cost_exec_max.median():.3f}R lab {g.cost_lab.median():.3f}R") pr(f" ALL {len(Q):4d} median cost: hdam ex-ante {Q.cost_ex_ante.median():.3f}R, execution worst case {Q.cost_exec_max.median():.3f}R, " f"lab model {Q.cost_lab.median():.3f}R (K4 limit 0.030R); share <= 0.030R: hdam {np.mean(Q.cost_ex_ante <= 0.03):.3f}, lab {np.mean(Q.cost_lab <= 0.03):.3f}; " f"spread-only median {np.median(Q.s_pre / Q.risk):.3f}R") pr("\nNo fills or outcomes were computed in this mode.") open(OUT_PRE, "w", encoding="utf-8").write("\n".join(LINES) + "\n") def freeze() -> None: lock = dict(prereg_sha256=sha256(PREREG), script_sha256=sha256(os.path.abspath(__file__)), hdam_sim_sha256=sha256(os.path.join(LAB, "hdam", "hdam_sim.py")), frozen_utc=pd.Timestamp.now("UTC").isoformat()) json.dump(lock, open(LOCK, "w"), indent=1) import ledger as LG # prop_gpu_miner/ledger.py lg = LG.Ledger() lg.log_run("ICTSB01", lock["prereg_sha256"], "ICT Silver Bullet US100 CFD Discovery; 1 primary rule + 5 declared diagnostics + placebo") lg.event("ICTSB01", "prereg_freeze", dict(n=1, lock=lock)) print(json.dumps(lock, indent=1)) def run() -> None: lock = json.load(open(LOCK)) assert lock["prereg_sha256"] == sha256(PREREG), "prereg changed after the freeze" assert lock["script_sha256"] == sha256(os.path.abspath(__file__)), "script changed after the freeze" assert lock["hdam_sim_sha256"] == sha256(os.path.join(LAB, "hdam", "hdam_sim.py")), "hdam_sim changed" book = H.load_book(SYM, "discovery") b = fast_book(book) pr(f"ICTSB01 run {pd.Timestamp.now('UTC'):%Y-%m-%d %H:%MZ} {SYM} discovery {pd.Timestamp(int(b.t[0]), tz='UTC'):%Y-%m-%d}..{pd.Timestamp(int(b.t[-1]), tz='UTC'):%Y-%m-%d}") pr(f"lock {lock}") D = build_days(b, naive=True) D["year"] = D.day.dt.year days = D.day.nunique() P = trades_of(D, "M5", "dol").sort_values("day").reset_index(drop=True) pr(f"\neligible days {days:,}; primary orders {int(((D.spec == 'M5') & (D.d == D.dol) & D.status.isin(['filled', 'no_fill', 'cancel_target'])).sum()):,}; " f"status {D[(D.spec == 'M5') & (D.d == D.dol)].status.value_counts().to_dict()}") R = P["x_R"] pr("\n== PRIMARY P (M5 FVG, draw on liquidity, limit at the near edge, target = draw, bid/ask pessimistic)") pr(" net 1x ", fmt(stats(R))) pr(" net 2x ", fmt(stats(R - P.x_cost_r))) pr(" net 3x ", fmt(stats(R - 2 * P.x_cost_r))) pr(f" win rate {np.mean(R > 0):.3f}; median 1R {P.x_plan_risk.median():.2f} pt; median cost {P.x_cost_r.median():.3f}R (ex-ante {P.x_cost_r_ex_ante.median():.3f}R); " f"long share {np.mean(P.d == 1):.3f}; exits {P.x_reason.value_counts().to_dict()}") nv = D[(D.spec == "M5") & (D.d == D.dol)].naive_R pr(" NAIVE (mid, touch fill, no cost)", fmt(stats(nv))) G = trades_of(build_days(fast_book(book, mid=True), specs=("M5",), slip=0.0), "M5", "dol") pr(" GROSS (same pessimistic fill rules on mid prices, zero spread, zero slip)", fmt(stats(G.x_R))) pr(" NET at the lab model (gross - 1.70 pt x price / ref)", fmt(stats(G.x_R - G.cost_lab)), f" median lab cost {G.cost_lab.median():.3f}R") pr(f" break-even cost per trade = GROSS mean {G.x_R.mean():+.4f}R (no cost level makes a non-positive gross profitable)") pr("\n by year (net 1x):") for y, g in P.groupby(P.day.dt.year): s = stats(g.x_R) pr(f" {y} n {s['n']:4d} mean {s['mean']:+.4f} t {s['t']:+.2f} sum {s['sum']:+7.1f} win {np.mean(g.x_R > 0):.2f}") pos_years = int((P.groupby(P.day.dt.year).x_R.sum() > 0).sum()) pr(f" positive years {pos_years} of {P.day.dt.year.nunique()}") med = P.day.iloc[len(P) // 2] h1, h2 = stats(P[P.day < med].x_R), stats(P[P.day >= med].x_R) pr(f" halves (split at {med.date()}): first {fmt(h1)} | second {fmt(h2)}") pr(" regimes (calendar): " + " | ".join(f"{a}-{z} {fmt(stats(P[(P.day.dt.year >= a) & (P.day.dt.year <= z)].x_R))}" for a, z in ((2008, 2008), (2011, 2011), (2015, 2016), (2020, 2020)))) pr("\n== DIAGNOSTICS (declared in the prereg; reported, never selected)") diag = {"D1 no draw filter (first FVG either side)": ("M5", "first"), "D2 sweep-reversal draw": ("M5", "sweep"), "D3 M1 bars": ("M1", "dol"), "D4 fixed 2R target": ("M5_2R", "dol"), "D5 limit at the FVG midpoint": ("M5_CE", "dol"), "always long (M5, target = nearest pool above)": ("M5", "long"), "always short (M5, nearest pool below)": ("M5", "short")} keep = [P.assign(variant="P")] for name, (sp, pick) in diag.items(): X = trades_of(D, sp, pick) s1, s2 = stats(X.x_R), stats(X.x_R - X.x_cost_r) pr(f" {name:48s} net {fmt(s1)} | 2x t {s2['t']:+.2f} | median cost {X.x_cost_r.median():.3f}R | win {np.mean(X.x_R > 0):.3f}") keep.append(X.assign(variant=name.split(" ")[0] if name.startswith("D") else name.split(" ")[1])) pr("\n== PLACEBO: random direction per day, same engine and target rule (M5), 200 seeds") L = D[(D.spec == "M5")].set_index(["day", "d"]) allday = np.array(sorted(D[D.spec == "M5"].day.unique())) rl = L.xs(1, level="d").reindex(allday) rs = L.xs(-1, level="d").reindex(allday) fl = (rl.status == "filled").to_numpy(); fs = (rs.status == "filled").to_numpy() Rl, Rs = rl.x_R.to_numpy(float), rs.x_R.to_numpy(float) means, ts = [], [] for seed in range(N_PLACEBO): coin = np.random.default_rng(seed).integers(0, 2, len(allday)) == 1 x = np.where(coin, np.where(fl, Rl, np.nan), np.where(fs, Rs, np.nan)) s = stats(x) means.append(s["mean"]); ts.append(s["t"]) means = np.array(means) p_plac = float(np.mean(means >= R.mean())) pr(f" placebo mean R: median {np.median(means):+.4f}, 5% {np.quantile(means, 0.05):+.4f}, 95% {np.quantile(means, 0.95):+.4f}, 99% {np.quantile(means, 0.99):+.4f}; " f"trades per seed ~{int(np.mean([np.isfinite(np.where(np.random.default_rng(s).integers(0, 2, len(allday)) == 1, np.where(fl, Rl, np.nan), np.where(fs, Rs, np.nan))).sum() for s in range(5)]))}") pr(f" primary mean {R.mean():+.4f}R -> share of placebo seeds >= primary: {p_plac:.3f}") pr("\n== BAR PHASE: the whole rule with the clock shifted by k minutes (primary, net 1x)") phase = {} for k in (-3, -2, -1, 0, 1, 2, 3): if k == 0: s = stats(R) else: Dk = build_days(fast_book(book, k), specs=("M5",)) s = stats(trades_of(Dk, "M5", "dol").x_R) phase[k] = s pr(f" phase {k:+d} min {fmt(s)}") s1 = stats(R); s2 = stats(R - P.x_cost_r) K = { "K1 net t >= 3.43": s1["t"] >= 3.43, "K2 same sign both halves": np.sign(h1["mean"]) == np.sign(h2["mean"]) and h1["mean"] > 0, "K3 2x cost t >= 1.0": s2["t"] >= 1.0, "K4 median ex-ante cost <= 0.030R": P.x_cost_r_ex_ante.median() <= 0.030, "K5 phases -3..+3 same sign, min t >= 0.5 x t0": all(np.sign(v["mean"]) == np.sign(s1["mean"]) for v in phase.values()) and s1["mean"] > 0 and min(v["t"] for v in phase.values()) >= 0.5 * s1["t"], "K6 placebo share >= primary < 0.01": p_plac < 0.01, } pr("\n== KILL GATES (Discovery, FAIL ANYWHERE = FAIL)") for k, v in K.items(): pr(f" {k:48s} {'PASS' if v else 'FAIL'}") pr(f" (K4 for information: lab-model median cost {P.cost_lab.median():.3f}R, declared-execution worst case {P.cost_exec_max.median():.3f}R)") pr(" K7 filter increment n/a (the draw chooses a side, it is not a keep/drop filter; the placebo tests it)") pr(" K8 overlap with the production strategy " + ("computed only if K1-K6 pass" if not all(K.values()) else "TO RUN")) verdict = "SURVIVES DISCOVERY (UNVERIFIED CANDIDATE - GATES PENDING)" if all(K.values()) else "KILLED IN DISCOVERY" pr(f"\nVERDICT ICTSB01: {verdict}. Validation (2021-01..2023-06) and holdout (2023-07..) not opened.") cols = ["day", "d", "dol", "sweep", "pool", "lim", "stop", "tgt", "risk", "atr", "gap", "x_t_in", "x_t_out", "x_entry", "x_exit", "x_reason", "x_R", "x_cost_r", "x_cost_r_ex_ante", "cost_lab"] keep.append(G.assign(variant="GROSS")) P[cols].rename(columns=lambda c: c[2:] if c.startswith("x_") else c).to_csv(TRADES, index=False) pd.concat(keep, ignore_index=True)[["variant"] + cols].rename(columns=lambda c: c[2:] if c.startswith("x_") else c).to_csv(DIAG, index=False) pr(f"\ntrades -> {os.path.relpath(TRADES, LAB)}; all variants -> {os.path.relpath(DIAG, LAB)}") open(OUT_RUN, "w", encoding="utf-8").write("\n".join(LINES) + "\n") if __name__ == "__main__": mode = sys.argv[1] if len(sys.argv) > 1 else "--precheck" {"--precheck": precheck, "--freeze": freeze, "--run": run}[mode]()