"""Intent application service (REQ-007; dcy-020/REQ-059·061 asset-centric). Turns a buy/sell intent submitted over messaging 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). dcy-020 (BAT-030): admission is asset-centric only — the legacy pool path (daily preallocation + idle∪closed candidate selection) was removed at the cutover. Buy is keyed by slot_id (REQ-061 canonical; portfolio_id is the legacy dual-accept mapping) and admits FCFS under the `max_active_slots` cap M when configured (no cap when unset). """ from __future__ import annotations from dataclasses import dataclass, field from datetime import 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_ADMIT_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-020 / REQ-061: slot_id is the canonical buy key (snapshot slots[] # round-trip); portfolio_id is accepted as a legacy dual-accept mapping. # symbol is not a client input — it is resolved server-side from the # universe at admission and carried onto the slot. intent_id: str portfolio_id: Optional[str] = None slot_id: Optional[UUID] = None symbol: Optional[str] = None qty: 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 # dcy-022 / REQ-068: execution policy (optional, backward-compat). # timeout_sec: cancel_order after N seconds without fill (None = config default or OFF). # slippage_pct: effective limit = entry_price*(1+pct/100) buy / *(1-pct/100) sell. timeout_sec: Optional[int] = None slippage_pct: Optional[float] = None 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 # dcy-021 / REQ-062: optional partial-sell quantity. None => sell the whole # position (broker balance); set => partial sell of min(qty, broker balance). qty: Optional[float] = None # dcy-022 / REQ-068: execution policy (optional, backward-compat). timeout_sec: Optional[int] = None slippage_pct: 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, default_order_timeout_sec: 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·060: M cap. None = no capacity cap (the operator is # expected to set max_active_slots in the instance TOML — design §5). self._instance_key = instance_key self._max_active_slots = max_active_slots # dcy-022 / REQ-068: default timeout when intent omits timeout_sec. # None = OFF = indefinite wait (legacy behavior). Per-intent value overrides. self._default_order_timeout_sec = default_order_timeout_sec # ------------------------------------------------------------ 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 = { "slot_id": str(intent.slot_id) if intent.slot_id else None, "portfolio_id": intent.portfolio_id, "symbol": intent.symbol, "qty": intent.qty, "entry_price": intent.entry_price, "strategy_id": intent.strategy_id, "sell_strategy_id": intent.sell_strategy_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, "qty": intent.qty, } 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-020 / REQ-059·REQ-061 + dcy-021/REQ-066: activate or size up the asset's permanent slot. Contract key is slot_id (REQ-061); portfolio_id is accepted as a legacy dual-accept mapping. Symbol is resolved from the slot's portfolio_id. - INACTIVE slot -> initial buy (FCFS, M cap applies, no eviction). - HOLDING slot -> position-sizing INCREASE (dcy-021): adds to the held position; M cap does NOT apply (active asset count unchanged). - PENDING_* slot -> in-flight conflict (already_active). Single intent thread + instance-scoped advisory lock serialize admissions (dcy-020 OQ-1).""" def _resolve(sl): pid = sl.portfolio_id if self._resolve_symbol is not None and pid: return self._resolve_symbol(pid) return intent.symbol or sl.symbol for _ in range(_MAX_ADMIT_RETRY): # resolve the target slot: slot_id (canonical) or portfolio_id (legacy) if intent.slot_id is not None: slot = self._repo.fetch_slot(intent.slot_id) miss_reason = "unknown_slot" elif intent.portfolio_id: slot = self._repo.find_slot_by_portfolio(self._instance_key, intent.portfolio_id) miss_reason = "unknown_portfolio" else: return IntentDecision(intent.intent_id, "rejected", None, "slot_id_or_portfolio_required") if slot is None: return IntentDecision(intent.intent_id, "rejected", None, miss_reason) # dcy-021: a buy on a HOLDING slot is a position-sizing increase. is_increase = slot.status is SlotState.HOLDING if slot.status in (SlotState.PENDING_BUY, SlotState.PENDING_SELL): return IntentDecision(intent.intent_id, "rejected", None, "already_active") if slot.status not in (SlotState.INACTIVE, SlotState.HOLDING): return IntentDecision( intent.intent_id, "rejected", None, f"slot_not_buyable:{slot.status.value}" ) symbol = _resolve(slot) if not symbol: return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio") # M cap applies only to NEW activations (initial buy). An increase on # an already-active asset does not change the active asset count. if (not is_increase and self._max_active_slots is not None and self._repo.count_active(self._instance_key) >= self._max_active_slots): return IntentDecision(intent.intent_id, "rejected", None, "capacity_full") # dcy-022 / REQ-068: resolve effective timeout (per-intent > config default). effective_timeout = ( intent.timeout_sec if intent.timeout_sec is not None else self._default_order_timeout_sec ) try: updated = self._svc.transit_slot( slot.slot_id, slot.status, SlotState.PENDING_BUY, reason="position_increase" if is_increase else "buy_intent", actor="intent_api", symbol=symbol, qty=intent.qty, # 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, timeout_sec=effective_timeout, slippage_pct=intent.slippage_pct, 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}" ) # dcy-022 / REQ-068: resolve effective timeout. effective_timeout = ( intent.timeout_sec if intent.timeout_sec is not None else self._default_order_timeout_sec ) 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, # dcy-021 / REQ-062: partial-sell order qty (None => sell all); # dispatch caps at the broker balance. Entering PENDING_SELL takes # this qty as-is (None = "sell all" sentinel). qty=intent.qty, timeout_sec=effective_timeout, slippage_pct=intent.slippage_pct, 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)