"""Order execution service (REQ-013, REQ-014, dcy-003). A single background execution loop (invariant A single writer, invariant B single broker session) that, per tick: dispatch: claim pending slots with broker_order_id IS NULL (option A queue), submit the order, record broker_order_id (idempotency guard). poll: observe fills on dispatched pending slots and transition the slot (pending_buy->holding / pending_sell->closed), or roll back on reject. reconcile() re-polls dispatched slots on restart so a crash after dispatch converges without duplicate orders (REQ-014). All slot writes go through SlotService (advisory lock). The broker is touched only here, never by the domain or the HTTP API. """ from __future__ import annotations from dataclasses import dataclass from trex_core.application.broker_port import BrokerPort, OrderRequest from trex_core.domain.slot import SlotService, SlotState, SlotStateMismatchError _MAX_DISPATCH_PER_TICK = 64 @dataclass class ExecutionResult: dispatched: int = 0 filled: int = 0 rolled_back: int = 0 class ExecutionService: def __init__(self, slot_service: SlotService, repo, broker: BrokerPort) -> None: self._svc = slot_service self._repo = repo self._broker = broker # ------------------------------------------------------------ def run_once(self) -> ExecutionResult: r = ExecutionResult() self._dispatch(r) self._poll(r) return r def reconcile(self) -> ExecutionResult: """Restart recovery: re-poll dispatched-but-pending slots (REQ-014). Undispatched pending slots (broker_order_id IS NULL) are simply picked up by the next dispatch, so a crash before dispatch causes no duplicate. """ r = ExecutionResult() self._poll(r) return r # ------------------------------------------------------------ def _dispatch(self, r: ExecutionResult) -> None: for _ in range(_MAX_DISPATCH_PER_TICK): slot = self._repo.find_pending_for_dispatch() if slot is None: return side = "buy" if slot.status is SlotState.PENDING_BUY else "sell" qty = int(slot.qty or 0) limit = int(slot.entry_price) if slot.entry_price else None ack = self._broker.submit_order( OrderRequest( slot_id=str(slot.slot_id), side=side, symbol=slot.symbol or "", qty=qty, limit_price=limit, order_type="00" if limit is not None else "01", ) ) if ack.accepted and ack.broker_order_id: self._repo.mark_dispatched(slot.slot_id, ack.broker_order_id) r.dispatched += 1 else: self._rollback(slot, reason=ack.reason or "order_rejected") r.rolled_back += 1 def _poll(self, r: ExecutionResult) -> None: for slot in self._repo.list_dispatched_pending(): side = "buy" if slot.status is SlotState.PENDING_BUY else "sell" fs = self._broker.poll_fill(slot.broker_order_id, side=side, symbol=slot.symbol or "") if fs.is_filled: self._on_filled(slot, fs) r.filled += 1 elif fs.is_terminal_reject: self._rollback(slot, reason=f"order_{fs.state}") r.rolled_back += 1 # accepted / partially_filled -> leave for next tick # ------------------------------------------------------------ def _on_filled(self, slot, fs) -> None: try: if slot.status is SlotState.PENDING_BUY: self._svc.transit_slot( slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="buy_filled", actor="execution_loop", avg_price=fs.avg_price, qty=float(fs.filled_qty) or slot.qty, ) else: # PENDING_SELL self._svc.transit_slot( slot.slot_id, SlotState.PENDING_SELL, SlotState.CLOSED, reason="sell_filled", actor="execution_loop", ) except SlotStateMismatchError: return # slot moved concurrently; next tick reconciles self._repo.clear_dispatch(slot.slot_id) # done with this order def _rollback(self, slot, *, reason: str) -> None: try: if slot.status is SlotState.PENDING_BUY: self._svc.transit_slot( slot.slot_id, SlotState.PENDING_BUY, SlotState.IDLE, reason=reason, actor="execution_loop", ) # idle reset clears broker_order_id in the domain else: # PENDING_SELL rejected -> back to holding, clear order for re-dispatch self._svc.transit_slot( slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING, reason=reason, actor="execution_loop", ) self._repo.clear_dispatch(slot.slot_id) except SlotStateMismatchError: return