"""Intent application service (REQ-007). Turns a buy/sell intent submitted over HTTP into a single-writer slot transition, with intent_id idempotency backed by a durable intent_log. Idempotency model (dcy-002 OQ-1): `intent_store.try_insert` is the dedup gate. On a duplicate intent_id it returns the prior decision without re-driving any transition (NFR-3). Slot transitions go exclusively through SlotService (advisory-lock serialized) so the single-writer invariant holds (NFR-2). Concurrency on the last idle slot is resolved by retrying on SlotStateMismatchError. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import date, datetime from decimal import Decimal from typing import Any, Optional from uuid import UUID from trex_core.application.ports import IntentStore from trex_core.domain.slot import ( SlotRepository, SlotService, SlotState, SlotStateMismatchError, ) _MAX_IDLE_RETRY = 8 @dataclass(frozen=True) class IntentDecision: """Result of submitting an intent.""" intent_id: str status: str # 'admitted' | 'rejected' slot_id: Optional[UUID] = None reason: Optional[str] = None @property def admitted(self) -> bool: return self.status == "admitted" @dataclass(frozen=True) class BuyIntent: # dcy-011 / REQ-007: portfolio_id (= asset_id) is the required canonical key. # symbol is no longer a client input — it is resolved server-side from the # universe (SnapshotStore) at admission. It remains here as the resolved # value carried onto the slot. intent_id: str trading_date: date portfolio_id: str symbol: Optional[str] = None qty: Optional[float] = None allocated_capital: Optional[float] = None entry_price: Optional[float] = None strategy_id: Optional[str] = None sell_strategy_id: Optional[str] = None forced_close_at: Optional[datetime] = None stop_loss_price: Optional[Decimal] = None # REQ-035 take_profit_price: Optional[Decimal] = None # REQ-035 extra: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True) class SellIntent: intent_id: str slot_id: UUID reason: str = "discretionary_sell" sell_strategy_id: Optional[str] = None # dcy-012 / REQ-007: optional sell limit price. None => market sell # (the path used by safety-sell / kill-switch). Set => limit sell at price. limit_price: Optional[float] = None extra: dict[str, Any] = field(default_factory=dict) class IntentService: """Admission + slot transition for buy/sell intents (single writer).""" def __init__( self, slot_service: SlotService, slot_repo: SlotRepository, intent_store: IntentStore, resolve_symbol=None, *, instance_key: str = "default", max_active_slots: Optional[int] = None, ) -> None: self._svc = slot_service self._repo = slot_repo self._intents = intent_store # dcy-011 / REQ-007: portfolio_id -> symbol resolver (e.g. SnapshotStore-backed). # callable(portfolio_id) -> Optional[str]. When set, admission resolves symbol # from the universe and rejects unknown_portfolio if absent. self._resolve_symbol = resolve_symbol # dcy-020 / REQ-059 (BAT-028): asset-centric mode gate. When max_active_slots # is set, buy admission activates the asset's permanent slot under an M cap # (FCFS, no eviction); otherwise the legacy pool path runs (unchanged). self._instance_key = instance_key self._max_active_slots = max_active_slots # ------------------------------------------------------------ def lookup(self, intent_id: str) -> Optional[IntentDecision]: """Return the recorded decision for an intent_id, or None (re-query).""" rec = self._intents.get(intent_id) if rec is None: return None return IntentDecision(rec.intent_id, rec.status, rec.slot_id, rec.reason) def submit_buy(self, intent: BuyIntent) -> IntentDecision: payload = { "trading_date": intent.trading_date.isoformat(), "symbol": intent.symbol, "qty": intent.qty, "allocated_capital": intent.allocated_capital, "entry_price": intent.entry_price, "strategy_id": intent.strategy_id, "sell_strategy_id": intent.sell_strategy_id, "portfolio_id": intent.portfolio_id, } prior = self._dedup(intent.intent_id, "buy", payload) if prior is not None: return prior decision = self._admit_buy(intent) self._intents.finalize( intent.intent_id, decision.status, decision.slot_id, decision.reason ) return decision def submit_sell(self, intent: SellIntent) -> IntentDecision: payload = { "slot_id": str(intent.slot_id), "reason": intent.reason, "sell_strategy_id": intent.sell_strategy_id, } prior = self._dedup(intent.intent_id, "sell", payload) if prior is not None: return prior decision = self._admit_sell(intent) self._intents.finalize( intent.intent_id, decision.status, decision.slot_id, decision.reason ) return decision # ------------------------------------------------------------ def _dedup(self, intent_id: str, intent_type: str, payload: dict) -> Optional[IntentDecision]: """Return prior decision if intent_id already seen; else None (and reserve it).""" inserted = self._intents.try_insert(intent_id, intent_type, payload) if inserted: return None rec = self._intents.get(intent_id) if rec is None: # extremely unlikely race; treat as not-yet-decided return IntentDecision(intent_id, "rejected", None, "duplicate_in_flight") return IntentDecision(intent_id, rec.status, rec.slot_id, rec.reason) def _admit_buy(self, intent: BuyIntent) -> IntentDecision: # dcy-011 / REQ-007: portfolio_id is required; symbol is resolved from the # universe (no client symbol). FR-1/FR-2/FR-3. if not intent.portfolio_id: return IntentDecision(intent.intent_id, "rejected", None, "portfolio_id_required") symbol = intent.symbol if self._resolve_symbol is not None: symbol = self._resolve_symbol(intent.portfolio_id) if not symbol: return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio") elif not symbol: # No resolver wired and no symbol carried — cannot determine the order target. return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio") # dcy-020 / REQ-059 (BAT-028): asset-centric admission when M is configured. if self._max_active_slots is not None: return self._admit_buy_asset_centric(intent, symbol) # Retry on the last-candidate race: another intent may claim the slot # between our find and transit (SlotStateMismatchError). # dcy-019 / REQ-007: admission candidate is idle ∪ closed (closed-slot # reallocation), ordered by entry_count asc → updated_at asc → slot_id. # The candidate may already be CLOSED, so the transit's from_state must # follow the slot's current status rather than a hardcoded IDLE. for _ in range(_MAX_IDLE_RETRY): slot = self._repo.find_allocatable_slot(intent.trading_date) if slot is None: return IntentDecision(intent.intent_id, "rejected", None, "no_idle_slot") try: updated = self._svc.transit_slot( slot.slot_id, slot.status, SlotState.PENDING_BUY, reason="buy_intent", actor="intent_api", symbol=symbol, qty=intent.qty, allocated_capital=intent.allocated_capital, # dcy-012 / REQ-007: buy entry price maps to the slot's # side-agnostic pending limit_price (cleared on fill→HOLDING). limit_price=intent.entry_price, strategy_id=intent.strategy_id, sell_strategy_id=intent.sell_strategy_id, portfolio_id=intent.portfolio_id, forced_close_at=intent.forced_close_at, stop_loss_price=intent.stop_loss_price, take_profit_price=intent.take_profit_price, extra={"intent_id": intent.intent_id, "symbol": symbol}, ) except SlotStateMismatchError: continue # slot was claimed concurrently; try another idle slot return IntentDecision(intent.intent_id, "admitted", updated.slot_id, None) return IntentDecision(intent.intent_id, "rejected", None, "idle_slot_contended") # ------------------------------------------------------------ def _admit_buy_asset_centric(self, intent: BuyIntent, symbol: str) -> IntentDecision: """dcy-020 / REQ-059: activate the asset's permanent slot under the M cap. FCFS + no eviction: if the asset is already active -> already_active; if the instance is at M active slots -> capacity_full (existing holdings untouched, incl. over-cap when M is lowered). Single intent thread + instance-scoped advisory lock serialize admissions (REQ-002 NFR-1 / dcy-020 OQ-1).""" _ACTIVE = (SlotState.PENDING_BUY, SlotState.HOLDING, SlotState.PENDING_SELL) for _ in range(_MAX_IDLE_RETRY): slot = self._repo.find_slot_by_portfolio(self._instance_key, intent.portfolio_id) if slot is None: # no asset slot for this instance (not in reconciled universe) return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio") if slot.status in _ACTIVE: return IntentDecision(intent.intent_id, "rejected", None, "already_active") if self._repo.count_active(self._instance_key) >= self._max_active_slots: return IntentDecision(intent.intent_id, "rejected", None, "capacity_full") try: updated = self._svc.transit_slot( slot.slot_id, slot.status, SlotState.PENDING_BUY, reason="buy_intent", actor="intent_api", symbol=symbol, qty=intent.qty, allocated_capital=intent.allocated_capital, limit_price=intent.entry_price, strategy_id=intent.strategy_id, sell_strategy_id=intent.sell_strategy_id, portfolio_id=intent.portfolio_id, forced_close_at=intent.forced_close_at, stop_loss_price=intent.stop_loss_price, take_profit_price=intent.take_profit_price, extra={"intent_id": intent.intent_id, "symbol": symbol}, ) except SlotStateMismatchError: continue # slot changed under us; re-fetch and re-evaluate return IntentDecision(intent.intent_id, "admitted", updated.slot_id, None) return IntentDecision(intent.intent_id, "rejected", None, "already_active") def _admit_sell(self, intent: SellIntent) -> IntentDecision: slot = self._repo.fetch_slot(intent.slot_id) if slot is None: return IntentDecision(intent.intent_id, "rejected", None, "slot_not_found") if slot.status is not SlotState.HOLDING: return IntentDecision( intent.intent_id, "rejected", None, f"slot_not_holding:{slot.status.value}" ) try: updated = self._svc.transit_slot( slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason=intent.reason, actor="intent_api", sell_strategy_id=intent.sell_strategy_id, # dcy-012 / REQ-007: optional sell limit (None => market sell). limit_price=intent.limit_price, extra={"intent_id": intent.intent_id}, ) except SlotStateMismatchError: return IntentDecision(intent.intent_id, "rejected", None, "slot_state_changed") return IntentDecision(intent.intent_id, "admitted", updated.slot_id, None)