experiments: register orderflow H1 (trigger-erasure exit) + H2 (flow-persistence gate)

From docs/research/orderflow_notes.md — the order-flow shelf distillation.
H1 (5-book convergence): exit on first post-fill 1-min close back below the
trigger within 5 bars — the transitory-impact/recross signature — else the
standing exit stack. H2 (4-book): keep-or-zero on pre-trigger volume-weighted
bar-sign persistence (<0.30 over 15 bars = choppy one-shot burst, zeroed);
strictly-before-entry window, fail-open on thin history. Both pre-registered
at the house bar (+0.05R, n>=300, standard fidelity); nightly adjudication
decides. 7 tests.

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

experiments/orderflow_h1_erasure_v1.json

@@ -0,0 +1,8 @@
+{
+  "id": "orderflow_h1_erasure_v1",
+  "hypothesis": "Trigger-erasure fast invalidation (orderflow_notes.md H1, 5-book convergence): a post-fill 1-min bar CLOSING back below the ORB trigger within 5 bars is the transitory-impact/balance-re-entry signature — exiting at that close instead of riding to the stop shrinks the loser tail and lifts fire-weighted expR. Kill criteria beyond the bar: lost-winner rate >15% of eventual +1R trades exited by the rule (measure from detail.erased x baseline_r>0), or redundancy with the BE@+5% ratchet.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "trigger_erasure", "window_bars": 5, "condition": "close_below"},
+  "bars": {"expR_delta_min": 0.05, "min_n": 300},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

experiments/orderflow_h2_persistence_v1.json

@@ -0,0 +1,8 @@
+{
+  "id": "orderflow_h2_persistence_v1",
+  "hypothesis": "Pre-trigger flow-persistence gate (orderflow_notes.md H2, 4-book convergence): fires whose 15-bar pre-trigger volume-weighted bar-sign persistence is below 0.30 (choppy/balanced run-up = one-shot burst, not an ongoing metaorder) convert worse; zeroing them lifts kept-population expR. Fixed 0.30 threshold approximates the note's bottom-tercile rule. Kill criteria beyond the bar: persistence-tercile monotonicity fails, or the effect collapses within gap%/RVOL strata (selection lever in disguise) — check at adjudication from detail.persistence.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "flow_persistence_gate", "lookback_bars": 15, "min_persistence": 0.30},
+  "bars": {"expR_delta_min": 0.05, "min_n": 300},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

research/experiments/replay.py

@@ -511,6 +511,72 @@ def _load_dilution_history(con) -> dict:
     return out
 
 
+# --------------------------------------------------------------------------- #
+# Order-flow shelf interventions (plan: docs/research/orderflow_notes.md)
+# --------------------------------------------------------------------------- #
+def iv_trigger_erasure(fire: "Fire", bars_after, baseline: float | None,
+                       params: dict, fidelity: dict) -> tuple[float | None, dict]:
+    """H1 (5-book convergence — Dalton/Bellafiore/Neill/Hasbrouck/Bouchaud):
+    a post-fill 1-min bar CLOSING back below the trigger within the window is
+    the transitory-impact / balance-re-entry signature — exit at that close
+    (adverse-adjusted) instead of riding to the stop. No erasure -> the
+    standing exit stack (baseline) unchanged. Trigger is the OR-high reference
+    (fire.trigger_price), independent of the stop — no tautology."""
+    risk = (fire.entry_price - fire.stop_price) if (
+        fire.entry_price is not None and fire.stop_price is not None) else None
+    trigger = fire.trigger_price if fire.trigger_price is not None else fire.entry_price
+    detail = {"kind": "trigger_erasure", "erased": False, "erasure_bar": None,
+              "window_bars": int(params.get("window_bars", 5))}
+    if not risk or risk <= 0 or trigger is None or bars_after is None or not len(bars_after):
+        return baseline, detail
+    adv = _adverse(fidelity)
+    entry_fill = fire.entry_price + adv
+    window = min(detail["window_bars"], len(bars_after))
+    for i in range(window):
+        close = float(bars_after.iloc[i]["close"])
+        if close < trigger:
+            detail["erased"] = True
+            detail["erasure_bar"] = i
+            return ((close - adv) - entry_fill) / risk, detail
+    return baseline, detail
+
+
+def iv_flow_persistence(bars, entry_idx: int, baseline: float | None,
+                        params: dict) -> tuple[float | None, dict]:
+    """H2 (Bouchaud/Cartea/Neill/Hasbrouck): keep-or-zero on pre-trigger
+    volume-weighted bar-sign persistence — |sum(sign(close-open)*vol)| /
+    sum(vol) over the last ``lookback_bars`` STRICTLY BEFORE the entry bar
+    (no look-ahead). Persistent one-sided flow = ongoing metaorder -> KEEP;
+    choppy/balanced run-up -> ZERO. Thin history (<5 bars) or zero volume is
+    the no-data case and FAILS OPEN (keep) — mirrors iv_regime_gate/
+    iv_dilution_gate discipline. Fixed pre-registered threshold approximates
+    the note's bottom-tercile rule (a tercile needs two passes)."""
+    lookback = int(params.get("lookback_bars", 15))
+    min_p = float(params.get("min_persistence", 0.30))
+    detail = {"kind": "flow_persistence_gate", "flagged": False,
+              "lookback_bars": lookback, "threshold": min_p, "persistence": None}
+    pre = bars.iloc[max(0, entry_idx - lookback):entry_idx] if bars is not None else None
+    if pre is None or len(pre) < 5:
+        detail["no_data"] = True
+        return baseline, detail
+    num = 0.0
+    denom = 0.0
+    for _, row in pre.iterrows():
+        o, c, v = float(row["open"]), float(row["close"]), float(row["volume"] or 0)
+        sign = 1 if c > o else (-1 if c < o else 0)
+        num += sign * v
+        denom += v
+    if denom <= 0:
+        detail["no_data"] = True
+        return baseline, detail
+    persistence = abs(num) / denom
+    detail["persistence"] = round(persistence, 4)
+    if persistence < min_p:
+        detail["flagged"] = True
+        return 0.0, detail
+    return baseline, detail
+
+
 # --------------------------------------------------------------------------- #
 # Rank (portfolio-level; receipts-only, starves without plan 005)
 # --------------------------------------------------------------------------- #
@@ -578,6 +644,21 @@ def _compute_fire(con, exp: dict, fire: Fire, favorable: dict | None = None,
         if fire.sim_baseline:
             d["source"] = "setup_sim"
         return {"baseline_r": baseline, "variant_r": v, "detail": d}
+    if kind == "trigger_erasure":
+        # orderflow H1: needs the baseline (standing exit stack) for the
+        # no-erasure branch, so computed the keep-or-zero way, not exit-chain.
+        baseline = _sim_baseline_r(fire, bars_after, fidelity) if fire.sim_baseline else fire.baseline_r
+        v, d = iv_trigger_erasure(fire, bars_after, baseline, params, fidelity)
+        if fire.sim_baseline:
+            d["source"] = "setup_sim"
+        return {"baseline_r": baseline, "variant_r": v, "detail": d}
+    if kind == "flow_persistence_gate":
+        # orderflow H2: gate computed from bars STRICTLY BEFORE the entry bar.
+        baseline = _sim_baseline_r(fire, bars_after, fidelity) if fire.sim_baseline else fire.baseline_r
+        v, d = iv_flow_persistence(bars, entry_idx, baseline, params)
+        if fire.sim_baseline:
+            d["source"] = "setup_sim"
+        return {"baseline_r": baseline, "variant_r": v, "detail": d}
     if kind == "time_stop":
         v, d = iv_time_stop(fire, bars_after, trigger_high, params, fidelity)
     elif kind == "probe_add":

tests/unit/experiments/test_orderflow_interventions.py

@@ -0,0 +1,114 @@
+"""Order-flow shelf interventions (docs/research/orderflow_notes.md H1/H2).
+
+H1 trigger_erasure: exit on the first post-fill 1-min close back below the
+trigger within a window — the 5-book transitory-impact/recross signature —
+else the standing exit stack (baseline). H2 flow_persistence_gate: keep-or-
+zero on pre-trigger volume-weighted bar-sign persistence (ongoing-metaorder
+signature); fail-open on thin history. Both mirror the regime/dilution
+keep-or-zero discipline: no-data can only KEEP, never zero."""
+
+from datetime import date, datetime
+
+import pandas as pd
+
+from research.experiments.replay import iv_flow_persistence, iv_trigger_erasure, Fire
+
+D = date(2026, 6, 10)
+
+
+def _fire(entry=10.0, stop=9.0, trigger=10.0):
+    return Fire(fire_id="f1", symbol="AAA", session_date=D,
+                fill_ts=datetime(2026, 6, 10, 13, 36), entry_price=entry,
+                stop_price=stop, exit_price=None, baseline_r=None,
+                trigger_price=trigger, sim_baseline=True)
+
+
+def _bars(rows):
+    return pd.DataFrame(rows, columns=["open", "high", "low", "close", "volume"])
+
+
+FID = {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+
+
+# ---------------------------------------------------------------- H1: erasure
+def test_erasure_exits_at_first_close_below_trigger():
+    bars_after = _bars([(10.1, 10.2, 9.85, 9.90, 1000),   # close 9.90 < trigger 10 -> exit here
+                        (9.9, 10.5, 9.8, 10.4, 1000)])
+    v, d = iv_trigger_erasure(_fire(), bars_after, baseline=0.5,
+                              params={"window_bars": 5}, fidelity=FID)
+    # exit fill 9.90 - 0.01 adverse; entry fill 10.00 + 0.01; risk 1.0
+    assert abs(v - ((9.89 - 10.01) / 1.0)) < 1e-9
+    assert d["kind"] == "trigger_erasure" and d["erased"] is True and d["erasure_bar"] == 0
+
+
+def test_no_erasure_keeps_baseline():
+    bars_after = _bars([(10.1, 10.3, 10.05, 10.2, 1000)] * 6)
+    v, d = iv_trigger_erasure(_fire(), bars_after, baseline=0.5,
+                              params={"window_bars": 5}, fidelity=FID)
+    assert v == 0.5 and d["erased"] is False
+
+
+def test_erasure_outside_window_keeps_baseline():
+    rows = [(10.1, 10.3, 10.05, 10.2, 1000)] * 5 + [(10.2, 10.2, 9.5, 9.6, 1000)]
+    v, d = iv_trigger_erasure(_fire(), _bars(rows), baseline=0.5,
+                              params={"window_bars": 5}, fidelity=FID)
+    assert v == 0.5 and d["erased"] is False
+
+
+# ------------------------------------------------------- H2: flow persistence
+def _pre_bars(signs, vol=1000):
+    rows = []
+    px = 10.0
+    for s in signs:
+        o = px
+        c = px + 0.05 * s
+        rows.append((o, max(o, c) + 0.01, min(o, c) - 0.01, c, vol))
+        px = c
+    return _bars(rows)
+
+
+def test_persistent_one_sided_flow_keeps_baseline():
+    bars = _pre_bars([1] * 15)          # all up bars -> persistence 1.0
+    v, d = iv_flow_persistence(bars, entry_idx=15, baseline=0.4,
+                               params={"lookback_bars": 15, "min_persistence": 0.30})
+    assert v == 0.4 and d["flagged"] is False and d["persistence"] > 0.9
+
+
+def test_choppy_flow_is_zeroed():
+    bars = _pre_bars([1, -1] * 8)       # alternating -> persistence ~0
+    v, d = iv_flow_persistence(bars, entry_idx=16, baseline=0.4,
+                               params={"lookback_bars": 15, "min_persistence": 0.30})
+    assert v == 0.0 and d["flagged"] is True
+
+
+def test_thin_history_fails_open():
+    bars = _pre_bars([1, -1, 1])        # only 3 pre-trigger bars -> no-data KEEP
+    v, d = iv_flow_persistence(bars, entry_idx=3, baseline=0.4,
+                               params={"lookback_bars": 15, "min_persistence": 0.30})
+    assert v == 0.4 and d["flagged"] is False and d.get("no_data") is True
+
+
+def test_lookback_excludes_entry_bar():
+    # 15 up bars then a huge DOWN entry bar at idx 15: excluded -> still persistent
+    bars = pd.concat([_pre_bars([1] * 15), _bars([(10.0, 10.0, 8.0, 8.0, 99999)])],
+                     ignore_index=True)
+    v, d = iv_flow_persistence(bars, entry_idx=15, baseline=0.4,
+                               params={"lookback_bars": 15, "min_persistence": 0.30})
+    assert d["flagged"] is False and d["persistence"] > 0.9
+
+
+# --------------------------------------------------------- registration JSONs
+def test_orderflow_experiments_declared():
+    import json, pathlib
+    root = pathlib.Path(__file__).resolve().parents[3] / "experiments"
+    h1 = json.loads((root / "orderflow_h1_erasure_v1.json").read_text())
+    h2 = json.loads((root / "orderflow_h2_persistence_v1.json").read_text())
+    assert h1["intervention"] == {"kind": "trigger_erasure", "window_bars": 5,
+                                  "condition": "close_below"}
+    assert h2["intervention"] == {"kind": "flow_persistence_gate",
+                                  "lookback_bars": 15, "min_persistence": 0.30}
+    for j in (h1, h2):
+        assert j["bars"] == {"expR_delta_min": 0.05, "min_n": 300}
+        assert j["fidelity"] == {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+        assert j["population"] == {"min_entry_gap_pct": 0.10, "min_day_volume": 300000,
+                                   "price_min": 2.0, "price_max": 20.0}