exec(unified): drive loop §7 — PINK _handle_expired_working ported + audit build items
drive_loop.py transcribes pink_direct.py:1089 (the 10-step expiry sequence) warts-and-all against injected ports (ExecPort seam: clock+kernel+venue), so no ambient state (C1). Carries the scar tissue: pump-before-cancel, re-classify-after-cancel (fill races cancel), EXIT never strands→MARKET, slot-busy double-entry guard, fail-safe venue-truth requote gate. SEAM learnings preserved as referenced comments (zero-silent-suppression rule). Decimal sizes (H1). - working.py: WorkingRegistry + WorkingOrder, injected clock, rejected→expire_now (one shared path). contract.py: client_order_id_core(attempt) — unique per attempt (audit H4/FIX). - _constants.py: REQUOTE_HOT_WINDOW_S=5.0 (prov: 2026-06-10 double-entry). - inventory §6: audit build items BI-1..5 folded in. - test_drive_loop.py: 15 scar-tissue tests. Mutation-verified RED under exit-no-escalate (kills 3) and no-hot-window (kills double-entry guard). Full exec_unified suite: 77 green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Codex
257
prod/exec_unified/drive_loop.py
Normal file
257
prod/exec_unified/drive_loop.py
Normal file
@@ -0,0 +1,257 @@
|
||||
"""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"})
|
||||
|
||||
|
||||
@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
|
||||
|
||||
|
||||
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)."""
|
||||
if wo.action is Action.ENTER:
|
||||
return slot.trade_id == wo.request_id and slot.size > 0 and slot.stage in OPENISH
|
||||
return (slot.trade_id != wo.request_id or 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,
|
||||
)
|
||||
|
||||
|
||||
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, 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,
|
||||
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)
|
||||
Reference in New Issue
Block a user