sentiment-engine/prod/exec_unified/drive_loop.py

"""The drive loop — PINK's beating heart, ported (spec §7).

This is a faithful transcription of ``pink_direct.py``'s working-order driver
(``_exec_after_submit`` L975, ``_handle_expired_working`` L1089, ``_exec_safe_to_requote``
L1049) against INJECTED seams (C1: clock, venue+kernel I/O behind one ``ExecPort``) so it
carries no ambient state and could run venue-side someday. **Every branch is a headstone —
transcribed warts-and-all, not tidied** (operator ruling: "we paid dearly for pink").

What this loop is NOT (contract §2 / audit): it never sees capital, leverage, posture, or
picks an asset — those are the kernel/brain SEAM, reached only through ``ExecPort``. Where a
PINK guard lived in that SEAM, its lesson is preserved here as a referenced comment (the
zero-silent-suppression rule), not silently dropped.

Ported: table rows #1-8,11-13,15-16 of PINK_DRIVE_LOOP_PORT_INVENTORY.md.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
from typing import Protocol

from . import _constants as K
from .contract import Side
from .router import ExecutionMethod, RoutingDecision
from .working import Action, WorkingOrder, WorkingRegistry

LOGGER = logging.getLogger(__name__)

# Kernel FSM stages that mean "a position exists" (incl. partials + exit-in-flight).
# pink_direct.py:936 `_SLOT_OPENISH`. Strings because they come from the kernel FSM seam.
OPENISH = frozenset({
    "PARTIAL_FILL", "POSITION_OPENED", "POSITION_OPEN", "EXIT_REQUESTED", "EXIT_SENT",
    "EXIT_ACKED", "EXIT_WORKING", "POSITION_PARTIALLY_CLOSED",
})
# Stages that mean the position is gone. pink_direct.py:1106.
CLOSED_STAGES = frozenset({"POSITION_CLOSED", "CLOSED", "TRADE_TERMINAL_WRITTEN", "IDLE"})
# Stages that mean the venue rejected our order (post-only would cross, etc.).
# pink_direct.py:1000. A rejected maker quote still registers → resolves via the TTL path.
REJECTED_STAGES = frozenset({"ORDER_REJECTED", "EXIT_REJECTED"})


@dataclass(frozen=True)
class SlotView:
    """Kernel-truth snapshot of the single slot (pink_direct.py:940 `_exec_slot_view`).
    ``size`` is Decimal — audit H1: PINK used float, the port does not."""

    trade_id: str
    stage: str
    size: Decimal


@dataclass(frozen=True)
class ResubmitPlan:
    """A rebuilt order the loop hands back to the venue (retry / market / exit-escalation).
    Venue-agnostic: the dialect turns method+limit_price into a venue payload."""

    request_id: str            # this attempt's id — unique per attempt (audit H4)
    base_request_id: str
    asset: str
    side: Side
    action: Action
    method: ExecutionMethod
    limit_price: Decimal       # ignored when method is TAKER
    attempt: int
    reduce_only: bool
    size: Decimal = Decimal("0")   # base qty for the venue (threaded from the request)


# The plan the venue receives is the same shape whether it is an initial submit (executor,
# attempt 0) or a rebuilt retry/escalation (drive loop). "Resubmit" is the origin name;
# OrderPlan is the role name the executor uses.
OrderPlan = ResubmitPlan


class ExecPort(Protocol):
    """The single seam to kernel + venue. Fakeable; the loop touches the world ONLY here."""

    def clock(self) -> float: ...
    def slot_view(self) -> SlotView: ...
    def last_own_fill_at(self) -> float: ...
    async def pump(self) -> None: ...                       # drain venue events → kernel
    async def cancel(self, wo: WorkingOrder) -> None: ...    # idempotent on the venue
    async def open_positions(self) -> list[dict]: ...        # venue-truth flat probe
    async def submit(self, plan: ResubmitPlan) -> None: ...  # resubmit rebuilt order


class MissAction(Enum):
    RETRY = "retry"      # re-quote maker (chase budget remains)
    MARKET = "market"    # cross now (cross_on_expiry urgencies)
    ABANDON = "abandon"  # ACQUIRE: a missed entry is free — never chase, never cross (§4-1)


class Resolved(Enum):
    ALREADY_RESOLVED = "already_resolved"      # fill/cancel raced the sweep
    FILLED_AFTER_TTL = "filled_after_ttl"      # fill surfaced during cancel round-trip
    ESCALATED_MARKET = "escalated_market"      # EXIT never strands / entry cross
    RETRIED = "retried"
    ABANDONED = "abandoned"                     # ACQUIRE miss
    SKIPPED_SLOT_BUSY = "skipped_slot_busy"    # raced remainder fill — never double-enter
    REQUOTE_BLOCKED = "requote_blocked"        # venue not provably flat — fail safe


class DriveLoop:
    """Orchestrates working-order resolution. One instance per exec instance."""

    def __init__(self, port: ExecPort, registry: WorkingRegistry,
                 logger: logging.Logger = LOGGER) -> None:
        self.port = port
        self.registry = registry
        self.log = logger

    # ── classification ───────────────────────────────────────────────────────
    def _is_resolved(self, wo: WorkingOrder, slot: SlotView) -> bool:
        """Entry filled or exit done, per kernel truth (pink _entry_filled/_exit_done, L1099).

        ENTER: our entry filled — the slot carries OUR clientOrderId (the kernel tags the new
        position with it; clientOrderId echo is the best-practice fill key, audit H5) and shows
        size + an open stage.

        EXIT: the position is gone — size drained or a closed stage. **Deliberately SIZE-based,
        not trade_id-based**: pink_direct.py:1105 could compare `slot_tid != wo.trade_id` only
        because it REUSED the position's trade_id for the exit intent. This layer is agnostic
        (contract §2) — the caller mints a fresh request_id for the exit and never hands us the
        position's id — so an exit is "done" iff the position closed, which is the size signal.
        """
        if wo.action is Action.ENTER:
            return slot.trade_id == wo.request_id and slot.size > 0 and slot.stage in OPENISH
        return slot.size <= 0 or slot.stage in CLOSED_STAGES

    def after_submit(self, wo: WorkingOrder, *, rejected: bool) -> str:
        """Classify a maker submit: filled-now / working / rejected. pink L975.

        A rejected post-only quote registers with an already-expired deadline so the TTL
        sweep resolves it through the ONE shared miss/escalation path — do not build a
        second path (pink L1010-1011)."""
        slot = self.port.slot_view()
        if self._is_resolved(wo, slot):
            return "immediate_fill"          # filled on submit — do not register
        self.registry.register(wo)
        if rejected:
            self.registry.expire_now(wo.request_id)
            return "working_reject"
        return "working"

    # ── the TTL sweep ────────────────────────────────────────────────────────
    async def resolve_expired(self) -> list[Resolved]:
        """One sweep over expired quotes. Caller drives cadence (injected — no literal 1 s;
        pink L1070 hardcoded 1.0, amended per spec NFR). An expiry-handler exception drops
        THAT order to a logged error and continues — never let one wedge the sweep."""
        out: list[Resolved] = []
        for wo in self.registry.expired():
            try:
                out.append(await self.handle_expired(wo))
            except Exception as exc:  # BI-3: log, never bare-swallow (pink's silent-demon)
                self.log.error("drive_loop: expiry handler failed for %s: %r",
                               wo.request_id, exc, exc_info=True)
        return out

    async def handle_expired(self, wo: WorkingOrder) -> Resolved:
        """The heart — pink_direct.py:1089, 10-step sequence, order preserved exactly."""
        # 1. already resolved (a fill/cancel notification raced this sweep)
        if self.registry.working(wo.request_id) is None:
            return Resolved.ALREADY_RESOLVED
        # 2. drain late venue events FIRST — the quote may already be filled
        await self.port.pump()
        if self.registry.working(wo.request_id) is None:
            return Resolved.ALREADY_RESOLVED
        # 3. cancel the quote (idempotent; CANCEL_REJECT on a filled order is harmless;
        #    on a partial entry this cancels the remainder). BI-3: log failures, don't swallow.
        try:
            await self.port.cancel(wo)
        except Exception as exc:
            self.log.warning("drive_loop: ttl-cancel %s failed: %r", wo.request_id, exc)
        # 4. pump again — the cancel round-trip may have surfaced a fill
        await self.port.pump()
        # 5. re-classify AFTER the cancel (fill may have raced the cancel — the core race)
        slot = self.port.slot_view()
        if self._is_resolved(wo, slot):
            self.registry.note_fill(wo.request_id)
            return Resolved.FILLED_AFTER_TTL
        self.registry.note_cancel(wo.request_id)

        # 6. EXIT never strands a position → escalate to MARKET, same base id (pink L1147)
        if wo.action is Action.EXIT:
            await self.port.submit(self._resubmit(wo, ExecutionMethod.TAKER, reduce_only=True))
            await self.port.pump()
            self.log.warning("drive_loop: exit TTL → MARKET fallback %s", wo.request_id)
            return Resolved.ESCALATED_MARKET

        # 7. ENTER miss policy: retry(bounded) | market | abandon
        miss = self._entry_miss_action(wo)
        # 8. slot-busy guard: never double-enter on a raced remainder fill (pink L1173)
        slot = self.port.slot_view()
        if slot.size > 0 or slot.stage in OPENISH:
            self.log.warning("drive_loop: entry miss %s: slot busy (%s) — skip",
                             wo.request_id, slot.stage)
            return Resolved.SKIPPED_SLOT_BUSY
        if miss is MissAction.ABANDON:
            # ACQUIRE: a missed entry is free, a chased/crossed entry is not (§4-1). Abandon.
            self.log.info("drive_loop: entry miss %s → abandon", wo.request_id)
            return Resolved.ABANDONED
        # 9. venue-truth requote gate: re-quote ONLY when provably flat. Ambiguity → skip;
        #    "a skipped entry is always safe, a doubled position is not" (pink L1180-1190).
        if not await self.safe_to_requote():
            self.log.warning("drive_loop: entry miss %s: venue not provably flat — skip",
                             wo.request_id)
            return Resolved.REQUOTE_BLOCKED
        # 10. resubmit retry(maker) | market(taker)
        method = ExecutionMethod.MAKER if miss is MissAction.RETRY else ExecutionMethod.TAKER
        await self.port.submit(self._resubmit(wo, method, reduce_only=False))
        await self.port.pump()
        return Resolved.RETRIED if miss is MissAction.RETRY else Resolved.ESCALATED_MARKET

    # ── gates & helpers ──────────────────────────────────────────────────────
    async def safe_to_requote(self) -> bool:
        """True only when the venue is PROVABLY flat. Fails SAFE on: recent own fill
        (REST reconcile lags WS by seconds), any live position, or any probe error
        (pink_direct.py:1049)."""
        if self.port.clock() - (self.port.last_own_fill_at() or 0.0) < K.REQUOTE_HOT_WINDOW_S:
            return False
        try:
            rows = await self.port.open_positions()
        except Exception as exc:
            self.log.warning("drive_loop: requote probe failed (%r) — fail safe", exc)
            return False
        for row in rows or []:
            qty = abs(_dec(row.get("positionAmt") or row.get("positionQty")
                           or row.get("qty") or 0))
            if qty > Decimal("1e-9"):
                return False
        return True

    def _entry_miss_action(self, wo: WorkingOrder) -> MissAction:
        """Chase budget from the routing decision (stored on the working order at register).
        ACQUIRE: max_reprices=0, cross_on_expiry=False → ABANDON. pink miss policy L1166."""
        max_reprices = int(wo.meta.get("max_reprices", 0))
        cross_on_expiry = bool(wo.meta.get("cross_on_expiry", False))
        if wo.attempt < max_reprices:
            return MissAction.RETRY
        if cross_on_expiry:
            return MissAction.MARKET
        return MissAction.ABANDON

    def _resubmit(self, wo: WorkingOrder, method: ExecutionMethod, *,
                  reduce_only: bool) -> ResubmitPlan:
        """Rebuild the order for a retry/market/escalation with a FRESH per-attempt id
        (audit H4 / FIX OrigClOrdID: retries must not reuse a clientOrderId)."""
        nxt = wo.attempt + 1
        base = wo.base_request_id
        return ResubmitPlan(
            request_id=f"{base}-{nxt}", base_request_id=base, asset=wo.asset,
            side=wo.side, action=wo.action, method=method,
            limit_price=wo.limit_price, attempt=nxt, reduce_only=reduce_only,
            size=wo.size,
        )


def build_working_order(*, request_id: str, base_request_id: str, asset: str, side: Side,
                        action: Action, limit_price: Decimal, decision: RoutingDecision,
                        clock: float, ttl_s: float, size: Decimal = Decimal("0"),
                        attempt: int = 0) -> WorkingOrder:
    """Construct a WorkingOrder, stamping the policy bits the drive loop needs at expiry
    (max_reprices / cross_on_expiry) so the loop never imports the brain's decision path."""
    return WorkingOrder(
        request_id=request_id, base_request_id=base_request_id, asset=asset, side=side,
        action=action, limit_price=limit_price, deadline=clock + ttl_s, created=clock,
        size=size, attempt=attempt,
        meta={"max_reprices": decision.max_reprices,
              "cross_on_expiry": decision.cross_on_expiry},
    )


def _dec(v: object) -> Decimal:
    try:
        return Decimal(str(v))
    except Exception:
        return Decimal(0)