diff --git a/experiments/depth_overhang_v1.json b/experiments/depth_overhang_v1.json new file mode 100644 index 0000000..03f099e --- /dev/null +++ b/experiments/depth_overhang_v1.json @@ -0,0 +1,8 @@ +{ + "id": "depth_overhang_v1", + "hypothesis": "L2 order-book overhang (plan 017): resting ask size within 1% ABOVE the ORB break, normalized to bid depth, from Databento depth. Sniff+reweighted analysis (n=602, 21 sessions) cleared the bar AND the random-gate null: keep fires with overhang>=0.034, zero the rest, lifts win rate 23%->28%. Counter-intuitive direction (more overhang above = MORE likely to win = absorption/fuel). This runs it under the REAL deployed asymmetric exit (simulate_exit, winners trail to +3R) which should exceed the conservative +1/-1 estimate. Coverage: only the 602 depth-backfilled fires have data; the rest SKIP (variant None, excluded from n). Random-gate null applies.", + "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0}, + "intervention": {"kind": "depth_overhang_gate", "min_overhang": 0.034}, + "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 d5e3751..106700a 100644 --- a/research/experiments/replay.py +++ b/research/experiments/replay.py @@ -699,6 +699,41 @@ def iv_quote_imbalance_gate(con, fire: "Fire", baseline: float | None, return baseline, detail +def iv_depth_overhang_gate(fire: "Fire", baseline: float | None, depth: dict, + params: dict) -> tuple[float | None, dict]: + """Plan 017 — the first premarket feature to clear the bar + random-gate + null (n=602, 21 sessions): resting ASK size within 1% ABOVE the ORB break, + normalized to bid depth, from the L2 order book (Databento). Counter to the + supply-trap intuition, MORE overhang predicts WINNING (absorption/fuel — a + break that clears a visible wall ignites; a break into thin air fizzles). + Keep-or-zero: overhang >= min_overhang -> KEEP, else ZERO. A fire absent + from depth_features is SKIPPED (variant None, excluded from n) — the + dilution_veto_v1 no-coverage lesson; never silently kept.""" + thr = float(params.get("min_overhang", 0.05)) + detail = {"kind": "depth_overhang_gate", "flagged": False, "threshold": thr, + "overhang": None} + ov = depth.get(fire.fire_id) + if ov is None: + detail["skipped"] = "no_depth_coverage" + return None, detail + detail["overhang"] = round(float(ov), 4) + if ov < thr: + detail["flagged"] = True + return 0.0, detail + return baseline, detail + + +def _load_depth_overhang(con) -> dict: + """{fire_id: overhang} from depth_features — loaded once per run(). Keyed + 'setup:'+setup_id to match reconstruct_fires_from_setups' fire_id.""" + try: + rows = con.execute("SELECT setup_id, overhang FROM depth_features " + "WHERE overhang IS NOT NULL").fetchall() + except Exception: # noqa: BLE001 — table absent = no coverage, all fires skip + return {} + return {f"setup:{sid}": float(ov) for sid, ov in rows} + + # --------------------------------------------------------------------------- # # Rank (portfolio-level; receipts-only, starves without plan 005) # --------------------------------------------------------------------------- # @@ -732,7 +767,7 @@ def _insert_run(con, exp_id: str, fire: Fire, row: dict) -> None: def _compute_fire(con, exp: dict, fire: Fire, favorable: dict | None = None, - dilution_history: dict | None = None) -> dict | None: + dilution_history: dict | None = None, depth_overhang: dict | None = None) -> dict | None: kind = (exp["intervention"] or {}).get("kind") bars = query_minute_bars(con, fire.symbol, fire.session_date) if bars is None or not len(bars): @@ -766,6 +801,13 @@ 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 == "depth_overhang_gate": + # plan 017: L2 order-book overhang gate (real asymmetric exit baseline). + baseline = _sim_baseline_r(fire, bars_after, fidelity) if fire.sim_baseline else fire.baseline_r + v, d = iv_depth_overhang_gate(fire, baseline, depth_overhang or {}, params) + 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. @@ -1023,9 +1065,12 @@ def run(store, since: str | None = None) -> int: dilution_history = None if kind == "dilution_gate": dilution_history = _load_dilution_history(con) + depth_overhang = None + if kind == "depth_overhang_gate": + depth_overhang = _load_depth_overhang(con) for fire in new_fires: row = _compute_fire(con, exp, fire, favorable=favorable, - dilution_history=dilution_history) + dilution_history=dilution_history, depth_overhang=depth_overhang) if row is None: continue _insert_run(con, exp["id"], fire, row) diff --git a/tests/unit/experiments/test_orderflow_interventions.py b/tests/unit/experiments/test_orderflow_interventions.py index 2cc13a9..0861493 100644 --- a/tests/unit/experiments/test_orderflow_interventions.py +++ b/tests/unit/experiments/test_orderflow_interventions.py @@ -246,3 +246,19 @@ def test_ofi_no_coverage_is_skipped(): v, d = iv_quote_imbalance_gate(store.con, f, baseline=0.4, params={"baseline_minutes": 10, "min_book_imbalance": 0.0}) assert v is None and d.get("skipped") == "no_quote_coverage" + + +# --------------------------------------------------- depth overhang (plan 017) +def test_depth_overhang_gate_zeros_low_keeps_high_skips_missing(): + from research.experiments.replay import iv_depth_overhang_gate + dh = {"setup:A": 0.30, "setup:B": 0.01} # A above thresh, B below + p = {"min_overhang": 0.05} + fa = _fire(); fa = type(fa)(**{**fa.__dict__, "fire_id": "setup:A"}) + fb = _fire(); fb = type(fb)(**{**fb.__dict__, "fire_id": "setup:B"}) + fc = _fire(); fc = type(fc)(**{**fc.__dict__, "fire_id": "setup:C"}) # not in table + va,da = iv_depth_overhang_gate(fa, 0.4, dh, p) + vb,db = iv_depth_overhang_gate(fb, 0.4, dh, p) + vc,dc = iv_depth_overhang_gate(fc, 0.4, dh, p) + assert va == 0.4 and da["flagged"] is False # high overhang -> keep + assert vb == 0.0 and db["flagged"] is True # low overhang -> zero + assert vc is None and dc.get("skipped") == "no_depth_coverage" # missing -> skip n