trex-core/trex_core/application/portfolio.py

"""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,
    }

    # 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] + "<redacted>"
            break
    return out[:200]