"""Slot domain service (REQ-002; dcy-020 asset-centric). - transit_slot(slot_id, from_state, to_state, reason, actor, **fields) - sell(slot_id, actor, reason, limit_price=None) - update_thresholds(slot_id, ...) - compute_slot_capacity(max_active_slots=None) All state-mutating APIs run inside repository.transactional_advisory_lock() to serialize Buy/Sell decision threads + operator CLI (REQ-002 AC-2). dcy-020 (BAT-030): the pool-model surface (daily preallocation, symbol entry counters) was removed at the cutover; slots are permanent asset slots. """ from __future__ import annotations from dataclasses import replace from datetime import datetime, timezone from decimal import Decimal from typing import Any, Optional from uuid import UUID from trex_core.domain.slot.errors import ( SlotNotFoundError, SlotStateMismatchError, SlotTransitionError, ) from trex_core.domain.slot.models import Slot, SlotCapacity, SlotTransitionRecord from trex_core.domain.slot.repository import SlotRepository from trex_core.domain.slot.state import ( SlotState, is_allowed_transition, slot_advisory_lock_id, ) # Sentinel distinguishing "field omitted (keep current)" from "field set to # None (clear threshold)" in update_thresholds (REQ-040 AC-4 vs AC-5). _UNSET: Any = object() class SlotService: """Slot lifecycle service. Race-free transitions via advisory lock + transaction.""" def __init__( self, repository: SlotRepository, instance_key: str = "default", *, strategy_state_store: Any = None, ) -> None: self._repo = repository # dcy-020/REQ-057: advisory lock is scoped per instance so a shared DB # does not serialize transitions across instances (NFR-1). self._instance_key = instance_key self._lock_id = slot_advisory_lock_id(instance_key) # dcy-028/REQ-081 (C-3): when injected, all strategy_state for a slot is # deleted on close (→INACTIVE). None keeps legacy behaviour (no store). self._strategy_state_store = strategy_state_store @property def instance_key(self) -> str: return self._instance_key # ------------------------------------------------------------ # transit_slot # ------------------------------------------------------------ def transit_slot( self, slot_id: UUID, from_state: SlotState, to_state: SlotState, reason: str, actor: str, *, symbol: Optional[str] = None, qty: Optional[float] = None, avg_price: Optional[float] = None, limit_price: Optional[float] = None, order_id: Optional[str] = None, strategy_id: Optional[str] = None, broker_order_id: Optional[str] = None, portfolio_id: Optional[str] = None, sell_strategy_id: Optional[str] = None, forced_close_at: Optional[datetime] = None, stop_loss_price: Optional[Decimal] = None, take_profit_price: Optional[Decimal] = None, # dcy-022 / REQ-068: execution policy (set on PENDING entry, cleared on exit). timeout_sec: Optional[int] = None, slippage_pct: Optional[float] = None, venue: Optional[str] = None, extra: Optional[dict[str, Any]] = None, ) -> Slot: """Transition slot from one state to another atomically. Within a single advisory-locked transaction: 1. fetch slot 2. verify current state == from_state (race detection) 3. verify (from_state, to_state) is in the allowed set 4. update slot row (fields per to_state semantics) 5. append transition_log row External I/O (broker, kis-data-hub, portfolio-hub) MUST NOT be performed inside this function — caller does I/O outside the lock (REQ-002 AC-5). """ if not is_allowed_transition(from_state, to_state): raise SlotTransitionError( f"transition not allowed: {from_state.value} -> {to_state.value}" ) with self._repo.transactional_advisory_lock(self._lock_id): slot = self._repo.fetch_slot(slot_id) if slot is None: raise SlotNotFoundError(f"slot not found: {slot_id}") # Race detection — current state must match caller's from_state. if slot.status is not from_state: raise SlotStateMismatchError( f"slot {slot_id} current state {slot.status.value} " f"!= requested from_state {from_state.value}" ) updated = self._apply_transition_fields( slot, to_state, symbol=symbol, qty=qty, avg_price=avg_price, limit_price=limit_price, order_id=order_id, strategy_id=strategy_id, broker_order_id=broker_order_id, portfolio_id=portfolio_id, sell_strategy_id=sell_strategy_id, forced_close_at=forced_close_at, stop_loss_price=stop_loss_price, take_profit_price=take_profit_price, timeout_sec=timeout_sec, slippage_pct=slippage_pct, venue=venue, ) self._repo.update_slot(updated) self._repo.append_transition_log( SlotTransitionRecord( slot_id=slot_id, from_state=from_state, to_state=to_state, ts=datetime.now(tz=timezone.utc), reason=reason, actor=actor, extra=extra or {}, ) ) # dcy-028/REQ-081 (C-3): close deletes the slot's strategy_state so a # reused slot starts empty. Covers every close path (strategy sell, # safety-sell, forced-close, operator) since all funnel to →INACTIVE. if to_state is SlotState.INACTIVE and self._strategy_state_store is not None: self._strategy_state_store.delete_all(slot_id) return updated @staticmethod def _apply_transition_fields( slot: Slot, to_state: SlotState, *, symbol: Optional[str], qty: Optional[float], avg_price: Optional[float], limit_price: Optional[float], order_id: Optional[str], strategy_id: Optional[str], broker_order_id: Optional[str] = None, portfolio_id: Optional[str] = None, sell_strategy_id: Optional[str] = None, forced_close_at: Optional[datetime] = None, stop_loss_price: Optional[Decimal] = None, take_profit_price: Optional[Decimal] = None, timeout_sec: Optional[int] = None, slippage_pct: Optional[float] = None, venue: Optional[str] = None, ) -> Slot: """Compute the post-transition Slot row. dcy-020/REQ-057 (BAT-030): entering INACTIVE deactivates the permanent asset slot — all position/occupancy fields are cleared while the asset identity (instance_key, portfolio_id, symbol, retired_at) is preserved. Covers buy rollback (pending_buy→inactive), sell fill (pending_sell→inactive) and operator error resolve (error→inactive). dcy-012 / REQ-002: limit_price is set only while a slot is in a pending state (the live order's limit); entering any non-pending state clears it so a buy limit never leaks into the next sell order (issue #10). """ if to_state is SlotState.INACTIVE: return replace( slot, status=SlotState.INACTIVE, qty=None, avg_price=None, limit_price=None, order_id=None, strategy_id=None, broker_order_id=None, sell_strategy_id=None, forced_close_at=None, stop_loss_price=None, take_profit_price=None, timeout_sec=None, slippage_pct=None, venue="KRX", updated_at=datetime.now(tz=timezone.utc), ) # limit_price lifecycle (dcy-012 FR-1): set only on entering a pending # state (the order's limit); cleared on entering HOLDING/ERROR. # execution policy (dcy-022/REQ-068): same lifecycle as limit_price. if to_state in {SlotState.PENDING_BUY, SlotState.PENDING_SELL}: new_limit_price = limit_price if limit_price is not None else slot.limit_price new_timeout_sec = timeout_sec if timeout_sec is not None else slot.timeout_sec new_slippage_pct = slippage_pct if slippage_pct is not None else slot.slippage_pct new_venue = venue if venue is not None else slot.venue else: # HOLDING / ERROR — pending order policy no longer applies. new_limit_price = None new_timeout_sec = None new_slippage_pct = None # A completed order's venue must not leak into a later autonomous # sell. The next order starts session-aware unless its intent # explicitly chooses KRX/NXT/SOR. new_venue = "AUTO" if to_state is SlotState.HOLDING else slot.venue # occupancy_seq (dcy-028/REQ-081): bump when a slot begins a new # occupancy (inactive→pending_buy). Stable across the rest of the # holding episode; the INACTIVE branch above preserves it (monotonic). if slot.status is SlotState.INACTIVE and to_state is SlotState.PENDING_BUY: new_occupancy_seq = slot.occupancy_seq + 1 else: new_occupancy_seq = slot.occupancy_seq # qty semantics during a pending state = the ORDER quantity (dcy-021 D1/D4). # Entering PENDING_SELL takes the caller qty AS-IS so a full sell # (qty=None) records a "sell all" sentinel rather than the stale held # mirror; a partial-sell intent records its order qty. Other states keep # the caller-override-else-preserve rule. if to_state is SlotState.PENDING_SELL: new_qty = qty else: new_qty = qty if qty is not None else slot.qty # Generic update: caller-supplied fields override; None means "keep". return replace( slot, status=to_state, symbol=symbol if symbol is not None else slot.symbol, qty=new_qty, avg_price=avg_price if avg_price is not None else slot.avg_price, limit_price=new_limit_price, order_id=order_id if order_id is not None else slot.order_id, strategy_id=strategy_id if strategy_id is not None else slot.strategy_id, broker_order_id=( broker_order_id if broker_order_id is not None else slot.broker_order_id ), portfolio_id=( portfolio_id if portfolio_id is not None else slot.portfolio_id ), sell_strategy_id=( sell_strategy_id if sell_strategy_id is not None else slot.sell_strategy_id ), forced_close_at=( forced_close_at if forced_close_at is not None else slot.forced_close_at ), stop_loss_price=( stop_loss_price if stop_loss_price is not None else slot.stop_loss_price ), take_profit_price=( take_profit_price if take_profit_price is not None else slot.take_profit_price ), timeout_sec=new_timeout_sec, slippage_pct=new_slippage_pct, occupancy_seq=new_occupancy_seq, venue=new_venue, updated_at=datetime.now(tz=timezone.utc), ) # ------------------------------------------------------------ # sell (REQ-036 / BAT-010) # ------------------------------------------------------------ def sell( self, slot_id: UUID, actor: str, reason: str, *, limit_price: Optional[float] = None, ) -> None: """Trigger an autonomous safety-sell for a holding slot (REQ-036). Idempotent: silently returns if the slot is already pending_sell. Raises SlotNotFoundError if slot_id is unknown. Raises SlotTransitionError if the slot is neither holding nor pending_sell. dcy-012 / REQ-036: ``limit_price`` defaults to None so autonomous sells (safety-sell / kill-switch) dispatch at market; only callers that pass a price (discretionary RPC sell) get a limit order. """ slot = self._repo.fetch_slot(slot_id) if slot is None: raise SlotNotFoundError(f"slot not found: {slot_id}") if slot.status is SlotState.PENDING_SELL: return # already in progress — idempotent (REQ-036 AC-2) if slot.status is not SlotState.HOLDING: raise SlotTransitionError( f"sell() requires holding state, got: {slot.status.value}" ) self.transit_slot( slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason=reason, actor=actor, limit_price=limit_price, ) # ------------------------------------------------------------ # update_thresholds (REQ-040 / BAT-011) # ------------------------------------------------------------ def update_thresholds( self, slot_id: UUID, *, stop_loss_price: Any = _UNSET, take_profit_price: Any = _UNSET, forced_close_at: Any = _UNSET, actor: str = "strategy", ) -> str: """Partially update a holding slot's safety-sell thresholds (REQ-040). Each of the three fields is independent: pass `_UNSET` (the default — i.e. omit the kwarg) to keep the current value, or pass an explicit value (including ``None``) to set/clear it (AC-4 vs AC-5). Returns "updated" on success, or "ignored" when the slot is not in the holding state (AC-6). Raises SlotNotFoundError for unknown slots. The caller (RPC handler) resets the watchdog last_seen_at (AC-3). """ with self._repo.transactional_advisory_lock(self._lock_id): slot = self._repo.fetch_slot(slot_id) if slot is None: raise SlotNotFoundError(f"slot not found: {slot_id}") if slot.status is not SlotState.HOLDING: return "ignored" # not_holding (AC-6) updated = replace( slot, stop_loss_price=( slot.stop_loss_price if stop_loss_price is _UNSET else stop_loss_price ), take_profit_price=( slot.take_profit_price if take_profit_price is _UNSET else take_profit_price ), forced_close_at=( slot.forced_close_at if forced_close_at is _UNSET else forced_close_at ), updated_at=datetime.now(tz=timezone.utc), ) self._repo.update_slot(updated) return "updated" # ------------------------------------------------------------ # compute_slot_capacity # ------------------------------------------------------------ def compute_slot_capacity(self, max_active_slots: Optional[int] = None) -> SlotCapacity: """Return instance-scoped capacity counts (REQ-002 AC-6; dcy-020/REQ-059). active = holding + pending (position-occupying). Under a `max_active_slots` cap M, available = max(0, M − active) and total_slots = M; without a cap, available = inactive slot count and total_slots = the instance's slot count. scarcity_ratio = 1 − available/total, clamped to [0.0, 1.0]. """ counts = self._repo.count_slots_by_status(self._instance_key) inactive = counts.get(SlotState.INACTIVE, 0) holding = counts.get(SlotState.HOLDING, 0) pending = counts.get(SlotState.PENDING_BUY, 0) + counts.get(SlotState.PENDING_SELL, 0) active = holding + pending if max_active_slots is not None and max_active_slots > 0: total = max_active_slots available = max(0, max_active_slots - active) else: total = sum(counts.values()) available = inactive scarcity = ( max(0.0, min(1.0, 1.0 - (available / total))) if total > 0 else 0.0 ) return SlotCapacity( inactive_count=inactive, holding_count=holding, pending_count=pending, active_count=active, available_count=available, total_slots=total, scarcity_ratio=scarcity, )