fix(slot-filled): 매도 체결 이벤트 NO_ROUTE 해소 — strategy_id 전이 전 캡처
INACTIVE 전이가 strategy_id를 None 초기화(REQ-057) → 라우팅키 slot_filled_sell._.{sym}
→ orb/tsl 바인딩 불일치 NO_ROUTE/DLQ. _on_filled/_on_partial_cancel에서 전이 직전
strategy_id 캡처해 emitter에 전달. wire 계약 불변(포맷 유지·값 교정). QC opus PASS, 747 passed/회귀 0.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5
This commit is contained in:
@@ -9,7 +9,7 @@ from trex_core.adapters.messaging import topology
|
||||
from trex_core.adapters.messaging.in_memory import InMemoryTransport
|
||||
from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
|
||||
from trex_core.application.execution import ExecutionService
|
||||
from trex_core.application.intents import BuyIntent, IntentService
|
||||
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
|
||||
from trex_core.application.ledger_outbox import InMemoryLedgerOutbox, LedgerWriter
|
||||
from trex_core.application.messaging_handlers import slot_transition_event_payload
|
||||
from trex_core.domain.slot import SlotService, SlotState
|
||||
@@ -27,8 +27,11 @@ def _wire(transport):
|
||||
writer = LedgerWriter(outbox, ledger)
|
||||
repo.reconcile_universe("default", {"PF-1": "SYM"})
|
||||
|
||||
def emitter(event_type: str, slot) -> None:
|
||||
rk = topology.event_routing_key(event_type, slot.strategy_id or "_", slot.symbol or "_")
|
||||
def emitter(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
|
||||
# Use routing_strategy_id hint when supplied (e.g. sell fills clear
|
||||
# slot.strategy_id via INACTIVE transition — NO_ROUTE fix).
|
||||
effective_sid = routing_strategy_id or getattr(slot, "strategy_id", None) or "_"
|
||||
rk = topology.event_routing_key(event_type, effective_sid, slot.symbol or "_")
|
||||
payload = slot_transition_event_payload(slot, event_type=event_type)
|
||||
try:
|
||||
transport.publish_event(rk, payload)
|
||||
@@ -124,3 +127,94 @@ def test_publish_failure_does_not_break_execution_loop():
|
||||
assert r.ledger_enqueued == 1
|
||||
pending = [s for s in repo.list_slots("default") if s.status is SlotState.HOLDING]
|
||||
assert len(pending) == 1
|
||||
|
||||
|
||||
def test_sell_fill_emits_slot_filled_sell_with_correct_strategy_routing():
|
||||
"""Regression: slot_filled_sell must route to strategy queue even after
|
||||
PENDING_SELL->INACTIVE clears strategy_id from the slot row.
|
||||
|
||||
M2 REL-009 feedback: slot_filled_sell was NO_ROUTEd (DLQ) because
|
||||
_on_filled used the post-transition slot (strategy_id=None) for the
|
||||
routing key, producing `slot_filled_sell._.{symbol}` which does not match
|
||||
the `*.{strategy_id}.#` binding pattern of trex.events.strategy.{sid}.
|
||||
"""
|
||||
transport = InMemoryTransport()
|
||||
received_sell: list[tuple[str, dict]] = []
|
||||
stop = threading.Event()
|
||||
th = threading.Thread(
|
||||
target=transport.consume_events_forever,
|
||||
args=("trex.events.strategy.A", topology.strategy_event_binding_pattern("A"),
|
||||
lambda rk, p: received_sell.append((rk, p)), stop),
|
||||
daemon=True,
|
||||
)
|
||||
th.start()
|
||||
time.sleep(0.05)
|
||||
|
||||
repo, svc, intents, execution, _ = _wire(transport)
|
||||
# Buy with strategy_id="A"
|
||||
d = intents.submit_buy(BuyIntent(
|
||||
intent_id="b1", qty=10, strategy_id="A", portfolio_id="PF-1",
|
||||
))
|
||||
execution.run_once() # buy fill -> holding
|
||||
slot = repo.fetch_slot(d.slot_id)
|
||||
assert slot.status is SlotState.HOLDING
|
||||
assert slot.strategy_id == "A"
|
||||
|
||||
# Sell — strategy_id_for_emit must be captured before INACTIVE clears it.
|
||||
intents.submit_sell(SellIntent(intent_id="s1", slot_id=d.slot_id))
|
||||
execution.run_once() # sell fill -> inactive
|
||||
|
||||
time.sleep(0.05)
|
||||
stop.set()
|
||||
th.join(timeout=1.0)
|
||||
|
||||
# The slot_filled_sell event must reach the strategy A queue.
|
||||
sell_events = [(rk, p) for rk, p in received_sell if "sell" in rk]
|
||||
assert len(sell_events) == 1, (
|
||||
f"Expected 1 slot_filled_sell event on strategy-A queue, got {len(sell_events)}. "
|
||||
f"All received: {[rk for rk, _ in received_sell]}"
|
||||
)
|
||||
rk, payload = sell_events[0]
|
||||
# Routing key must use strategy_id="A", not "_" (NO_ROUTE sentinel).
|
||||
assert rk == "slot_filled_sell.A.005930", (
|
||||
f"NO_ROUTE regression: routing key is '{rk}', expected 'slot_filled_sell.A.005930'"
|
||||
)
|
||||
assert payload["event_type"] == "slot_filled_sell"
|
||||
|
||||
|
||||
def test_sell_fill_dlq_sentinel_absent_when_strategy_id_set():
|
||||
"""Routing key must NOT contain '_' as strategy_id token when a real
|
||||
strategy_id was set on the slot (no DLQ sentinel in the key)."""
|
||||
transport = InMemoryTransport()
|
||||
all_published: list[str] = []
|
||||
|
||||
def emitter(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
|
||||
effective_sid = routing_strategy_id or getattr(slot, "strategy_id", None) or "_"
|
||||
rk = topology.event_routing_key(event_type, effective_sid, slot.symbol or "_")
|
||||
all_published.append(rk)
|
||||
payload = slot_transition_event_payload(slot, event_type=event_type)
|
||||
try:
|
||||
transport.publish_event(rk, payload)
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
intents = IntentService(svc, repo, InMemoryIntentStore(),
|
||||
resolve_symbol=lambda pid: "005930")
|
||||
from trex_core.application.ledger_outbox import InMemoryLedgerOutbox
|
||||
outbox = InMemoryLedgerOutbox()
|
||||
execution = ExecutionService(svc, repo, PaperBroker(fill_after_polls=0),
|
||||
ledger_outbox=outbox, event_emitter=emitter)
|
||||
repo.reconcile_universe("default", {"PF-1": "SYM"})
|
||||
|
||||
d = intents.submit_buy(BuyIntent(intent_id="b1", qty=10, strategy_id="orb-v1", portfolio_id="PF-1"))
|
||||
execution.run_once()
|
||||
intents.submit_sell(SellIntent(intent_id="s1", slot_id=d.slot_id))
|
||||
execution.run_once()
|
||||
|
||||
# Neither buy nor sell events should use "_" as strategy token.
|
||||
bad_keys = [rk for rk in all_published if "._.".encode() in rk.encode() or rk.split(".")[1] == "_"]
|
||||
assert bad_keys == [], (
|
||||
f"DLQ-sentinel routing keys found (strategy_id=None leak): {bad_keys}"
|
||||
)
|
||||
|
||||
@@ -41,7 +41,14 @@ _log = logging.getLogger(__name__)
|
||||
# slot transition events through MessagingTransport. Application layer stays
|
||||
# transport-agnostic (NFR-1). Slot type is duck-typed to keep the application
|
||||
# layer free of domain re-imports (the slot is just used for attribute access).
|
||||
SlotEventEmitter = Callable[[str, object], None] # (event_type, slot) -> None
|
||||
#
|
||||
# The optional third argument `routing_strategy_id` is a hint for the routing
|
||||
# key when the slot's own strategy_id has been cleared by the transition (e.g.
|
||||
# PENDING_SELL -> INACTIVE clears strategy_id on the slot row — REQ-057
|
||||
# _apply_transition_fields). The emitter MUST prefer this hint over
|
||||
# `slot.strategy_id` so that slot_filled_sell events reach the correct
|
||||
# per-strategy queue instead of being NO_ROUTEd to the DLQ.
|
||||
SlotEventEmitter = Callable[[str, object, Optional[str]], None] # (event_type, slot, routing_strategy_id) -> None
|
||||
|
||||
_MAX_DISPATCH_PER_TICK = 64
|
||||
|
||||
@@ -288,6 +295,13 @@ class ExecutionService:
|
||||
|
||||
def _on_filled(self, slot, fs, side: str, r: ExecutionResult) -> None:
|
||||
broker_order_id = slot.broker_order_id # captured BEFORE clear_dispatch
|
||||
# Capture strategy_id BEFORE the transition clears it.
|
||||
# PENDING_SELL -> INACTIVE clears strategy_id on the slot row
|
||||
# (_apply_transition_fields REQ-057), so we must capture it here to
|
||||
# supply as a routing hint for the slot_filled_sell event — otherwise the
|
||||
# routing key uses "_" and the message is NO_ROUTEd to the DLQ instead
|
||||
# of reaching trex.events.strategy.{strategy_id} (M2 REL-009 feedback).
|
||||
routing_strategy_id = slot.strategy_id
|
||||
# REQ-095: resolve fill price with limit_price backfill guard.
|
||||
fill_price, _was_backfilled = self._resolve_fill_price(fs, slot)
|
||||
if fill_price is None:
|
||||
@@ -342,7 +356,9 @@ class ExecutionService:
|
||||
self._record_ledger(slot, fs, side, broker_order_id, r, resolved_price=fill_price)
|
||||
# REQ-021/023: emit event AFTER transition + ledger enqueue (single
|
||||
# record subject). Best-effort: a publish failure must not roll back.
|
||||
self._emit_event(event_type, updated)
|
||||
# Pass routing_strategy_id so sell fills route correctly even after
|
||||
# strategy_id is cleared from the slot row (slot_filled_sell NO_ROUTE fix).
|
||||
self._emit_event(event_type, updated, routing_strategy_id)
|
||||
self._repo.clear_dispatch(slot.slot_id) # done with this order
|
||||
|
||||
def _on_partial_cancel(self, slot, fs, side: str, r: ExecutionResult) -> None:
|
||||
@@ -355,6 +371,9 @@ class ExecutionService:
|
||||
portion that actually filled.
|
||||
"""
|
||||
broker_order_id = slot.broker_order_id # captured BEFORE clear_dispatch
|
||||
# Capture strategy_id BEFORE the transition (same reason as _on_filled:
|
||||
# PENDING_SELL -> INACTIVE clears strategy_id, causing NO_ROUTE).
|
||||
routing_strategy_id = slot.strategy_id
|
||||
# REQ-095: resolve fill price with limit_price backfill guard.
|
||||
fill_price, _was_backfilled = self._resolve_fill_price(fs, slot)
|
||||
if fill_price is None:
|
||||
@@ -395,7 +414,7 @@ class ExecutionService:
|
||||
except SlotStateMismatchError:
|
||||
return # slot moved concurrently; next tick reconciles
|
||||
self._record_ledger(slot, fs, side, broker_order_id, r, resolved_price=fill_price)
|
||||
self._emit_event(event_type, updated)
|
||||
self._emit_event(event_type, updated, routing_strategy_id)
|
||||
self._repo.clear_dispatch(slot.slot_id)
|
||||
|
||||
def _check_stale_pending(self, slot, now: datetime, r: ExecutionResult) -> None:
|
||||
@@ -454,11 +473,11 @@ class ExecutionService:
|
||||
slot.slot_id, slot.broker_order_id,
|
||||
)
|
||||
|
||||
def _emit_event(self, event_type: str, slot) -> None:
|
||||
def _emit_event(self, event_type: str, slot, routing_strategy_id: Optional[str] = None) -> None:
|
||||
if self._event_emitter is None:
|
||||
return
|
||||
try:
|
||||
self._event_emitter(event_type, slot)
|
||||
self._event_emitter(event_type, slot, routing_strategy_id)
|
||||
except Exception: # noqa: BLE001 - event publish failure is non-critical
|
||||
return
|
||||
|
||||
|
||||
@@ -264,10 +264,18 @@ class Daemon:
|
||||
|
||||
# ------------------------------------------------------------ event hook
|
||||
def _make_event_emitter(self):
|
||||
def emit(event_type: str, slot) -> None:
|
||||
def emit(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
|
||||
payload = slot_transition_event_payload(slot, event_type=event_type)
|
||||
# routing_strategy_id is supplied when the transition clears
|
||||
# slot.strategy_id (e.g. PENDING_SELL -> INACTIVE for slot_filled_sell).
|
||||
# Without this hint the routing key would use "_" and the message would
|
||||
# be NO_ROUTEd to the DLQ instead of reaching the strategy queue
|
||||
# (M2 REL-009 feedback / slot_filled NO_ROUTE fix).
|
||||
effective_strategy_id = (
|
||||
routing_strategy_id or getattr(slot, "strategy_id", None) or "_"
|
||||
)
|
||||
rk = topology.event_routing_key(
|
||||
event_type, slot.strategy_id or "_", slot.symbol or "_",
|
||||
event_type, effective_strategy_id, slot.symbol or "_",
|
||||
)
|
||||
try:
|
||||
self.event_publisher.publish_event(rk, payload)
|
||||
|
||||
Reference in New Issue
Block a user