From 07ddbd30b958785ad2b31be9ad74a5642019706f Mon Sep 17 00:00:00 2001 From: momentum dev Date: Fri, 17 Jul 2026 18:32:10 -0500 Subject: [PATCH] experiments: register orderflow H3 (spread-expansion gate) + H4 (volume-spike gate) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- experiments/orderflow_h3_spread_v1.json | 8 ++ experiments/orderflow_h4_spike_v1.json | 8 ++ research/experiments/replay.py | 95 +++++++++++++++++++ .../test_orderflow_interventions.py | 82 ++++++++++++++++ 4 files changed, 193 insertions(+) create mode 100644 experiments/orderflow_h3_spread_v1.json create mode 100644 experiments/orderflow_h4_spike_v1.json diff --git a/experiments/orderflow_h3_spread_v1.json b/experiments/orderflow_h3_spread_v1.json new file mode 100644 index 0000000..741ce1e --- /dev/null +++ b/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} +} diff --git a/experiments/orderflow_h4_spike_v1.json b/experiments/orderflow_h4_spike_v1.json new file mode 100644 index 0000000..162bc69 --- /dev/null +++ b/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} +} diff --git a/research/experiments/replay.py b/research/experiments/replay.py index f107b5b..77200ad 100644 --- a/research/experiments/replay.py +++ b/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": diff --git a/tests/unit/experiments/test_orderflow_interventions.py b/tests/unit/experiments/test_orderflow_interventions.py index 47c4116..263b0f2 100644 --- a/tests/unit/experiments/test_orderflow_interventions.py +++ b/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"