"""Portfolio snapshot read + selection (REQ-027/028/029/030/031, dcy-005). Application-layer port for reading portfolio-hub universe + in-process snapshot store + asset-selection service. Domain stays free of HTTP and RabbitMQ deps (NFR-1). The snapshot is split between: - in-memory `SnapshotStore` universe + cash/position + warnings (process-local; warning is automatic/temporary, reset on restart) - durable `PauseStore` portfolio_id -> paused (Postgres; manual/persistent, survives restart and daily refresh — REQ-032) """ from __future__ import annotations import threading from dataclasses import dataclass, field, replace from datetime import datetime, timezone from enum import Enum from typing import Optional, Protocol # ---------------------------------------------------------------- portfolio-hub @dataclass(frozen=True) class PortfolioItem: """One asset in the daemon's portfolio universe (REQ-027 AC-1). Mirrors portfolio-hub's `list_portfolios_v1` PortfolioItem with **no price field** — pricing is the strategy module's concern, not core's. """ portfolio_id: str # == asset_id, dcy-004 OQ-1 symbol: str cash_balance: float asset_quantity: float # integer shares semantically; float for tolerance active_strategy_id: Optional[str] = None strategy_params: dict = field(default_factory=dict) # opaque, pass-through class PortfolioHubReadError(Exception): """Base class for portfolio-hub read failures.""" class PortfolioHubUnavailableError(PortfolioHubReadError): """portfolio-hub unreachable / 5xx / timeout — degraded signal upstream.""" class PortfolioHubReadPort(Protocol): """Read-side contract for portfolio-hub universe (REQ-027).""" def get_portfolios(self, daemon_id: str) -> list[PortfolioItem]: """Return all portfolios where `managed_by_daemon_id == daemon_id`.""" ... def whoami(self) -> str: """Resolve daemon_id from the credentials/token (REQ-027 AC-4). Called once at startup so `daemon_id` lookup never depends on `TREX_DAEMON_ID` env (single source of truth = auth token). """ ... class DegradedReason(str, Enum): """Machine-readable degraded reason codes (REQ-028 AC-2, dcy-005 OQ-5).""" PORTFOLIO_HUB_UNAVAILABLE = "portfolio_hub_unavailable" SNAPSHOT_FETCH_FAILED = "snapshot_fetch_failed" CASH_POSITION_STALE = "cash_position_stale" # ---------------------------------------------------------------- snapshot store @dataclass(frozen=True) class SnapshotAsset: """A portfolio item augmented with overlays (warning, sequence).""" item: PortfolioItem warning: bool = False updated_at: Optional[datetime] = None class SnapshotStore: """In-memory snapshot — universe baseline + warning overlay. Two-phase: callers `swap(new_universe)` to atomically replace the universe after a fetch, and `set_warning(portfolio_id, on)` to toggle the automatic warning overlay (REQ-031). pause overlay lives in PauseStore (REQ-032). Thread-safe; the snapshot RPC handler can read concurrently with the scheduler refresh. """ def __init__(self) -> None: self._lock = threading.Lock() self._assets: dict[str, SnapshotAsset] = {} self._degraded: Optional[DegradedReason] = None self._degraded_message: Optional[str] = None self._last_refresh_at: Optional[datetime] = None # --- universe --- def swap(self, items: list[PortfolioItem]) -> None: """Replace the entire universe atomically (immutable swap, Decision 6).""" now = datetime.now(tz=timezone.utc) with self._lock: prior_warnings = {pid for pid, a in self._assets.items() if a.warning} new_assets: dict[str, SnapshotAsset] = {} for item in items: # Carry forward warning if the same portfolio was warned before; # reconciliation will clear it on the next successful check. warn = item.portfolio_id in prior_warnings new_assets[item.portfolio_id] = SnapshotAsset( item=item, warning=warn, updated_at=now, ) self._assets = new_assets self._degraded = None self._degraded_message = None self._last_refresh_at = now def list_assets(self) -> list[SnapshotAsset]: with self._lock: return list(self._assets.values()) def get(self, portfolio_id: str) -> Optional[SnapshotAsset]: with self._lock: return self._assets.get(portfolio_id) # --- warning overlay (automatic, in-memory only) --- def set_warning(self, portfolio_id: str, on: bool) -> None: with self._lock: a = self._assets.get(portfolio_id) if a is None or a.warning is on: return self._assets[portfolio_id] = replace(a, warning=on) def warnings(self) -> set[str]: with self._lock: return {pid for pid, a in self._assets.items() if a.warning} # --- option D cash/position (REQ-030 AC-2/AC-3 — wired but no-op until # portfolio-hub#8 ships) --- def update_cash_position( self, portfolio_id: str, cash_balance: float, asset_quantity: float, ) -> None: with self._lock: a = self._assets.get(portfolio_id) if a is None: return self._assets[portfolio_id] = replace( a, item=replace(a.item, cash_balance=cash_balance, asset_quantity=asset_quantity), updated_at=datetime.now(tz=timezone.utc), ) # --- degraded signal (REQ-028 AC-2) --- def set_degraded(self, reason: DegradedReason, message: str) -> None: with self._lock: self._degraded = reason self._degraded_message = _sanitize(message) def clear_degraded(self) -> None: with self._lock: self._degraded = None self._degraded_message = None def degraded_state(self) -> tuple[Optional[DegradedReason], Optional[str]]: with self._lock: return self._degraded, self._degraded_message def last_refresh_at(self) -> Optional[datetime]: with self._lock: return self._last_refresh_at # ---------------------------------------------------------------- pause store class PauseStore(Protocol): """Durable pause overlay — `portfolio_id` PK (REQ-032).""" def list_paused(self) -> set[str]: ... def pause(self, portfolio_id: str, *, actor: str = "operator") -> None: ... def resume(self, portfolio_id: str) -> None: ... class InMemoryPauseStore: """Process-local pause store for tests + CI.""" def __init__(self, initial: Optional[set[str]] = None) -> None: self._lock = threading.Lock() self._paused: set[str] = set(initial or set()) def list_paused(self) -> set[str]: with self._lock: return set(self._paused) def pause(self, portfolio_id: str, *, actor: str = "operator") -> None: with self._lock: self._paused.add(portfolio_id) def resume(self, portfolio_id: str) -> None: with self._lock: self._paused.discard(portfolio_id) # ---------------------------------------------------------------- kill-switch # Day-scoped exclusion overlay (REQ-044/046 / BAT-012). Distinct from pause: # keyed by (portfolio_id, trading_date) so it auto-clears next trading day. GLOBAL_KILLSWITCH = "*" # sentinel: whole-universe exclusion for the day class KillSwitchStore(Protocol): """Day-scoped kill-switch exclusion overlay (REQ-044).""" def list_excluded(self, trading_date) -> set[str]: ... def exclude(self, portfolio_id: str, trading_date, *, actor: str = "kill-switch") -> None: ... def clear(self, portfolio_id: str, trading_date) -> None: ... class InMemoryKillSwitchStore: """Process-local day-scoped kill-switch store for tests + CI.""" def __init__(self) -> None: self._lock = threading.Lock() self._excluded: set[tuple[str, object]] = set() # (portfolio_id, trading_date) def list_excluded(self, trading_date) -> set[str]: with self._lock: return {pid for pid, d in self._excluded if d == trading_date} def exclude(self, portfolio_id: str, trading_date, *, actor: str = "kill-switch") -> None: with self._lock: self._excluded.add((portfolio_id, trading_date)) def clear(self, portfolio_id: str, trading_date) -> None: with self._lock: self._excluded.discard((portfolio_id, trading_date)) def slot_to_dict( s, *, snapshot_store: "Optional[SnapshotStore]" = None, watchdog_store=None, pause_store: "Optional[PauseStore]" = None, watchdog_timeout_sec: int = 120, ) -> dict: """Merged slot row for trex.slots.req (REQ-073/074). Carries slot/core state + hub asset data + watchdog + overlay flags. Field-level authority: slot/core = slot_id/status/qty/avg_price/entry_price/ forced_close_at; hub = asset_quantity/cash_balance; watchdog = watchdog{}; overlay = paused/warned. strategy fields excluded (REQ-074 AC-5). """ d = { "slot_id": str(s.slot_id), "status": s.status.value, "symbol": s.symbol, "qty": float(s.qty) if s.qty is not None else None, "avg_price": float(s.avg_price) if s.avg_price is not None else None, "entry_price": float(s.limit_price) if s.limit_price is not None else None, "portfolio_id": s.portfolio_id, "forced_close_at": s.forced_close_at.isoformat() if s.forced_close_at else None, "universe_detached": getattr(s, "retired_at", None) is not None, # dcy-028/REQ-081 (BAT-042): per-occupancy guard counter — the strategy # daemon echoes this as expected_occupancy_seq on strategy_state writes. "occupancy_seq": getattr(s, "occupancy_seq", 0), } # hub authority: asset_quantity (hub primary) + cash_balance asset_qty: Optional[float] = None cash_balance: Optional[float] = None if snapshot_store is not None and s.portfolio_id: snap = snapshot_store.get(s.portfolio_id) if snap is not None: asset_qty = float(snap.item.asset_quantity) cash_balance = float(snap.item.cash_balance) # qty (slot mirror) vs asset_quantity (hub): divergence → warned=True slot_qty = float(s.qty) if s.qty is not None else None warned = bool(slot_qty != asset_qty) if (slot_qty is not None and asset_qty is not None) else False d["asset_quantity"] = asset_qty d["cash_balance"] = cash_balance d["cash_estimated"] = False # handler overlays ledger_outbox estimate if available d["warned"] = warned # watchdog nested object (WatchdogStore) if watchdog_store is not None: sid = str(s.slot_id) last = watchdog_store.last_seen(sid) d["watchdog"] = { "expired": watchdog_store.is_expired(sid, watchdog_timeout_sec), "last_seen": last.isoformat() if last else None, } else: d["watchdog"] = None # paused overlay (PauseStore) — orthogonal to status paused = False if pause_store is not None and s.portfolio_id: paused = s.portfolio_id in pause_store.list_paused() d["paused"] = paused return d def _sanitize(message: str) -> str: """Strip credentials/URLs/tokens from degraded messages.""" if not message: return "" out = message for needle in ("authorization:", "Bearer ", "token=", "appsecret="): idx = out.lower().find(needle.lower()) if idx >= 0: out = out[:idx] + "" break return out[:200]