feat: propagate intent venue through execution

2026-07-29 16:55:20 +09:00
parent 7084304c9f
commit 479cb39a7e
22 changed files with 314 additions and 19 deletions
@@ -0,0 +1,152 @@
+"""Venue propagation from the strategy wire contract to broker execution."""
+from __future__ import annotations
+
+from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
+from trex_core.application.broker_port import BalanceItem, FillStatus, OrderAck, OrderRequest
+from trex_core.application.execution import ExecutionService
+from trex_core.application.intents import IntentService
+from trex_core.application.messaging import RpcRequest
+from trex_core.application.messaging_handlers import make_intent_rpc_handler
+from trex_core.domain.slot import SlotService, SlotState
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
+
+
+class RecordingBroker:
+    def __init__(self) -> None:
+        self.orders: list[OrderRequest] = []
+        self.cancel_venues: list[str] = []
+
+    def submit_order(self, req: OrderRequest) -> OrderAck:
+        self.orders.append(req)
+        return OrderAck(accepted=True, broker_order_id=f"order-{len(self.orders)}")
+
+    def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus:
+        return FillStatus(state="accepted")
+
+    def cancel_order(
+        self,
+        broker_order_id: str,
+        *,
+        side: str,
+        symbol: str,
+        qty: int,
+        venue: str = "KRX",
+    ) -> bool:
+        self.cancel_venues.append(venue)
+        return True
+
+    def get_balance(self) -> list[BalanceItem]:
+        return []
+
+
+def _wire() -> tuple[
+    InMemorySlotRepository,
+    IntentService,
+    ExecutionService,
+    RecordingBroker,
+]:
+    repo = InMemorySlotRepository()
+    service = SlotService(repo)
+    intents = IntentService(
+        service,
+        repo,
+        InMemoryIntentStore(),
+        resolve_symbol=lambda portfolio_id: "005930",
+    )
+    repo.reconcile_universe("default", {"PF-1": "005930"})
+    broker = RecordingBroker()
+    return repo, intents, ExecutionService(service, repo, broker), broker
+
+
+def test_nxt_buy_wire_value_reaches_broker_order() -> None:
+    repo, intents, execution, broker = _wire()
+    handler = make_intent_rpc_handler(intents)
+
+    response = handler(
+        RpcRequest(
+            payload={
+                "intent_type": "buy",
+                "intent_id": "nxt-buy",
+                "portfolio_id": "PF-1",
+                "qty": 1,
+                "venue": "NXT",
+            },
+            correlation_id="nxt-buy",
+        )
+    )
+
+    assert response.payload["status"] == "admitted"
+    pending = next(slot for slot in repo.list_slots("default") if slot.status is SlotState.PENDING_BUY)
+    assert pending.venue == "NXT"
+    execution.run_once()
+    assert broker.orders[0].venue == "NXT"
+
+
+def test_invalid_venue_is_rejected_before_slot_transition() -> None:
+    repo, intents, _, _ = _wire()
+    handler = make_intent_rpc_handler(intents)
+
+    response = handler(
+        RpcRequest(
+            payload={
+                "intent_type": "buy",
+                "intent_id": "bad-venue",
+                "portfolio_id": "PF-1",
+                "qty": 1,
+                "venue": "UNKNOWN",
+            },
+            correlation_id="bad-venue",
+        )
+    )
+
+    assert response.payload["status"] == "rejected"
+    assert response.payload["reason"] == "invalid_venue"
+    assert repo.list_slots("default")[0].status is SlotState.INACTIVE
+
+
+def test_inactive_transition_resets_venue_to_krx() -> None:
+    repo, intents, _, _ = _wire()
+    handler = make_intent_rpc_handler(intents)
+    handler(
+        RpcRequest(
+            payload={
+                "intent_type": "buy",
+                "intent_id": "nxt-buy-reset",
+                "portfolio_id": "PF-1",
+                "qty": 1,
+                "venue": "NXT",
+            },
+            correlation_id="nxt-buy-reset",
+        )
+    )
+    slot = repo.list_slots("default")[0]
+    service = SlotService(repo)
+    reset = service.transit_slot(
+        slot.slot_id,
+        SlotState.PENDING_BUY,
+        SlotState.INACTIVE,
+        reason="rollback",
+        actor="test",
+    )
+    assert reset.venue == "KRX"
+
+
+def test_operator_cancel_uses_original_nxt_venue() -> None:
+    repo, intents, execution, broker = _wire()
+    handler = make_intent_rpc_handler(intents)
+    response = handler(
+        RpcRequest(
+            payload={
+                "intent_type": "buy",
+                "intent_id": "nxt-buy-cancel",
+                "portfolio_id": "PF-1",
+                "qty": 1,
+                "venue": "NXT",
+            },
+            correlation_id="nxt-buy-cancel",
+        )
+    )
+    execution.run_once()
+
+    assert execution.cancel_pending_slot(response.payload["slot_id"]) == "ok"
+    assert broker.cancel_venues == ["NXT"]