trex-core/trex_core/application/execution.py

"""Order execution service (REQ-013, REQ-014, REQ-018, dcy-003, dcy-004).

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).
            dcy-020/REQ-062 (BAT-030): a SELL is dispatched with the broker
            balance quantity (hldg_qty) as the single source of truth — never
            the slot's mirror qty. Broker flat (qty 0) -> skip + WARN
            (divergence surfaces via reconcile, REQ-062 A).
  poll:     observe fills on dispatched pending slots and transition the slot
            (pending_buy->holding / pending_sell->inactive), or roll back on reject.
            On a successful fill, enqueue a LedgerWorkItem (slot transition is
            kept ON the critical path; the ledger write is eventually-consistent
            via a separate writer — REQ-019).
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. The ledger outbox is the
single producer for ledger work items (NFR-3, single record subject).
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable, Optional

from trex_core.application.broker_port import BrokerPort, OrderRequest
from trex_core.application.ledger_outbox import LedgerOutbox
from trex_core.application.ledger_port import LedgerTrade
from trex_core.domain.slot import SlotService, SlotState, SlotStateMismatchError

_log = logging.getLogger(__name__)

# REQ-021/023: event emission hook. Runtime wires a callable that publishes
# 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

_MAX_DISPATCH_PER_TICK = 64


@dataclass
class ExecutionResult:
    dispatched: int = 0
    filled: int = 0
    rolled_back: int = 0
    ledger_enqueued: int = 0     # REQ-019: ledger work items appended this tick
    ledger_skipped: int = 0      # REQ-018 AC-4: fills without portfolio_id
    stale_pending: int = 0       # dcy-013 / REQ-014 FR-3: pending orders over threshold
    sell_skipped_no_balance: int = 0  # dcy-020 / REQ-062: broker flat -> sell skipped


class ExecutionService:
    def __init__(
        self,
        slot_service: SlotService,
        repo,
        broker: BrokerPort,
        ledger_outbox: Optional[LedgerOutbox] = None,
        event_emitter: Optional[SlotEventEmitter] = None,
        stale_pending_threshold_sec: Optional[int] = None,
        instance_key: str = "default",
    ) -> None:
        self._svc = slot_service
        self._repo = repo
        self._broker = broker
        self._ledger_outbox = ledger_outbox
        self._event_emitter = event_emitter
        # dcy-020/REQ-056 (BAT-030): dispatch/poll are instance-scoped so two
        # daemons sharing a DB never touch each other's orders.
        self._instance_key = instance_key
        # dcy-013 / REQ-014 (issue #9): Alert-only pending-order stale watchdog.
        # <=0 or None disables it. When set, a dispatched-pending slot whose
        # last update is older than the threshold is surfaced (WARN + counter)
        # without changing its state — the backstop for a cancel-detection miss.
        self._stale_threshold_sec = (
            stale_pending_threshold_sec
            if stale_pending_threshold_sec and stale_pending_threshold_sec > 0
            else None
        )

    # ------------------------------------------------------------
    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:
        sell_balance: Optional[dict[str, int]] = None  # lazy, once per tick
        for slot in self._repo.list_undispatched_pending(
            self._instance_key, _MAX_DISPATCH_PER_TICK
        ):
            side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
            if side == "sell":
                # dcy-020 / REQ-062 (BAT-030, D3): the sell quantity is the
                # BROKER balance (hldg_qty) — the slot's mirror qty is never
                # trusted. Broker flat -> skip dispatch + WARN; the divergence
                # surfaces via reconcile_core_holdings (REQ-062 A). No mirror
                # fallback by design.
                if sell_balance is None:
                    sell_balance = self._balance_map()
                bal = sell_balance.get(slot.symbol or "", 0)
                # dcy-021 / REQ-062 (D4): full sell (slot.qty is None) => broker
                # balance; partial sell => min(order qty, broker balance). Broker
                # flat -> skip + WARN (divergence surfaces via reconcile).
                qty = bal if slot.qty is None else min(int(slot.qty), bal)
                if qty <= 0:
                    r.sell_skipped_no_balance += 1
                    _log.warning(
                        "sell dispatch skipped: slot=%s symbol=%s broker balance is 0 "
                        "(divergence — no order submitted; operator/reconcile attention)",
                        slot.slot_id, slot.symbol,
                    )
                    continue
            else:
                qty = int(slot.qty or 0)
            # dcy-012 / REQ-013: side-agnostic — use the slot's pending-order
            # limit_price only. A sell never reuses a buy limit (issue #10):
            # limit_price is cleared on entering HOLDING, so a market sell
            # carries None -> order_type "01".
            limit = int(slot.limit_price) if slot.limit_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:
        now = datetime.now(tz=timezone.utc)
        for slot in self._repo.list_dispatched_pending(self._instance_key):
            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, side, r)
                r.filled += 1
            elif fs.is_partial_cancel:
                # dcy-013 / REQ-013 FR-2: partial fill then external cancel.
                self._on_partial_cancel(slot, fs, side, r)
                r.filled += 1
            elif fs.is_terminal_reject:
                # rejected or full external cancel (dcy-013 FR-1) -> rollback.
                self._rollback(slot, reason=f"order_{fs.state}")
                r.rolled_back += 1
            else:
                # accepted / partially_filled (not cancelled) -> leave for next
                # tick, but surface if it has been stuck too long (FR-3).
                self._check_stale_pending(slot, now, r)

    # ------------------------------------------------------------
    def _balance_map(self) -> dict:
        """Current broker balance as {symbol: hldg_qty} (REQ-062 ground truth)."""
        return {b.symbol: int(b.hldg_qty) for b in self._broker.get_balance()}

    def _on_filled(self, slot, fs, side: str, r: ExecutionResult) -> None:
        broker_order_id = slot.broker_order_id  # captured BEFORE clear_dispatch
        try:
            if slot.status is SlotState.PENDING_BUY:
                # dcy-021/REQ-066: held qty = broker balance (authority, includes
                # the just-filled increase); avg_price = weighted average. For an
                # initial buy slot.avg_price is None (=0) so new_avg = fill price.
                total = self._balance_map().get(slot.symbol or "", 0)
                filled = float(fs.filled_qty)
                new_qty = float(total) if total > 0 else filled
                old_qty = max(0.0, new_qty - filled)
                old_avg = float(slot.avg_price) if slot.avg_price else 0.0
                fill_price = float(fs.avg_price) if fs.avg_price is not None else old_avg
                new_avg = (
                    (old_qty * old_avg + filled * fill_price) / new_qty
                    if new_qty > 0 else fill_price
                )
                updated = self._svc.transit_slot(
                    slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
                    reason="buy_filled", actor="execution_loop",
                    avg_price=new_avg, qty=new_qty,
                )
                event_type = "slot_filled_buy"
            else:  # PENDING_SELL — dcy-021/REQ-066: remainder>0 stays holding,
                #     full exit -> inactive. Held qty = broker balance.
                total = self._balance_map().get(slot.symbol or "", 0)
                if total > 0:
                    updated = self._svc.transit_slot(
                        slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING,
                        reason="sell_filled_partial", actor="execution_loop",
                        qty=float(total),  # avg_price (cost basis) preserved
                    )
                else:
                    updated = self._svc.transit_slot(
                        slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
                        reason="sell_filled", actor="execution_loop",
                    )
                event_type = "slot_filled_sell"
        except SlotStateMismatchError:
            return  # slot moved concurrently; next tick reconciles
        # REQ-018: enqueue ledger work item AFTER the slot transition is durable.
        # Slot transition is the critical path; ledger write is async (REQ-019).
        self._record_ledger(slot, fs, side, broker_order_id, r)
        # 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)
        self._repo.clear_dispatch(slot.slot_id)  # done with this order

    def _on_partial_cancel(self, slot, fs, side: str, r: ExecutionResult) -> None:
        """Partial fill then external cancel (dcy-013 / REQ-013 FR-2, OQ-1 A).

        Record the filled portion to the ledger ONCE, decrement the held qty,
        and return the slot to holding with the remainder — so the remaining
        position can be re-sold and no over-sell / double-record occurs (NFR-3).
        avg_price (cost basis) is preserved on a sell; a buy keeps only the
        portion that actually filled.
        """
        broker_order_id = slot.broker_order_id  # captured BEFORE clear_dispatch
        filled = float(fs.filled_qty)
        try:
            if slot.status is SlotState.PENDING_BUY:
                # dcy-021: held qty = broker balance; avg_price weighted average
                # (consistent with _on_filled). Only the filled portion recorded.
                total = self._balance_map().get(slot.symbol or "", 0)
                new_qty = float(total) if total > 0 else filled
                old_qty = max(0.0, new_qty - filled)
                old_avg = float(slot.avg_price) if slot.avg_price else 0.0
                fill_price = float(fs.avg_price) if fs.avg_price is not None else old_avg
                new_avg = (
                    (old_qty * old_avg + filled * fill_price) / new_qty
                    if new_qty > 0 else fill_price
                )
                updated = self._svc.transit_slot(
                    slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
                    reason="buy_partial_cancel", actor="execution_loop",
                    avg_price=new_avg, qty=new_qty,
                )
                event_type = "slot_filled_buy"
            else:  # PENDING_SELL: remainder = broker balance, cost basis unchanged.
                total = self._balance_map().get(slot.symbol or "", 0)
                if total > 0:
                    updated = self._svc.transit_slot(
                        slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING,
                        reason="sell_partial_cancel", actor="execution_loop",
                        qty=float(total),
                    )
                else:
                    updated = self._svc.transit_slot(
                        slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
                        reason="sell_partial_cancel", actor="execution_loop",
                    )
                event_type = "slot_filled_sell"
        except SlotStateMismatchError:
            return  # slot moved concurrently; next tick reconciles
        self._record_ledger(slot, fs, side, broker_order_id, r)
        self._emit_event(event_type, updated)
        self._repo.clear_dispatch(slot.slot_id)

    def _check_stale_pending(self, slot, now: datetime, r: ExecutionResult) -> None:
        """Alert-only: surface a dispatched-pending slot stuck past the threshold.

        dcy-013 / REQ-014 FR-3 — state is NOT changed (auto-remediation is a
        follow-up). This is the backstop for a cancel-detection miss (R-1).
        """
        if self._stale_threshold_sec is None or slot.updated_at is None:
            return
        age = (now - slot.updated_at).total_seconds()
        if age >= self._stale_threshold_sec:
            r.stale_pending += 1
            _log.warning(
                "pending order stale: slot=%s status=%s broker_order_id=%s "
                "age_sec=%.0f threshold_sec=%d (no state change — ops attention)",
                slot.slot_id, slot.status.value, slot.broker_order_id,
                age, self._stale_threshold_sec,
            )

    def _emit_event(self, event_type: str, slot) -> None:
        if self._event_emitter is None:
            return
        try:
            self._event_emitter(event_type, slot)
        except Exception:  # noqa: BLE001 - event publish failure is non-critical
            return

    def _record_ledger(self, slot, fs, side: str, broker_order_id: str, r: ExecutionResult) -> None:
        """Append a LedgerWorkItem for the just-filled trade (REQ-018 + REQ-019)."""
        if self._ledger_outbox is None:
            return
        if not slot.portfolio_id:
            # dcy-011 / REQ-018 (issue #11): ledger skip must NOT be silent.
            # Slot transition still proceeded; surface for ops (FR-4).
            r.ledger_skipped += 1
            _log.warning(
                "ledger skip: slot=%s symbol=%s side=%s reason=missing_portfolio_id "
                "(trade NOT recorded to portfolio-hub)",
                slot.slot_id, slot.symbol, side,
            )
            return
        if fs.execution_id is None:
            # Broker contract regression — slot already transitioned; surface for ops.
            r.ledger_skipped += 1
            _log.warning(
                "ledger skip: slot=%s portfolio_id=%s side=%s reason=missing_execution_id "
                "(trade NOT recorded to portfolio-hub)",
                slot.slot_id, slot.portfolio_id, side,
            )
            return
        price = float(fs.avg_price) if fs.avg_price is not None else 0.0
        trade = LedgerTrade(
            asset_id=str(slot.portfolio_id),
            order_id=str(broker_order_id),
            execution_id=str(fs.execution_id),
            side=side,
            qty=float(fs.filled_qty),
            price=price,
            fee=float(fs.fee),
            tax=float(fs.tax),
            executed_at=datetime.now(tz=timezone.utc),
        )
        self._ledger_outbox.enqueue(trade)
        r.ledger_enqueued += 1

    def _rollback(self, slot, *, reason: str) -> None:
        total = self._balance_map().get(slot.symbol or "", 0)
        try:
            if slot.status is SlotState.PENDING_BUY:
                if total > 0:
                    # dcy-021: increase rejected — the prior holding is intact
                    # (broker still holds), so return to holding, not inactive.
                    self._svc.transit_slot(
                        slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
                        reason=reason, actor="execution_loop", qty=float(total),
                    )
                    self._repo.clear_dispatch(slot.slot_id)
                else:  # initial buy rejected -> inactive (clears broker_order_id)
                    self._svc.transit_slot(
                        slot.slot_id, SlotState.PENDING_BUY, SlotState.INACTIVE,
                        reason=reason, actor="execution_loop",
                    )
            else:  # PENDING_SELL rejected -> back to holding (broker authoritative qty)
                self._svc.transit_slot(
                    slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING,
                    reason=reason, actor="execution_loop",
                    qty=float(total) if total > 0 else None,
                )
                self._repo.clear_dispatch(slot.slot_id)
        except SlotStateMismatchError:
            return