experiments: register orderflow H3 (spread-expansion gate) + H4 (volume-spike gate)

H4 (3-book): keep-or-zero on trigger-bar volume concentration vs prior 5-bar
sum (>1.0 = completed one-shot burst). H3 (5-book, direction disputed):
fragility-camp primary — zero fires whose fill-time relative spread exceeds
2x the name's own 10-min median from quote_snapshots (which carry bid/ask
SIZES since 07-02 — the note's data-upgrade #1 was already satisfied).
Coverage design learns the dilution_veto_v1 lesson: no-quote fires are
SKIPPED (variant None, excluded from n), so thin coverage cannot false-FAIL.
H5 deliberately deferred until H1's verdict (its incremental value is defined
net of H1's catches). 6 tests.

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

experiments/orderflow_h3_spread_v1.json

@@ -0,0 +1,8 @@
+{
+  "id": "orderflow_h3_spread_v1",
+  "hypothesis": "Spread-expansion at trigger (orderflow_notes.md H3, 5 books, direction disputed): PRIMARY = fragility camp (Bouchaud/CKS/Bellafiore) — relative spread at fill above 2.0x the name's own 10-min pre-trigger median (liquidity vacuum) marks fragile breakouts; zeroing them lifts kept expR. Two-sidedness handled by recording detail.expansion_ratio per fire: a strongly NEGATIVE delta here supports registering the flipped (adverse-selection) v2. COVERAGE: fires without >=5 usable pre-trigger snapshots are SKIPPED (variant None, excluded from n — the dilution_veto_v1 false-FAIL lesson); quote_snapshots exist since 2026-07-02, so n accrues ~30 fires/day and min_n is reached in ~2 weeks.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "spread_expansion_gate", "baseline_minutes": 10, "max_expansion_ratio": 2.0},
+  "bars": {"expR_delta_min": 0.05, "min_n": 300},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

experiments/orderflow_h4_spike_v1.json

@@ -0,0 +1,8 @@
+{
+  "id": "orderflow_h4_spike_v1",
+  "hypothesis": "Volume ramp vs isolated spike (orderflow_notes.md H4, 3-book convergence): fires whose trigger-bar volume exceeds the sum of the prior 5 bars (concentration > 1.0 = completed one-shot burst; Bouchaud: impact reverts once the metaorder is done; Neill: the vertical spike that dies at the top) convert worse; zeroing them lifts kept expR. detail.concentration recorded per fire for tercile-monotonicity and RVOL-stratification checks at adjudication.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "volume_spike_gate", "lookback_bars": 5, "max_concentration": 1.0},
+  "bars": {"expR_delta_min": 0.05, "min_n": 300},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

research/experiments/replay.py

@@ -577,6 +577,87 @@ def iv_flow_persistence(bars, entry_idx: int, baseline: float | None,
     return baseline, detail
 
 
+def iv_volume_spike_gate(bars, entry_idx: int, baseline: float | None,
+                         params: dict) -> tuple[float | None, dict]:
+    """H4 (Bouchaud/Neill/Hasbrouck): keep-or-zero on trigger-bar volume
+    concentration — trigger_vol / sum(prior ``lookback_bars`` vols). A ratio
+    above ``max_concentration`` is the isolated one-shot burst (completed
+    metaorder / short-covering spike) whose impact reverts; a ramp is an
+    unfinished metaorder still executing. Strictly pre/at-trigger — no
+    post-entry volume conditioning (distinct from the refuted volume-fade).
+    Thin history / zero volume fails OPEN."""
+    lookback = int(params.get("lookback_bars", 5))
+    max_c = float(params.get("max_concentration", 1.0))
+    detail = {"kind": "volume_spike_gate", "flagged": False,
+              "lookback_bars": lookback, "threshold": max_c, "concentration": None}
+    if bars is None or entry_idx < lookback:
+        detail["no_data"] = True
+        return baseline, detail
+    pre_vol = float(bars.iloc[entry_idx - lookback:entry_idx]["volume"].fillna(0).sum())
+    trig_vol = float(bars.iloc[entry_idx]["volume"] or 0)
+    if pre_vol <= 0:
+        detail["no_data"] = True
+        return baseline, detail
+    concentration = trig_vol / pre_vol
+    detail["concentration"] = round(concentration, 4)
+    if concentration > max_c:
+        detail["flagged"] = True
+        return 0.0, detail
+    return baseline, detail
+
+
+def iv_spread_expansion_gate(con, fire: "Fire", baseline: float | None,
+                             params: dict) -> tuple[float | None, dict]:
+    """H3 (5 books, direction disputed): the relative spread at trigger vs the
+    name's OWN 10-min pre-trigger baseline. Primary registered direction is
+    the practitioner/fragility camp (Bouchaud/CKS/Bellafiore): expansion_ratio
+    above ``max_expansion_ratio`` = liquidity vacuum -> ZERO. The ratio is
+    recorded per fire so a reversed (adverse-selection) signal shows up in the
+    accumulated detail and can be registered as a flipped v2.
+
+    COVERAGE DESIGN (the dilution_veto_v1 lesson): quote_snapshots only exist
+    for watched names since 2026-07-02 — a fire WITHOUT a usable pre-trigger
+    baseline is SKIPPED (variant None, excluded from n), never silently kept,
+    so the experiment cannot false-FAIL on thin coverage."""
+    from zoneinfo import ZoneInfo
+
+    minutes = int(params.get("baseline_minutes", 10))
+    max_r = float(params.get("max_expansion_ratio", 2.0))
+    detail = {"kind": "spread_expansion_gate", "flagged": False,
+              "baseline_minutes": minutes, "threshold": max_r, "expansion_ratio": None}
+    try:
+        ts = fire.fill_ts if isinstance(fire.fill_ts, datetime) \
+            else datetime.fromisoformat(str(fire.fill_ts))
+        fill_utc = ts.replace(tzinfo=ZoneInfo("America/New_York")) if ts.tzinfo is None else ts
+        from datetime import timedelta
+        rows = con.execute(
+            "SELECT bid_price, ask_price FROM quote_snapshots "
+            "WHERE symbol = ? AND ts >= ? AND ts <= ? ORDER BY ts",
+            [fire.symbol, fill_utc - timedelta(minutes=minutes), fill_utc]).fetchall()
+    except Exception:  # noqa: BLE001 — table absent / query fault = no coverage
+        detail["skipped"] = "no_quote_coverage"
+        return None, detail
+    rels = []
+    for bid, ask in rows:
+        if bid and ask and ask > bid > 0:
+            mid = (float(ask) + float(bid)) / 2.0
+            rels.append((float(ask) - float(bid)) / mid)
+    if len(rels) < 5:
+        detail["skipped"] = "no_quote_coverage"
+        return None, detail
+    base_rels = sorted(rels[:-1])
+    baseline_spread = base_rels[len(base_rels) // 2]
+    if baseline_spread <= 0:
+        detail["skipped"] = "no_quote_coverage"
+        return None, detail
+    ratio = rels[-1] / baseline_spread
+    detail["expansion_ratio"] = round(ratio, 4)
+    if ratio > max_r:
+        detail["flagged"] = True
+        return 0.0, detail
+    return baseline, detail
+
+
 # --------------------------------------------------------------------------- #
 # Rank (portfolio-level; receipts-only, starves without plan 005)
 # --------------------------------------------------------------------------- #
@@ -659,6 +740,20 @@ 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 == "volume_spike_gate":
+        # orderflow H4: trigger-bar volume concentration vs the prior ramp.
+        baseline = _sim_baseline_r(fire, bars_after, fidelity) if fire.sim_baseline else fire.baseline_r
+        v, d = iv_volume_spike_gate(bars, entry_idx, baseline, params)
+        if fire.sim_baseline:
+            d["source"] = "setup_sim"
+        return {"baseline_r": baseline, "variant_r": v, "detail": d}
+    if kind == "spread_expansion_gate":
+        # orderflow H3: quote-snapshot spread ratio; no coverage -> SKIP (None).
+        baseline = _sim_baseline_r(fire, bars_after, fidelity) if fire.sim_baseline else fire.baseline_r
+        v, d = iv_spread_expansion_gate(con, fire, 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

@@ -112,3 +112,85 @@ def test_orderflow_experiments_declared():
         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}
+
+
+# ------------------------------------------------- H4: volume concentration
+def test_isolated_spike_trigger_is_zeroed():
+    from research.experiments.replay import iv_volume_spike_gate
+    rows = [(10, 10.1, 9.9, 10, 100)] * 5 + [(10, 10.5, 10, 10.4, 5000)]  # spike bar
+    v, d = iv_volume_spike_gate(_bars(rows), entry_idx=5, baseline=0.4,
+                                params={"lookback_bars": 5, "max_concentration": 1.0})
+    assert v == 0.0 and d["flagged"] is True and d["concentration"] == 10.0
+
+
+def test_volume_ramp_trigger_is_kept():
+    from research.experiments.replay import iv_volume_spike_gate
+    rows = [(10, 10.1, 9.9, 10, v) for v in (500, 800, 1200, 1800, 2500)] + [(10, 10.5, 10, 10.4, 2000)]
+    v, d = iv_volume_spike_gate(_bars(rows), entry_idx=5, baseline=0.4,
+                                params={"lookback_bars": 5, "max_concentration": 1.0})
+    assert v == 0.4 and d["flagged"] is False
+
+
+def test_spike_gate_thin_history_fails_open():
+    from research.experiments.replay import iv_volume_spike_gate
+    rows = [(10, 10.1, 9.9, 10, 100), (10, 10.5, 10, 10.4, 5000)]
+    v, d = iv_volume_spike_gate(_bars(rows), entry_idx=1, baseline=0.4,
+                                params={"lookback_bars": 5, "max_concentration": 1.0})
+    assert v == 0.4 and d.get("no_data") is True
+
+
+# ------------------------------------------------- H3: spread expansion
+def _seed_quotes(store, symbol, rows):
+    store.con.execute(
+        "CREATE TABLE IF NOT EXISTS quote_snapshots (symbol VARCHAR, ts TIMESTAMPTZ DEFAULT now(), "
+        "bid_price DOUBLE PRECISION, bid_size BIGINT, ask_price DOUBLE PRECISION, ask_size BIGINT)")
+    for ts, bid, ask in rows:
+        store.con.execute(
+            "INSERT INTO quote_snapshots (symbol, ts, bid_price, bid_size, ask_price, ask_size) "
+            "VALUES (?,?,?,?,?,?)", [symbol, ts, bid, 100, ask, 100])
+
+
+def test_spread_blowout_at_trigger_is_zeroed():
+    from datetime import timezone, timedelta
+    from storage.event_store import EventStore
+    from research.experiments.replay import iv_spread_expansion_gate
+    store = EventStore(":memory:")
+    fill_naive_et = datetime(2026, 6, 10, 9, 40)          # naive ET fill
+    fill_utc = datetime(2026, 6, 10, 13, 40, tzinfo=timezone.utc)
+    quotes = [(fill_utc - timedelta(seconds=15 * (i + 2)), 10.00, 10.02) for i in range(8)]
+    quotes.append((fill_utc - timedelta(seconds=5), 9.95, 10.15))  # last: spread 10x baseline
+    _seed_quotes(store, "AAA", quotes)
+    f = _fire()
+    f = type(f)(**{**f.__dict__, "fill_ts": fill_naive_et})
+    v, d = iv_spread_expansion_gate(store.con, f, baseline=0.4,
+                                    params={"baseline_minutes": 10, "max_expansion_ratio": 2.0})
+    assert v == 0.0 and d["flagged"] is True and d["expansion_ratio"] > 2.0
+
+
+def test_spread_normal_at_trigger_is_kept():
+    from datetime import timezone, timedelta
+    from storage.event_store import EventStore
+    from research.experiments.replay import iv_spread_expansion_gate
+    store = EventStore(":memory:")
+    fill_utc = datetime(2026, 6, 10, 13, 40, tzinfo=timezone.utc)
+    quotes = [(fill_utc - timedelta(seconds=15 * (i + 1)), 10.00, 10.02) for i in range(9)]
+    _seed_quotes(store, "AAA", quotes)
+    f = _fire()
+    f = type(f)(**{**f.__dict__, "fill_ts": datetime(2026, 6, 10, 9, 40)})
+    v, d = iv_spread_expansion_gate(store.con, f, baseline=0.4,
+                                    params={"baseline_minutes": 10, "max_expansion_ratio": 2.0})
+    assert v == 0.4 and d["flagged"] is False
+
+
+def test_no_quote_coverage_is_skipped_not_kept():
+    """The dilution-v1 lesson: fires WITHOUT feature coverage must be EXCLUDED
+    from n (variant None -> evaluate() skips), never silently counted."""
+    from storage.event_store import EventStore
+    from research.experiments.replay import iv_spread_expansion_gate
+    store = EventStore(":memory:")
+    _seed_quotes(store, "OTHER", [])   # table exists, no rows for AAA
+    f = _fire()
+    f = type(f)(**{**f.__dict__, "fill_ts": datetime(2026, 6, 10, 9, 40)})
+    v, d = iv_spread_expansion_gate(store.con, f, baseline=0.4,
+                                    params={"baseline_minutes": 10, "max_expansion_ratio": 2.0})
+    assert v is None and d.get("skipped") == "no_quote_coverage"