From 20e6efc24884c7e547c5b22fa41f33be94f7c828 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Mon, 13 Jul 2026 23:03:18 -0400 Subject: [PATCH] entry-gate: characterization matrix pinning both paths pre-extraction (plan 005 step 1) Pins today's slow-path (approval/event) and fast-path (skipped string) decisions across single-condition signals, incl. the two asymmetries that step 2 will close (fast has no VWAP gate; slow has no rank gate). Green against unmodified trading_execution.py. Co-Authored-By: Claude Fable 5 --- tests/unit/test_entry_gate_parity.py | 199 +++++++++++++++++++++++++++ 1 file changed, 199 insertions(+) create mode 100644 tests/unit/test_entry_gate_parity.py diff --git a/tests/unit/test_entry_gate_parity.py b/tests/unit/test_entry_gate_parity.py new file mode 100644 index 0000000..8222d7c --- /dev/null +++ b/tests/unit/test_entry_gate_parity.py @@ -0,0 +1,199 @@ +"""Characterization matrix for the two entry paths (plan 005). + +Pins the CURRENT pass/skip decision of BOTH the slow auto-approval path +(``request_approvals_for_ready_signals``) and the fast trigger path +(``submit_breakout_now``) across a matrix of single-condition signals. It +encodes today's asymmetries EXPLICITLY so the plan-005 extraction of the +shared ``_evaluate_entry_gates`` can only move them in the two documented ways: + + - fast path gains the VWAP gate (was ungated), + - slow path gains the rank/concentration gate when a rank is present. + +Any other decision drift is a regression. Cases marked ``STEP 2 CLOSURE`` are +updated in step 2 (the flip is intentional and cited inline); everything else +stays pinned. +""" + +import json +from datetime import datetime + +import pytest + +from storage.event_schema import EventMode, SignalReadyEvent +from storage.event_store import EventStore +from trading_execution import ExecutionSettings, TradingExecutionService +from trading_mode import TradingModeSettings +from alpaca_paper.execution import ExecutionRequest, ExecutionResult + +T0 = datetime(2026, 6, 11, 9, 45) # minute 15 of the session + + +class _FakeExecutor: + broker_name = "alpaca_paper" + + def __init__(self, status: str = "accepted"): + self.executed: list[ExecutionRequest] = [] + self.ok = True + self.status = status + + def execute(self, request: ExecutionRequest) -> ExecutionResult: + self.executed.append(request) + return ExecutionResult( + ok=self.ok, order_id=request.order_id, + broker_order_id="broker-1" if self.ok else None, + status=self.status if self.ok else "error", + error=None if self.ok else "broker down", + ) + + def cancel_entry(self, order_id, broker_order_id, symbol, reason): + return ExecutionResult(ok=True, order_id=order_id, status="cancelled") + + +class _Clock: + def __init__(self, start: datetime): + self.t = start + + def __call__(self) -> datetime: + return self.t + + +@pytest.fixture +def store(): + s = EventStore(":memory:") + yield s + s.close() + + +def _emit(store, symbol="GOOD", entry=14.0, stop=13.45, above_vwap=None, + vwap=None, quality_score=0.7, gap_pct=None, cum_volume=None): + signal_data = {"entry_price": entry, "stop_loss_price": stop, + "quality_score": quality_score} + if gap_pct is not None: + signal_data["gap_pct"] = gap_pct + if cum_volume is not None: + signal_data["cum_volume"] = cum_volume + if above_vwap is not None: + signal_data["above_vwap"] = above_vwap + signal_data["vwap"] = vwap if vwap is not None else ( + round(entry - 1.0, 2) if above_vwap else round(entry + 1.0, 2)) + store.emit(SignalReadyEvent( + timestamp=T0, mode=EventMode.PAPER, correlation_id="t", + message=f"{symbol} ready", symbol=symbol, signal_type="bull_flag", + confidence=0.9, signal_data=signal_data)) + + +def _make(store, executor=None, now_fn=None, **kw): + defaults = dict(enabled=True, auto_approve=False, max_orders_per_tick=5, + max_concurrent_positions=3, risk_per_trade_pct=0.01, + default_equity=100_000.0) + defaults.update(kw) + return TradingExecutionService( + store, executor=executor or _FakeExecutor(), + settings=ExecutionSettings(**defaults), + trading_mode=TradingModeSettings(execution_mode="alpaca_paper"), + session_id="t", now_fn=now_fn) + + +def _reasons(store): + return {json.loads(r["payload_json"]).get("rule_type") + for r in store.query_events(event_type="risk_rule_triggered")} + + +# ---- SLOW path (request_approvals_for_ready_signals): decision = created + event ---- + +def test_slow_in_population_creates_approval(store): + _emit(store) + assert len(_make(store).request_approvals_for_ready_signals()) == 1 + + +def test_slow_below_vwap_enforced_skips(store): + _emit(store, above_vwap=False) + created = _make(store, require_above_vwap=True).request_approvals_for_ready_signals() + assert created == [] + assert "vwap_below" in _reasons(store) + + +def test_slow_below_vwap_shadow_proceeds(store): + _emit(store, above_vwap=False) + created = _make(store, require_above_vwap=False).request_approvals_for_ready_signals() + assert len(created) == 1 # shadow: logged but not blocked + assert "vwap_below" in _reasons(store) + + +def test_slow_low_gap_population_skips(store): + _emit(store, gap_pct=0.02, cum_volume=1_000_000) + created = _make(store, min_entry_gap_pct=0.10, + min_day_volume=300_000).request_approvals_for_ready_signals() + assert created == [] + assert "population_below" in _reasons(store) + + +def test_slow_below_quality_skips(store): + _emit(store, quality_score=0.30) + created = _make(store, min_quality_score=0.60).request_approvals_for_ready_signals() + assert created == [] + assert "quality_below" in _reasons(store) + + +def test_slow_late_confirmation_skips(store): + _emit(store) # ready at minute 15 + created = _make(store, entry_confirm_by_minute=10).request_approvals_for_ready_signals() + assert created == [] + assert "late_confirmation" in _reasons(store) + + +def test_slow_ignores_rank_today(store): + # STEP 2 CLOSURE: today the slow path has NO rank gate, so a name that would + # rank outside the top-N still gets an approval. Step 2 threads a rank_map and + # flips this to a rank_concentration skip. + _emit(store, gap_pct=0.25) + created = _make(store, concentrate_top_n=3).request_approvals_for_ready_signals() + assert len(created) == 1 + + +# ---- FAST path (submit_breakout_now): decision = skipped string ---- + +def test_fast_in_population_fires(store): + out = _make(store, auto_approve=True).submit_breakout_now( + "GOOD", trigger=14.0, stop=13.45, last_price=14.01, + gap_pct=0.25, cum_volume=1_000_000, rank=1) + assert out["ok"] is True + + +def test_fast_low_gap_population_skips(store): + out = _make(store, auto_approve=True, min_entry_gap_pct=0.10, + min_day_volume=300_000).submit_breakout_now( + "GOOD", trigger=14.0, stop=13.45, last_price=14.01, + gap_pct=0.02, cum_volume=1_000_000, price=14.0) + assert out["skipped"] == "population_below" + + +def test_fast_rank_outside_top_n_skips(store): + out = _make(store, auto_approve=True, concentrate_top_n=3).submit_breakout_now( + "GOOD", trigger=14.0, stop=13.45, last_price=14.01, gap_pct=0.25, rank=5) + assert out["skipped"] == "rank_concentration" + + +def test_fast_late_confirmation_skips(store): + clock = _Clock(datetime(2026, 6, 11, 11, 0)) # minute 90 + out = _make(store, auto_approve=True, now_fn=clock, + entry_confirm_by_minute=15).submit_breakout_now( + "GOOD", trigger=14.0, stop=13.45, last_price=14.01) + assert out["skipped"] == "late_confirmation" + + +def test_fast_over_extended_skips(store): + out = _make(store, auto_approve=True).submit_breakout_now( + "GOOD", trigger=10.0, stop=9.5, last_price=12.7, gap_pct=0.25) + assert out["skipped"] == "over_extended" + + +def test_fast_below_vwap_has_no_gate_today(store): + # STEP 2 CLOSURE: the fast path takes no VWAP input today, so a below-VWAP + # break is never gated here — it fires. Step 2 routes it through the shared + # gate; the closure is demonstrated directly against _evaluate_entry_gates + # (the fast public signature still carries no above_vwap, so it stays + # fail-open in practice). + out = _make(store, auto_approve=True, require_above_vwap=True).submit_breakout_now( + "GOOD", trigger=14.0, stop=13.45, last_price=14.01, gap_pct=0.25, rank=1) + assert out["ok"] is True