"""Slot state machine: 6 states + 8 allowed transitions (REQ-001, REQ-002). The transitions match dcy-001 FR-2 and BAT-001 design. """ from __future__ import annotations from enum import Enum # Shared PostgreSQL advisory lock id for Buy/Sell decision threads and # operator CLI manual transitions. Single constant so all callers serialize # through the same lock (REQ-002 NFR-1, dcy-001 R-1). SLOT_ADVISORY_LOCK_ID: int = 0x5107_4D31 # arbitrary mnemonic constant class SlotState(str, Enum): """6-state slot lifecycle (dcy-001 5.3.5.1).""" IDLE = "idle" PENDING_BUY = "pending_buy" HOLDING = "holding" PENDING_SELL = "pending_sell" CLOSED = "closed" ERROR = "error" # 8 allowed transitions per dcy-001 FR-2. Any pair not in this set must be # rejected by SlotService. # Note: `* -> error` is represented as transitions from every non-error state. _BASE_TRANSITIONS: frozenset[tuple[SlotState, SlotState]] = frozenset( { (SlotState.IDLE, SlotState.PENDING_BUY), # 1. buy decision (SlotState.PENDING_BUY, SlotState.HOLDING), # 2. buy filled (SlotState.PENDING_BUY, SlotState.IDLE), # 3. buy rejected/rollback (SlotState.HOLDING, SlotState.PENDING_SELL), # 4. sell decision (SlotState.PENDING_SELL, SlotState.CLOSED), # 5. sell filled (SlotState.PENDING_SELL, SlotState.HOLDING), # 6. sell rejected/rollback (SlotState.CLOSED, SlotState.IDLE), # 7. recovery (with daily_symbol_entries +1) } ) # 8th transition family: any non-error state may move to error (operator manual # resolve uses `error -> idle | holding` which is also covered explicitly). _TO_ERROR_TRANSITIONS: frozenset[tuple[SlotState, SlotState]] = frozenset( (s, SlotState.ERROR) for s in SlotState if s is not SlotState.ERROR ) # Operator manual resolve transitions out of error (REQ-004). _FROM_ERROR_TRANSITIONS: frozenset[tuple[SlotState, SlotState]] = frozenset( { (SlotState.ERROR, SlotState.IDLE), (SlotState.ERROR, SlotState.HOLDING), } ) ALLOWED_TRANSITIONS: frozenset[tuple[SlotState, SlotState]] = ( _BASE_TRANSITIONS | _TO_ERROR_TRANSITIONS | _FROM_ERROR_TRANSITIONS ) def is_allowed_transition(from_state: SlotState, to_state: SlotState) -> bool: """Return True iff (from_state, to_state) is one of the 8 transition families.""" return (from_state, to_state) in ALLOWED_TRANSITIONS