"""RPC handlers + event payload builders for messaging transport (REQ-021/022). Handlers are transport-agnostic: they accept a parsed `RpcRequest` payload and return an `RpcResponse`. Wiring them onto `MessagingTransport.consume_*` happens in the runtime layer. """ from __future__ import annotations import logging from datetime import date as date_cls from datetime import datetime, timezone from typing import Optional, Protocol from uuid import UUID from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.application.account_snapshot import AccountSnapshotUnavailable from trex_core.application.broker_port import BalanceItem from trex_core.application.messaging import RpcRequest, RpcResponse from trex_core.application.portfolio import ( BLOCKING_RECONCILE_CASES, PauseStore, ReconcileWarningStore, SnapshotStore, slot_to_dict, ) from trex_core.application.watchdog import WatchdogStore from trex_core.domain.slot import ( SlotNotFoundError, SlotRepository, SlotService, ) _log = logging.getLogger(__name__) # ---- intent RPC handler ----------------------------------------------------- def make_intent_rpc_handler( intent_service: IntentService, *, slot_service: Optional[SlotService] = None, watchdog_store: Optional[WatchdogStore] = None, ): """Build the RPC handler for `trex.intent.req`. Payload shape: { "intent_type": "buy" | "sell", "intent_id": "", ... fields per BuyIntent / SellIntent ... } or (REQ-040): {"action": "update_thresholds", "slot_id": ..., "stop_loss_price"?, "take_profit_price"?, "forced_close_at"?} Response shape: {"intent_id": str, "status": "admitted"|"rejected", "slot_id": str|None, "reason": str|None} or for update_thresholds: {"status": "updated"|"ignored"|"error", "slot_id": str, "reason"?: str} Dedup uses `intent_log.try_insert(intent_id, ...)` (REQ-007 gate). The request's `correlation_id` is reused as `intent_id` so the messaging layer and the application share a single idempotency source of truth (REQ-023 AC-1). """ def handler(req: RpcRequest) -> RpcResponse: body = req.payload or {} # REQ-040: update_thresholds action (separate from buy/sell intents). if str(body.get("action") or "").lower() == "update_thresholds": return _handle_update_thresholds(body, slot_service, watchdog_store) intent_id = str(body.get("intent_id") or req.correlation_id) intent_type = str(body.get("intent_type") or "").lower() try: if intent_type == "buy": # dcy-020 / REQ-061: buy is keyed by slot_id (canonical); # portfolio_id is the legacy dual-accept mapping. symbol is NOT # a client field — resolved from the universe. trading_date was # dropped from the contract at the cutover (BAT-030). buy = BuyIntent( intent_id=intent_id, portfolio_id=_opt_str(body.get("portfolio_id")) or None, slot_id=(UUID(str(body["slot_id"])) if body.get("slot_id") else None), qty=_opt_float(body.get("qty")), # dcy-023 / REQ-007 AC-6: allocated_capital is no longer a # contract field — senders may still include it; it is ignored. entry_price=_opt_float(body.get("entry_price")), strategy_id=_opt_str(body.get("strategy_id")), sell_strategy_id=_opt_str(body.get("sell_strategy_id")), forced_close_at=_opt_dt(body.get("forced_close_at")), stop_loss_price=_opt_decimal(body.get("stop_loss_price")), take_profit_price=_opt_decimal(body.get("take_profit_price")), # dcy-022 / REQ-068: execution policy (optional, backward-compat). timeout_sec=_opt_int(body.get("timeout_sec")), slippage_pct=_opt_float(body.get("slippage_pct")), venue=_opt_str(body.get("venue")) or "SOR", ) d = intent_service.submit_buy(buy) elif intent_type == "sell": sell = SellIntent( intent_id=intent_id, slot_id=UUID(str(body["slot_id"])), reason=str(body.get("reason") or "discretionary_sell"), sell_strategy_id=_opt_str(body.get("sell_strategy_id")), # dcy-012 / REQ-007: optional sell limit price; absent => market sell. limit_price=_opt_float(body.get("limit_price")), # dcy-021 / REQ-062: optional partial-sell qty; absent => sell all. qty=_opt_float(body.get("qty")), # dcy-022 / REQ-068: execution policy (optional, backward-compat). timeout_sec=_opt_int(body.get("timeout_sec")), slippage_pct=_opt_float(body.get("slippage_pct")), venue=_opt_str(body.get("venue")) or "SOR", ) d = intent_service.submit_sell(sell) else: return RpcResponse(payload={ "intent_id": intent_id, "status": "rejected", "slot_id": None, "reason": f"unknown_intent_type:{intent_type}", }) except (KeyError, ValueError) as exc: payload = { "intent_id": intent_id, "status": "rejected", "slot_id": None, "reason": f"bad_request:{exc}", } return RpcResponse(payload=payload) payload = { "intent_id": d.intent_id, "status": d.status, "slot_id": (str(d.slot_id) if d.slot_id else None), "reason": d.reason, } return RpcResponse(payload=payload) return handler def _handle_update_thresholds( body: dict, slot_service: Optional[SlotService], watchdog_store: Optional[WatchdogStore], ) -> RpcResponse: """REQ-040: update stop_loss/take_profit/forced_close_at on a holding slot. Only the keys actually present in the payload are forwarded, so an absent field keeps its value while an explicit ``null`` clears it (AC-4 vs AC-5). On success the slot's watchdog last_seen_at is reset (AC-3) — receiving an update_thresholds is itself a strategy-module liveness signal. """ if slot_service is None: return RpcResponse(payload={"status": "error", "reason": "update_thresholds_unsupported"}) try: slot_id = UUID(str(body["slot_id"])) except (KeyError, ValueError) as exc: return RpcResponse(payload={"status": "error", "reason": f"bad_request:{exc}"}) kwargs: dict = {} if "stop_loss_price" in body: kwargs["stop_loss_price"] = _opt_decimal(body["stop_loss_price"]) if "take_profit_price" in body: kwargs["take_profit_price"] = _opt_decimal(body["take_profit_price"]) if "forced_close_at" in body: kwargs["forced_close_at"] = _opt_dt(body["forced_close_at"]) actor = str(body.get("actor") or "strategy") try: status = slot_service.update_thresholds(slot_id, actor=actor, **kwargs) except SlotNotFoundError: return RpcResponse(payload={ "status": "ignored", "slot_id": str(slot_id), "reason": "slot_not_found", }) if status == "updated" and watchdog_store is not None: watchdog_store.update(str(slot_id)) # AC-3: liveness signal payload = {"status": status, "slot_id": str(slot_id)} if status == "ignored": payload["reason"] = "not_holding" # AC-6 return RpcResponse(payload=payload) # ---- slots RPC handler (REQ-073/074) ---------------------------------------- # Maps a `trex.slots.req` filter token to the set of slot status values it # selects (issue #25 — pushed to the repository via list_slots status_filter). # "all" and unrecognized tokens map to None (no filter — REQ-073 AC-2). _FILTER_STATUS: dict[str, set[str]] = { "inactive": {"inactive"}, "pending": {"pending_buy", "pending_sell"}, "holding": {"holding"}, "error": {"error"}, } class SlotsBrokerQuantityReader(Protocol): def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ... def make_slots_rpc_handler( slot_repo: SlotRepository, *, instance_key: str = "default", snapshot_store: Optional[SnapshotStore] = None, pause_store: Optional[PauseStore] = None, warn_store: Optional[ReconcileWarningStore] = None, watchdog_store: Optional[WatchdogStore] = None, watchdog_timeout_sec: int = 120, ledger_outbox=None, max_active_slots: Optional[int] = None, strategy_state_store=None, broker_quantity_reader: Optional[SlotsBrokerQuantityReader] = None, ): """Build the RPC handler for `trex.slots.req` (REQ-073/074/116). Request: {filter?: str, spy?: bool, strategy?: str, strategy_instance_id?: str} filter: "all"|"inactive"|"pending"|"holding"|"error" (default "all") spy=true: watchdog not updated (operator view, REQ-034 CHG-01) strategy: exact-match on strategy_name (REQ-116, optional). When specified only slots whose strategy_name == strategy are returned. AND-combined with filter. available_slots is always computed before this filter (REQ-073 AC-6). Response: {slots: [...], available_slots: int|null, degraded: bool, reason: str|null, message: str|null} available_slots: capacity hint — null when uncapped, max(0, M-active) when capped (REQ-073 AC-6). Does not replace the admission gate. """ def handler(req: RpcRequest) -> RpcResponse: body = req.payload or {} filter_val = str(body.get("filter") or "all").lower() spy_value = body.get("spy", False) if not isinstance(spy_value, bool): return RpcResponse(payload={ "slots": [], "available_slots": None, "degraded": False, "reason": "bad_request:spy", "message": "spy must be a boolean", }) spy = spy_value strategy_instance_id = body.get("strategy_instance_id") if not spy and (not isinstance(strategy_instance_id, str) or not strategy_instance_id): return RpcResponse(payload={ "slots": [], "available_slots": None, "degraded": False, "reason": "bad_request:strategy_instance_id", "message": "strategy_instance_id must be a non-empty string", }) # REQ-116 (BAT-055): optional strategy filter — exact-match on strategy_name. # None means no filter (all slots returned regardless of strategy_name). strategy_filter = _opt_str(body.get("strategy")) # issue #25: filter by status at the DB (REQ-002/006 list_slots # status_filter) instead of fetching every row and filtering in Python. # "all"/unrecognized → None (no filter, REQ-073 AC-2). status_filter = _FILTER_STATUS.get(filter_val) slots = slot_repo.list_slots(instance_key, status_filter=status_filter) # REQ-034 CHG-01: update watchdog for holding slots unless spy=True. # When the fetched set already contains every holding slot (filter "all" # or "holding"), reuse it and skip the second connection (issue #25 (2)). if not spy and watchdog_store is not None: if status_filter is None or filter_val == "holding": holding_slots = [s for s in slots if s.status.value == "holding"] else: holding_slots = slot_repo.list_holding_slots(instance_key) for s in holding_slots: watchdog_store.update(str(s.slot_id)) warning_cases_by_slot: dict[str, set[str]] = {} if warn_store is not None: try: warned_slot_ids = warn_store.list_warned() for sid in warned_slot_ids: info = warn_store.get_warning_info(sid) if info is None: raise RuntimeError(f"missing warning info for slot_id={sid}") raw_codes = info.get("case_codes") if isinstance(raw_codes, list): codes = {str(code) for code in raw_codes if code} else: case_code = str(info.get("case_code", "")) codes = {case_code} if case_code else set() if not codes: raise RuntimeError(f"empty warning cases for slot_id={sid}") if not codes.issubset(BLOCKING_RECONCILE_CASES): raise RuntimeError(f"unsupported warning cases for slot_id={sid}") warning_cases_by_slot[str(sid)] = codes except Exception: # noqa: BLE001 _log.exception("warn_store warning query failed — fail-closed") return RpcResponse(payload={ "slots": [], "available_slots": None, "degraded": True, "reason": "warn_store_unavailable", "message": ( "reconcile warning store unavailable; " "returning no slots (fail-closed)" ), "excluded_slots": { "reconcile_quantity_mismatch": 0, "active_reconcile_warning_cases": {}, }, }) # Build merged slot rows (REQ-074) slot_rows = [ slot_to_dict( s, snapshot_store=snapshot_store, watchdog_store=watchdog_store, pause_store=pause_store, warning_cases_by_slot=warning_cases_by_slot, watchdog_timeout_sec=watchdog_timeout_sec, ) for s in slots ] # Exclusion must precede all optional data readers so blocked slots cannot # expose state or make an otherwise valid response fail. excluded_count = 0 excluded_case_counts: dict[str, int] = {} if warn_store is not None: included: list[tuple[object, dict]] = [] for slot, row in zip(slots, slot_rows, strict=True): blocked_holding = ( slot.status.value == "holding" and bool(set(row.get("reconcile_warning_cases", [])) & BLOCKING_RECONCILE_CASES) ) if blocked_holding: excluded_count += 1 for code in row.get("reconcile_warning_cases", []): excluded_case_counts[code] = excluded_case_counts.get(code, 0) + 1 continue included.append((slot, row)) if included: slots = [slot for slot, _ in included] slot_rows = [row for _, row in included] else: slots = [] slot_rows = [] broker_qty_by_symbol: dict[str, int] = {} broker_qty_available = False broker_qty_captured_at: Optional[datetime] = None if broker_quantity_reader is not None and slots: try: holdings, broker_qty_captured_at = broker_quantity_reader.holdings_for_slots_rpc() broker_qty_by_symbol = { str(item.symbol): int(item.hldg_qty) for item in holdings } broker_qty_available = True except AccountSnapshotUnavailable: broker_qty_available = False for row, slot in zip(slot_rows, slots, strict=True): row["slot_quantity"] = row.get("qty") row["qty_source"] = None row["broker_quantity_available"] = False row["broker_quantity_captured_at"] = None if slot.status.value != "holding": continue if broker_qty_available: broker_qty = int(broker_qty_by_symbol.get(str(slot.symbol), 0)) row["qty"] = float(broker_qty) row["qty_source"] = "broker" row["broker_quantity_available"] = True row["broker_quantity_captured_at"] = broker_qty_captured_at.isoformat() else: row["qty"] = None row["qty_source"] = "unavailable" # Strategy requests receive only their exact entry; operator requests # receive the complete slot map without changing watchdog behavior. if strategy_state_store is not None and slots: slot_ids = [s.slot_id for s in slots] state_map = strategy_state_store.get_for_slots(slot_ids) for row, s in zip(slot_rows, slots, strict=True): entries = state_map.get(s.slot_id) if entries: if spy: row["strategy_state"] = { instance_id: { "version": e.version, "data": e.data, "updated_at": e.updated_at.isoformat(), } for instance_id, e in entries.items() } elif (entry := entries.get(strategy_instance_id)) is not None: row["strategy_state"] = { "version": entry.version, "data": entry.data, "updated_at": entry.updated_at.isoformat(), } # dcy-018 / REQ-029: cash-estimated overlay (best-effort) if ledger_outbox is not None: try: deltas = ledger_outbox.pending_cash_delta() if deltas: for row in slot_rows: pid = row.get("portfolio_id") d = deltas.get(pid) if d: base = row.get("cash_balance") or 0.0 row["cash_balance"] = base + d row["cash_estimated"] = True except Exception: # noqa: BLE001 - estimate must never break the response pass # Degraded state from snapshot store degraded = False reason = None message = None if snapshot_store is not None: reason_code, msg = snapshot_store.degraded_state() degraded = reason_code is not None reason = reason_code.value if reason_code is not None else None message = msg # REQ-073 AC-6: available_slots computed from instance-wide active count — # independent of strategy filter (REQ-116). Calculate before filtering. if max_active_slots is not None and max_active_slots > 0: available_slots = max(0, max_active_slots - slot_repo.count_active(instance_key)) else: available_slots = None # REQ-116 (BAT-055): strategy filter — Python-level post-filter after # available_slots is already computed. AND-combined with status filter. # Unmatched strategy (no slots) returns [] without error. if strategy_filter is not None: slot_rows = [r for r in slot_rows if r.get("strategy_name") == strategy_filter] base_payload: dict = { "slots": slot_rows, "available_slots": available_slots, "degraded": degraded, "reason": reason, "message": message, } if warn_store is not None: base_payload["excluded_slots"] = { "reconcile_quantity_mismatch": excluded_count, "active_reconcile_warning_cases": excluded_case_counts, } return RpcResponse(payload=base_payload) return handler # ---- strategy-state write RPC handler (dcy-028 / REQ-080) ------------------- def make_strategy_state_set_handler( slot_repo: SlotRepository, strategy_state_store, ): """Build the RPC handler for `trex.slots.strategy_state.set` (REQ-080). Request: {slot_id, strategy_instance_id, expected_version:int, data:str} Response (ok): {ok:true, slot_id, strategy_instance_id, version:int} Response (fail): {ok:false, slot_id, strategy_instance_id, reason, version?} The store owns entry-version CAS; the handler supplies active-ness and never interprets opaque data. """ _ACTIVE_STATES = {"pending_buy", "holding", "pending_sell"} def handler(req: RpcRequest) -> RpcResponse: body = req.payload or {} try: slot_id = UUID(str(body["slot_id"])) strategy_instance_id = body["strategy_instance_id"] if not isinstance(strategy_instance_id, str) or not strategy_instance_id: raise ValueError("strategy_instance_id must be a non-empty string") expected_version = int(body["expected_version"]) data = body["data"] if not isinstance(data, str): raise ValueError("data must be a string") except (KeyError, ValueError, TypeError) as exc: return RpcResponse(payload={ "ok": False, "slot_id": body.get("slot_id"), "strategy_instance_id": body.get("strategy_instance_id"), "reason": f"bad_request:{exc}", }) slot = slot_repo.fetch_slot(slot_id) slot_active = slot is not None and slot.status.value in _ACTIVE_STATES result = strategy_state_store.set( slot_id, strategy_instance_id, expected_version=expected_version, data=data, slot_active=slot_active, ) payload = { "ok": result.ok, "slot_id": str(slot_id), "strategy_instance_id": strategy_instance_id, } if result.ok: payload["version"] = result.version else: payload["reason"] = result.reason if result.version is not None: payload["version"] = result.version return RpcResponse(payload=payload) return handler # ---- pause/resume RPC handler (REQ-032 AC-4) -------------------------------- def make_pause_rpc_handler(pause_store: PauseStore): """RPC handler for the `trex.pause.req` control queue. Payload: {"action": "pause" | "resume" | "list", "portfolio_id": "..."} Response: {"status": "ok"|"error", "paused": [list-of-portfolio_id], "reason"?: str} """ def handler(req: RpcRequest) -> RpcResponse: body = req.payload or {} action = str(body.get("action") or "").lower() pid = _opt_str(body.get("portfolio_id")) actor = str(body.get("actor") or "operator") try: if action == "pause": if not pid: return _err("portfolio_id required") pause_store.pause(pid, actor=actor) elif action == "resume": if not pid: return _err("portfolio_id required") pause_store.resume(pid) elif action == "list": pass else: return _err(f"unknown action: {action!r}") except Exception as exc: # noqa: BLE001 return _err(f"pause store error: {exc.__class__.__name__}") return RpcResponse(payload={ "status": "ok", "paused": sorted(pause_store.list_paused()), "action": action, }) return handler def _err(reason: str) -> RpcResponse: return RpcResponse(payload={"status": "error", "reason": reason}) # ---- control RPC handler (REQ-043, BAT-012) --------------------------------- def make_control_rpc_handler( pause_store: PauseStore, *, kill_switch_service=None, cancel_pending_fn=None, ): """RPC handler for `trex.control.req` — action discriminator dispatch (REQ-043). Actions: `pause`/`resume`/`list` (REQ-032, delegated), `kill_switch` (REQ-045), `resume_kill_switch` (REQ-046), `cancel_pending` (dcy-022/REQ-071). Unknown actions return an error response. cancel_pending_fn: callable(slot_id: str) -> str — injected from ExecutionService. """ pause_handler = make_pause_rpc_handler(pause_store) def handler(req: RpcRequest) -> RpcResponse: body = req.payload or {} action = str(body.get("action") or "").lower() if action == "kill_switch": return _handle_kill_switch(body, kill_switch_service) if action == "resume_kill_switch": return _handle_resume_kill_switch(body, kill_switch_service) if action in ("pause", "resume", "list"): return pause_handler(req) if action == "cancel_pending": return _handle_cancel_pending(body, cancel_pending_fn) return _err(f"unknown_action:{action}") return handler def _killswitch_scope(body: dict): """Validate scope/portfolio_id. Returns (scope, pid) or (None, RpcResponse error).""" scope = str(body.get("scope") or "").lower() pid = _opt_str(body.get("portfolio_id")) if scope not in ("portfolio", "global"): return None, _err(f"bad_scope:{scope}") if scope == "portfolio" and not pid: return None, _err("portfolio_id_required") return (scope, pid), None def _handle_kill_switch(body: dict, svc) -> RpcResponse: """REQ-045: liquidate holding slots in scope + day-exclude.""" if svc is None: return _err("kill_switch_unsupported") parsed, err = _killswitch_scope(body) if err is not None: return err scope, pid = parsed actor = str(body.get("actor") or "operator") try: liquidated = svc.fire(scope=scope, portfolio_id=pid, actor=actor) except Exception as exc: # noqa: BLE001 return _err(f"kill_switch_failed:{exc.__class__.__name__}") payload = {"status": "ok", "scope": scope, "liquidated": liquidated} if pid: payload["portfolio_id"] = pid return RpcResponse(payload=payload) def _handle_cancel_pending(body: dict, cancel_fn) -> RpcResponse: """dcy-022 / REQ-071: operator backstop cancel for a pending slot.""" if cancel_fn is None: return _err("cancel_pending_unsupported") try: slot_id = str(body["slot_id"]) except KeyError: return _err("slot_id_required") result = cancel_fn(slot_id) if result == "ok": return RpcResponse(payload={"status": "ok", "slot_id": slot_id}) if result == "not_pending": return RpcResponse(payload={"status": "ignored", "slot_id": slot_id, "reason": "not_pending"}) if result == "slot_not_found": return RpcResponse(payload={"status": "error", "slot_id": slot_id, "reason": "slot_not_found"}) return RpcResponse(payload={"status": "error", "slot_id": slot_id, "reason": result}) def _handle_resume_kill_switch(body: dict, svc) -> RpcResponse: """REQ-046: clear the day-scoped exclusion (sold slots stay sold).""" if svc is None: return _err("resume_kill_switch_unsupported") parsed, err = _killswitch_scope(body) if err is not None: return err scope, pid = parsed try: svc.resume(scope=scope, portfolio_id=pid) except Exception as exc: # noqa: BLE001 return _err(f"resume_kill_switch_failed:{exc.__class__.__name__}") payload = {"status": "ok", "scope": scope} if pid: payload["portfolio_id"] = pid return RpcResponse(payload=payload) # ---- event payload builder -------------------------------------------------- def slot_transition_event_payload(slot, *, event_type: str) -> dict: """Build the event payload for a slot transition (REQ-023 AC-5). Carries the *current state* + a monotonic sequence (`updated_at` iso) so receivers can apply last-wins without ordering guarantees. """ return { "event_type": event_type, "slot_id": str(slot.slot_id), "strategy_id": slot.strategy_id, "symbol": slot.symbol, "status": slot.status.value, "qty": float(slot.qty) if slot.qty is not None else None, "avg_price": float(slot.avg_price) if slot.avg_price is not None else None, "portfolio_id": slot.portfolio_id, "sequence_at": (slot.updated_at or datetime.now(tz=timezone.utc)).isoformat(), } # ---- helpers ---------------------------------------------------------------- def _opt_str(v) -> Optional[str]: if v is None or v == "": return None return str(v) def _opt_decimal(v): if v is None: return None from decimal import Decimal return Decimal(str(v)) def _opt_float(v) -> Optional[float]: if v is None: return None return float(v) def _opt_int(v) -> Optional[int]: if v is None: return None if not isinstance(v, int) or isinstance(v, bool): raise ValueError("integer required") return v def _opt_dt(v) -> Optional[datetime]: if v is None: return None if isinstance(v, datetime): return v return datetime.fromisoformat(str(v)) def _parse_date(v) -> date_cls: if isinstance(v, date_cls): return v return date_cls.fromisoformat(str(v)) def _today() -> date_cls: return datetime.now(tz=timezone.utc).date()