"""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). BAT-050 / REQ-104: reconcile-warning gate added to _admit_buy/_admit_sell. warn_store is Optional — when None, gate is skipped (backward compat). Safety-sell / kill-switch / watchdog bypass this gate by design (OQ-B): those paths call SlotService.sell() directly, not through IntentService. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from decimal import Decimal from typing import Any, Callable, Optional from uuid import UUID from trex_core.application.ports import IntentStore from trex_core.domain.slot import ( SlotAdmissionRejectedError, SlotRepository, SlotService, SlotState, SlotStateMismatchError, ) _MAX_ADMIT_RETRY = 8 _HARD_ORDER_TIMEOUT_SEC = 3600 @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 # Missing strategy venue uses KIS SOR; explicit KRX/NXT/SOR is preserved. venue: str = "SOR" 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 venue: str = "SOR" 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, warn_store=None, # Optional[ReconcileWarningStore] — BAT-050 / REQ-104 entry_allowed_fn: Optional[Callable[[], bool]] = None, buy_guard_fn: Optional[Callable[[BuyIntent], Optional[str]]] = 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 # Omission never disables timeout. Runtime validates configured values; # this seam remains fail-safe for direct application construction. self._default_order_timeout_sec = ( default_order_timeout_sec if isinstance(default_order_timeout_sec, int) and not isinstance(default_order_timeout_sec, bool) and 1 <= default_order_timeout_sec <= _HARD_ORDER_TIMEOUT_SEC else _HARD_ORDER_TIMEOUT_SEC ) # BAT-050 / REQ-104: reconcile-warning gate store (Optional). # When None, gate is skipped (backward compat, NFR-4). self._warn_store = warn_store # dcy-036 / REQ-114: day-start entry gate (Optional callable → bool). # When None or returns True, gate passes (backward compat / opt-in=False). # When returns False, BuyIntent is rejected with reason 'day_start_load_failed'. # SellIntent is NOT gated by this (REQ-114 AC-5: 청산 경로 영향 없음). self._entry_allowed_fn: Optional[Callable[[], bool]] = entry_allowed_fn # dcy-041 / GitHub #45: opt-in bounded cash-capacity admission guard. # The callable returns a stable rejection reason or None. SellIntent is # deliberately outside this gate. self._buy_guard_fn = buy_guard_fn # ------------------------------------------------------------ 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, "venue": intent.venue, } 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, "venue": intent.venue, } 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). dcy-036 / REQ-114: entry gate — if entry_allowed_fn returns False, BuyIntent is rejected before any slot lookup.""" timeout_error = self._timeout_error(intent.timeout_sec) if timeout_error is not None: return IntentDecision(intent.intent_id, "rejected", None, timeout_error) if intent.venue not in {"KRX", "NXT", "SOR"}: return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") # dcy-036 / REQ-114: day-start entry gate (fail-closed). # None or True → pass; False → reject new entry. # SellIntent is NOT checked (청산 경로는 차단하지 않음 AC-5). if self._entry_allowed_fn is not None: try: if not self._entry_allowed_fn(): return IntentDecision( intent.intent_id, "rejected", None, "day_start_load_failed" ) except Exception: # noqa: BLE001 - gate failure → fail-open (conservative) pass 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) # BAT-050 / REQ-104: reconcile-warning gate (external strategy intent only). # Safety-sell / kill-switch / watchdog bypass this gate by design (OQ-B): # those paths call SlotService.sell() directly. if self._warn_store is not None: try: if self._warn_store.is_warned(str(slot.slot_id)): return IntentDecision( intent.intent_id, "rejected", None, "reconcile_warned" ) except Exception: # noqa: BLE001 - fail-open pass # 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") def _admission_precondition(_locked_slot) -> Optional[str]: # Runs inside SlotService's instance-scoped advisory lock. This # makes capacity and cached-cash reservation checks atomic with # the PENDING_BUY transition across same-instance daemons. 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 "capacity_full" if self._buy_guard_fn is None: return None try: return self._buy_guard_fn(intent) except Exception: # noqa: BLE001 - buy guard fails closed return "account_snapshot_unavailable" # 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, venue=intent.venue, extra={"intent_id": intent.intent_id, "symbol": symbol}, precondition=_admission_precondition, ) except SlotAdmissionRejectedError as exc: return IntentDecision( intent.intent_id, "rejected", None, exc.reason ) 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: timeout_error = self._timeout_error(intent.timeout_sec) if timeout_error is not None: return IntentDecision(intent.intent_id, "rejected", None, timeout_error) if intent.venue not in {"KRX", "NXT", "SOR"}: return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") slot = self._repo.fetch_slot(intent.slot_id) if slot is None: return IntentDecision(intent.intent_id, "rejected", None, "slot_not_found") # BAT-050 / REQ-104: reconcile-warning gate for external strategy sell intent. # Internal safety-sell / kill-switch / watchdog bypass this gate: they call # SlotService.sell() / transit_slot() directly without IntentService (OQ-B). if self._warn_store is not None: try: if self._warn_store.is_warned(str(slot.slot_id)): return IntentDecision( intent.intent_id, "rejected", None, "reconcile_warned" ) except Exception: # noqa: BLE001 - fail-open pass 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, venue=intent.venue, 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) @staticmethod def _timeout_error(timeout_sec: object) -> Optional[str]: if timeout_sec is None: return None if not isinstance(timeout_sec, int) or isinstance(timeout_sec, bool): return "bad_request:timeout_sec_must_be_integer" if not 1 <= timeout_sec <= _HARD_ORDER_TIMEOUT_SEC: return "bad_request:timeout_sec_out_of_range" return None