trex-core/trex_core/domain/slot/service.py

"""Slot domain service — 4 main APIs (REQ-002).

- preallocate_daily_slots(trading_date, count)
- transit_slot(slot_id, from_state, to_state, reason, actor, **fields)
- get_today_entry_count(trading_date, symbol)
- compute_slot_capacity(trading_date, total_slots)

All state-mutating APIs run inside repository.transactional_advisory_lock()
to serialize Buy/Sell decision threads + operator CLI (REQ-002 AC-2).
"""
from __future__ import annotations

from dataclasses import replace
from datetime import date, 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 SLOT_ADVISORY_LOCK_ID, SlotState, is_allowed_transition

# 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) -> None:
        self._repo = repository

    # ------------------------------------------------------------
    # preallocate_daily_slots
    # ------------------------------------------------------------
    def preallocate_daily_slots(self, trading_date: date, count: int) -> int:
        """Insert `count` idle slots for trading_date if none exist.

        Idempotent: second+ call on the same trading_date returns 0.
        Returns number of slots inserted.

        REQ-002 AC-4, REQ-003 AC-5 (count must be > 0).
        """
        if count <= 0:
            raise ValueError(f"slot count must be > 0, got {count}")
        with self._repo.transactional_advisory_lock(SLOT_ADVISORY_LOCK_ID):
            return self._repo.pre_allocate_slots(trading_date, count)

    # ------------------------------------------------------------
    # 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,
        allocated_capital: 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,
        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
        6. on closed->idle, increment daily_symbol_entries (REQ-002 AC-3)

        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(SLOT_ADVISORY_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,
                allocated_capital=allocated_capital,
                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,
            )
            self._repo.update_slot(updated)

            # closed -> idle recovery: bump daily_symbol_entries atomically
            # (REQ-002 AC-3, REQ-001 split-by-purpose).
            if from_state is SlotState.CLOSED and to_state is SlotState.IDLE:
                # Use the symbol carried by the slot before reset.
                symbol_for_entry = slot.symbol
                if symbol_for_entry:
                    self._repo.increment_symbol_entry(slot.trading_date, symbol_for_entry)

            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 {},
                )
            )
            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],
        allocated_capital: 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,
    ) -> Slot:
        """Compute the post-transition Slot row.

        Special: closed -> idle resets occupancy fields (FR-2 #7).
        Special: pending_buy -> idle resets occupancy (REQ-011 AC-6 rollback).
        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
        (IDLE/HOLDING/CLOSED) clears it so a buy limit never leaks into the next
        sell order (issue #10 root cause).
        """
        if to_state is SlotState.IDLE and slot.status in {
            SlotState.CLOSED,
            SlotState.PENDING_BUY,
        }:
            # Recovery / rollback: clear occupancy + broker handle.
            # transition_log retains audit trail.
            return replace(
                slot,
                status=SlotState.IDLE,
                symbol=None,
                qty=None,
                avg_price=None,
                limit_price=None,
                allocated_capital=None,
                order_id=None,
                strategy_id=None,
                broker_order_id=None,
                portfolio_id=None,
                sell_strategy_id=None,
                forced_close_at=None,
                stop_loss_price=None,
                take_profit_price=None,
                updated_at=datetime.now(tz=timezone.utc),
            )

        # dcy-019 / REQ-002 (BAT-025): closed-slot direct reallocation. A closed
        # slot still carries the prior cycle's occupancy; start from a clean
        # slate so a previous position/broker handle never leaks into the new
        # entry. entry_count is intentionally preserved (slot-scoped cumulative).
        if to_state is SlotState.PENDING_BUY and slot.status is SlotState.CLOSED:
            slot = replace(
                slot,
                symbol=None, qty=None, avg_price=None, limit_price=None,
                allocated_capital=None, order_id=None, strategy_id=None,
                broker_order_id=None, portfolio_id=None, sell_strategy_id=None,
                forced_close_at=None, stop_loss_price=None, take_profit_price=None,
            )

        # limit_price lifecycle (dcy-012 FR-1): set only on entering a pending
        # state (the order's limit); cleared on entering HOLDING/CLOSED.
        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
        else:  # HOLDING / CLOSED — pending order limit no longer applies.
            new_limit_price = None

        # dcy-019 / REQ-002: +1 on each allocation transition (→pending_buy),
        # for both idle→pending_buy and closed→pending_buy (FR-4).
        new_entry_count = slot.entry_count + (1 if to_state is SlotState.PENDING_BUY else 0)

        # 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=qty if qty is not None else slot.qty,
            avg_price=avg_price if avg_price is not None else slot.avg_price,
            limit_price=new_limit_price,
            allocated_capital=(
                allocated_capital if allocated_capital is not None else slot.allocated_capital
            ),
            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
            ),
            entry_count=new_entry_count,  # dcy-019 / REQ-002
            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(SLOT_ADVISORY_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"

    # ------------------------------------------------------------
    # get_today_entry_count
    # ------------------------------------------------------------
    def get_today_entry_count(self, trading_date: date, symbol: str) -> int:
        """Return today's entry_count for a symbol (0 if no entries yet).

        Used by buy decision state gate to enforce re-entry policy.
        Reads daily_symbol_entries (decision input), NEVER queries
        slot_transition_log (audit only). REQ-001, dcy-001 5.3.5.5.
        """
        return self._repo.get_symbol_entry_count(trading_date, symbol)

    # ------------------------------------------------------------
    # compute_slot_capacity
    # ------------------------------------------------------------
    def compute_slot_capacity(self, trading_date: date, total_slots: int) -> SlotCapacity:
        """Return idle/holding/pending counts + scarcity_ratio (REQ-002 AC-6).

        scarcity_ratio = 1 - idle_count / total_slots, clamped to [0.0, 1.0].
        """
        if total_slots <= 0:
            raise ValueError(f"total_slots must be > 0, got {total_slots}")
        counts = self._repo.count_slots_by_status(trading_date)
        idle = counts.get(SlotState.IDLE, 0)
        holding = counts.get(SlotState.HOLDING, 0)
        pending = counts.get(SlotState.PENDING_BUY, 0) + counts.get(SlotState.PENDING_SELL, 0)
        scarcity = max(0.0, min(1.0, 1.0 - (idle / total_slots)))
        return SlotCapacity(
            idle_count=idle,
            holding_count=holding,
            pending_count=pending,
            total_slots=total_slots,
            scarcity_ratio=scarcity,
        )