momentum devandClaude Fable 5
b6a4040e45
New EntryGateEvaluatedEvent (symbol, source, passed, reject_reason, gates map) emitted once per _evaluate_entry_gates decision on both paths, behind TRADING_GATE_RECEIPTS (default on; dataclass default off so unit tests opt in). Added to the 14-day nightly purge tuple (high-volume diagnostic) and .env.example. Turns 'why didn't it trade X?' into a query and is the substrate plan 006 replays. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""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 trading_execution import EntryContext
|
|
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_rank_gate_closure(store):
|
|
# STEP 2 CLOSURE (plan 005): the slow path now honours concentrate-by-rank when
|
|
# a rank_map is threaded in from the trigger book. A name ranked outside the
|
|
# top-N is rejected — before extraction the slow path had no rank gate and this
|
|
# same call created an approval (see git history of this test).
|
|
_emit(store, symbol="RANKED", gap_pct=0.25)
|
|
svc = _make(store, concentrate_top_n=3)
|
|
created = svc.request_approvals_for_ready_signals(rank_map={"RANKED": 5})
|
|
assert created == []
|
|
|
|
|
|
def test_slow_rank_within_top_n_still_creates(store):
|
|
# A ranked name INSIDE the top-N still gets an approval (rank-1 = full size).
|
|
_emit(store, symbol="RANKED", gap_pct=0.25)
|
|
svc = _make(store, concentrate_top_n=3)
|
|
created = svc.request_approvals_for_ready_signals(rank_map={"RANKED": 1})
|
|
assert len(created) == 1
|
|
|
|
|
|
def test_fast_vwap_gate_closure(store):
|
|
# STEP 2 CLOSURE (plan 005): the fast path now routes through the shared gate,
|
|
# which contains the VWAP check. Demonstrated directly on _evaluate_entry_gates
|
|
# with source="fast" (the fast public signature carries no above_vwap, so the
|
|
# closure is only reachable when a below-VWAP value is supplied). Before
|
|
# extraction the fast path had no VWAP gate at all.
|
|
svc = _make(store, require_above_vwap=True)
|
|
ctx = EntryContext(symbol="X", source="fast", entry=14.0, stop=13.45,
|
|
trigger=14.0, entry_ref=14.0, gap_pct=0.25, above_vwap=False,
|
|
vwap=15.0, confirm_ts=T0)
|
|
decision = svc._evaluate_entry_gates(ctx)
|
|
assert decision.passed is False
|
|
assert decision.reason == "vwap_below"
|
|
|
|
|
|
def test_gate_parity_below_vwap_enforced_both_paths(store):
|
|
# With above_vwap supplied, BOTH sources now reach the same vwap_below verdict —
|
|
# the asymmetry is closed. (The auto source also emits the shadow/skip event;
|
|
# the fast source stays event-free, only the decision unifies.)
|
|
from trading_execution import EntryContext
|
|
svc = _make(store, require_above_vwap=True)
|
|
for src in ("auto", "fast"):
|
|
d = svc._evaluate_entry_gates(EntryContext(
|
|
symbol="X", source=src, entry=14.0, stop=13.45, trigger=14.0,
|
|
entry_ref=14.0, gap_pct=0.25, above_vwap=False, vwap=15.0, confirm_ts=T0))
|
|
assert d.reason == "vwap_below", src
|
|
|
|
|
|
# ---- decision receipts (step 3) ----
|
|
|
|
def _receipts(store):
|
|
return [json.loads(r["payload_json"])
|
|
for r in store.query_events(event_type="entry_gate_evaluated")]
|
|
|
|
|
|
def _ctx(symbol="X", source="fast", **kw):
|
|
base = dict(symbol=symbol, source=source, entry=14.0, stop=13.45,
|
|
trigger=14.0, entry_ref=14.0, gap_pct=0.25, confirm_ts=T0)
|
|
base.update(kw)
|
|
return EntryContext(**base)
|
|
|
|
|
|
def test_receipt_on_reject_carries_reason_and_gate_map(store):
|
|
svc = _make(store, gate_receipts=True, min_entry_gap_pct=0.10, min_day_volume=300_000)
|
|
d = svc._evaluate_entry_gates(_ctx(gap_pct=0.02, cum_volume=1_000_000, price=14.0))
|
|
assert d.passed is False
|
|
r = _receipts(store)
|
|
assert len(r) == 1
|
|
assert r[0]["passed"] is False
|
|
assert r[0]["reject_reason"] == "population_below"
|
|
assert r[0]["gates"]["population"]["outcome"] == "reject"
|
|
|
|
|
|
def test_receipt_on_pass_records_gate_map(store):
|
|
svc = _make(store, gate_receipts=True)
|
|
d = svc._evaluate_entry_gates(_ctx(rank=1))
|
|
assert d.passed is True
|
|
r = _receipts(store)
|
|
assert len(r) == 1
|
|
assert r[0]["passed"] is True
|
|
assert r[0]["reject_reason"] is None
|
|
assert r[0]["gates"]["population"]["outcome"] == "pass"
|
|
|
|
|
|
def test_receipt_off_by_flag_emits_nothing(store):
|
|
svc = _make(store, gate_receipts=False, min_entry_gap_pct=0.10)
|
|
svc._evaluate_entry_gates(_ctx(gap_pct=0.02, cum_volume=1_000_000, price=14.0))
|
|
assert _receipts(store) == []
|
|
|
|
|
|
# ---- 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
|