feat: orb_pmhigh_gate_v1 — premarket-high reclaim gate, forward OOS test

H3 (enter the pm-high break at 9:30 instead of the ORB) FAILED its
pre-registered bar on the year replay: paired delta -0.069R (n=689) —
entering earlier on the same names does not pay; the 5-minute wait stays.

The decomposition found the real signal: ORB fires where RTH price also
reclaimed the premarket high ran +0.316R (n=689) vs -0.298R (n=477) where
it never did — 100th percentile of a 2000-draw random-gate null at the
same keep-rate (null p99 +0.127), sign-consistent 14/14 months. First
lever of 15 to survive both checks in-sample.

Because the split was found in this dataset, it counts for nothing until
fresh fires confirm: new 'pmhigh_gate' intervention kind (fail-open on a
missing premarket_structure row, RTH-only reclaim check at/before the
entry bar) + registered spec with expR_delta >= 0.05 at n >= 200 forward
bar. Cohort self-registers on the next nightly experiments run.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

experiments/orb_pmhigh_gate_v1.json

@@ -0,0 +1,9 @@
+{
+  "id": "orb_pmhigh_gate_v1",
+  "hypothesis": "Allow an ORB entry only if regular-hours price has reclaimed the PREMARKET high at/before entry. In-sample (year replay, 2026-08-01): kept fires +0.316R (n=689) vs gated -0.298R (n=477), 100th percentile of a 2000-draw random-gate null at the same keep-rate, 14/14 months sign-consistent. This experiment is the FORWARD out-of-sample test on fresh fires; the in-sample split does not count toward the bar.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "pmhigh_gate"},
+  "bars": {"expR_delta_min": 0.05, "min_n": 200},
+  "futility": {"min_n": 400, "expR_delta_max": -0.02},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

research/experiments/replay.py

@@ -476,6 +476,31 @@ def iv_asym_exit(con, fire, bars, bars_after, params, fidelity):
 # multi-fire coordination needed, so it plugs into the per-fire _compute_fire
 # path instead of a _run_rank-style batch).
 # --------------------------------------------------------------------------- #
+def iv_pmhigh_gate(con, fire: "Fire", bars, entry_idx: int,
+                   baseline: float | None) -> tuple[float | None, dict]:
+    """variant_r = baseline if REGULAR-HOURS price reclaimed the PREMARKET high
+    at/before the entry bar, else 0.0 (fire gated out).
+
+    Discovered 2026-08-01 in the H3 replay decomposition: ORB fires where the
+    pm high also broke ran +0.316R (n=689) vs -0.298R (n=477) where it never
+    broke — 100th percentile of a 2000-draw random-gate null, 14/14 months
+    sign-consistent. This experiment is the FORWARD out-of-sample test of that
+    in-sample split. pm_high comes from premarket_structure (nightly labeler);
+    a missing row is KEPT + flagged (fail-open, same discipline as the
+    regime/dilution gates)."""
+    row = con.execute(
+        "SELECT pm_high FROM premarket_structure WHERE session_date = ? AND symbol = ?",
+        [str(fire.session_date), fire.symbol]).fetchall()
+    if not row or row[0][0] is None:
+        return baseline, {"kind": "pmhigh_gate", "broke": None, "reason": "no_pm_row"}
+    pm_high = float(row[0][0])
+    upto = bars.iloc[:entry_idx + 1]
+    rth = upto[upto["is_regular_hours"] == True]  # noqa: E712 — pandas mask
+    broke = bool(len(rth)) and float(rth["high"].max()) >= pm_high
+    detail = {"kind": "pmhigh_gate", "broke": broke, "pm_high": round(pm_high, 4)}
+    return (baseline if broke else 0.0), detail
+
+
 def iv_regime_gate(fire: "Fire", baseline: float | None, favorable: dict,
                    model_version: str | None) -> tuple[float | None, dict]:
     """variant_r = baseline if the session's regime score says favorable, else
@@ -826,6 +851,12 @@ def _compute_fire(con, exp: dict, fire: Fire, favorable: dict | None = None,
     bars_after = bars.iloc[entry_idx + 1:].reset_index(drop=True)
     trigger_high = float(bars.iloc[entry_idx]["high"])
     params, fidelity = exp["intervention"], exp["fidelity"]
+    if kind == "pmhigh_gate":
+        baseline = _sim_baseline_r(fire, bars_after, fidelity) if fire.sim_baseline else fire.baseline_r
+        v, d = iv_pmhigh_gate(con, fire, bars, entry_idx, baseline)
+        if fire.sim_baseline:
+            d["source"] = "setup_sim"
+        return {"baseline_r": baseline, "variant_r": v, "detail": d}
     if kind == "regime_gate":
         # needs the baseline BEFORE computing variant (unlike the other kinds,
         # which compute variant first and attach fire.sim_baseline's simulated

tests/unit/experiments/test_pmhigh_gate.py

@@ -0,0 +1,56 @@
+"""iv_pmhigh_gate: keep when RTH reclaimed the premarket high, zero otherwise."""
+
+from datetime import date
+
+import pandas as pd
+
+from research.experiments.replay import Fire, iv_pmhigh_gate
+
+
+class _Con:
+    def __init__(self, pm_high):
+        self._pm = pm_high
+    def execute(self, sql, params=None):
+        self._rows = [(self._pm,)] if self._pm is not None else []
+        return self
+    def fetchall(self):
+        return self._rows
+
+
+def _bars(rows):
+    """rows: (high, is_regular_hours)"""
+    return pd.DataFrame([{"timestamp": i, "high": h, "low": h - 0.2, "open": h - 0.1,
+                          "close": h - 0.05, "is_regular_hours": reg}
+                         for i, (h, reg) in enumerate(rows)])
+
+
+def _fire():
+    return Fire(fire_id="o1", symbol="TST", session_date=date(2026, 7, 31), fill_ts=None,
+                entry_price=10.0, stop_price=9.0, exit_price=11.0, baseline_r=1.0,
+                trigger_price=10.0)
+
+
+def test_kept_when_rth_broke_pm_high():
+    bars = _bars([(9.5, False), (10.6, True), (10.2, True)])  # RTH high 10.6 >= 10.5
+    v, d = iv_pmhigh_gate(_Con(10.5), _fire(), bars, 2, 1.0)
+    assert v == 1.0 and d["broke"] is True
+
+
+def test_zeroed_when_pm_high_never_reclaimed():
+    # premarket touched 10.5 (definitionally its own high) but RTH never did
+    bars = _bars([(10.5, False), (10.1, True), (10.2, True)])
+    v, d = iv_pmhigh_gate(_Con(10.5), _fire(), bars, 2, 1.0)
+    assert v == 0.0 and d["broke"] is False
+
+
+def test_fail_open_without_pm_row():
+    bars = _bars([(10.6, True)])
+    v, d = iv_pmhigh_gate(_Con(None), _fire(), bars, 0, 1.0)
+    assert v == 1.0 and d["broke"] is None and d["reason"] == "no_pm_row"
+
+
+def test_bars_after_entry_do_not_count():
+    # pm high broken only AFTER the entry bar (index 1) -> gated at entry time
+    bars = _bars([(9.5, False), (10.1, True), (10.9, True)])
+    v, d = iv_pmhigh_gate(_Con(10.5), _fire(), bars, 1, 1.0)
+    assert v == 0.0 and d["broke"] is False