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)
  - durable `ReconcileWarningStore`    slot_id -> reconcile-warning (Postgres; automatic,
                                        set by reconcile B/C cases, cleared by sync/verify
                                        — REQ-102, dcy-034)
"""
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."""


# ---------------------------------------------------------------- realized PnL (REQ-109)

@dataclass(frozen=True)
class RealizedPnlItem:
    """Single sell-trade realized PnL item from portfolio-hub WAC response.

    Maps to portfolio-hub RealizedPnlItem schema (REQ-040).
    Minimum required: symbol, realized_pnl, sell_qty.
    Additional fields preserved for completeness.
    """
    symbol: Optional[str]
    realized_pnl: float
    sell_qty: float
    cost_basis_per_unit: float = 0.0
    fee_amount: float = 0.0
    tax_amount: float = 0.0


@dataclass(frozen=True)
class RealizedPnlResult:
    """Aggregated WAC realized PnL result for one portfolio (REQ-109).

    Returned by get_realized_pnl port method.
    """
    items: list[RealizedPnlItem]
    cost_basis_from: Optional[str]           # ISO date string or None
    data_completeness_warning: bool          # True when portfolio-hub signals incomplete history
    total_realized_pnl: float


class PortfolioHubReadPort(Protocol):
    """Read-side contract for portfolio-hub universe (REQ-027) + realized PnL (REQ-109)."""

    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).
        """
        ...

    def get_realized_pnl(
        self,
        portfolio_id: str,
        from_dt: str,
        to_dt: str,
        *,
        symbol: Optional[str] = None,
    ) -> "RealizedPnlResult":
        """GET /v1/portfolios/{portfolio_id}/pnl/realized — WAC realized PnL (REQ-109).

        Args:
            portfolio_id: target portfolio.
            from_dt: ISO date string (YYYY-MM-DD), sell-date inclusive start.
            to_dt: ISO date string (YYYY-MM-DD), sell-date inclusive end.
            symbol: optional symbol filter.

        Returns:
            RealizedPnlResult with per-sell items, total, cost_basis_from,
            data_completeness_warning.

        Raises:
            PortfolioHubUnavailableError: on network error or non-2xx response.
        """
        ...


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))


# ---------------------------------------------------------------- reconcile-warning store (REQ-102)
# BAT-050 / dcy-034: durable reconcile-warning channel — distinct from pause (manual).
# Keyed by slot_id (TEXT UUID string). case_code: 'B' (slot↔broker qty mismatch)
# or 'C' (ledger↔broker mismatch). Survives restart (PostgreSQL-backed in production).


class ReconcileWarningStore(Protocol):
    """Durable reconcile-warning overlay — slot_id PK (REQ-102, dcy-034)."""

    def set_warned(
        self,
        slot_id: str,
        *,
        portfolio_id: str,
        case_code: str,
        actor: str = "reconcile",
        reason: str = "",
    ) -> None:
        ...

    def clear_warned(self, slot_id: str) -> None:
        ...

    def is_warned(self, slot_id: str) -> bool:
        ...

    def list_warned(self) -> "set[str]":
        ...

    def get_warning_info(self, slot_id: str) -> "Optional[dict]":
        """Return {"portfolio_id", "case_code", "actor", "reason"} or None.

        Used by verify to determine whether the warning is case_code='C'
        (verify's responsibility) vs 'B' (sync's responsibility, REQ-106 OQ-C).
        Returns None when the slot has no active warning.
        """
        ...


class InMemoryReconcileWarningStore:
    """Process-local reconcile-warning store for tests + CI (REQ-102)."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        # slot_id -> {"portfolio_id", "case_code", "actor", "reason"}
        self._warned: dict[str, dict] = {}

    def set_warned(
        self,
        slot_id: str,
        *,
        portfolio_id: str,
        case_code: str,
        actor: str = "reconcile",
        reason: str = "",
    ) -> None:
        with self._lock:
            self._warned[slot_id] = {
                "portfolio_id": portfolio_id,
                "case_code": case_code,
                "actor": actor,
                "reason": reason,
            }

    def clear_warned(self, slot_id: str) -> None:
        with self._lock:
            self._warned.pop(slot_id, None)

    def is_warned(self, slot_id: str) -> bool:
        with self._lock:
            return slot_id in self._warned

    def list_warned(self) -> "set[str]":
        with self._lock:
            return set(self._warned.keys())

    def get_warning_info(self, slot_id: str) -> "Optional[dict]":
        with self._lock:
            info = self._warned.get(slot_id)
            return dict(info) if info is not None else None


def slot_to_dict(
    s,
    *,
    snapshot_store: "Optional[SnapshotStore]" = None,
    watchdog_store=None,
    pause_store: "Optional[PauseStore]" = None,
    warn_store: "Optional[ReconcileWarningStore]" = 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/reconcile_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

    # BAT-050 / REQ-104: reconcile_warned overlay (ReconcileWarningStore) — additive field.
    # Bulk list_warned() called once per RPC (caller pre-fetches; see make_slots_rpc_handler).
    # warn_store=None → reconcile_warned=False (backward compatible, NFR-4).
    reconcile_warned = False
    if warn_store is not None:
        reconcile_warned = str(s.slot_id) in warn_store.list_warned()
    d["reconcile_warned"] = reconcile_warned

    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]