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 <noreply@anthropic.com>
2026-07-13 23:03:27 -04:00
1 parent 124d816cb9
commit 20e6efc248
@@ -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