genichinandClaude Opus 4.8
63da87c9b3
`trex-core slots` 가 snapshot RPC 의 universe `assets`(portfolio)를 슬롯 상태로 오인 표시하던 갭 정정. snapshot envelope 이 슬롯 상태를 운반하지 않아 발생 (10개 슬롯인데 universe 1개·status 없음 — BAT-007 HTTP 슬롯 read 제거 후 슬롯 상태 read 표면 부재가 근인). - REQ-028 (CHG-20260608-01): snapshot envelope 에 read-only per-slot `slots[]` additive 추가(AC-4). strategy `assets[]`(universe, no-price) 불변. - REQ-032 (CHG-20260608-03): `_cmd_slots` 가 envelope `slots[]`(실제 슬롯 상태) 표시 + `--status` 를 슬롯 status 기준 client-side 필터 + watchdog 결합. `universe` 는 `assets[]` 유지. `universe --json` 도 universe 데이터로 scope. - canonical `portfolio.slot_to_dict` 단일화 — BAT-007 stub / BAT-008 envelope 두 경로가 동일 슬롯 스키마(외부명 `entry_price` 유지, dcy-012 호환). - interface-contract 에 `slots[]` 계약 + `assets[]` vs `slots[]` 구분 반영. 검증: 단위 338 passed + ruff clean + 신규 코드 데몬으로 실 PostgreSQL/RabbitMQ 운영 재현 PASS(slots=10·status·DB 일치·universe 분리·비침습). REQ-028/032 Implemented, BAT-022 release-candidate. pytest 인프라 게이트는 Human Approval Memo 수용(운영 재현이 실 경로 커버). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
444 lines
16 KiB
Python
444 lines
16 KiB
Python
"""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).
|
||
"""
|
||
...
|
||
|
||
|
||
# ---------------------------------------------------------------- selection enums
|
||
|
||
class View(str, Enum):
|
||
"""Snapshot view modes (REQ-029, dcy-005 OQ-11/OQ-12)."""
|
||
ALL = "all" # universe ∪ holding slots
|
||
BUYABLE = "buyable" # universe − occupied slots
|
||
HELD = "held" # holding slots only
|
||
|
||
|
||
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))
|
||
|
||
|
||
# ---------------------------------------------------------------- selection
|
||
|
||
@dataclass(frozen=True)
|
||
class SnapshotEnvelope:
|
||
"""REQ-028 response envelope (transport-agnostic)."""
|
||
assets: list[dict]
|
||
view: str
|
||
available_slots: int
|
||
degraded: bool
|
||
reason: Optional[str]
|
||
message: Optional[str] = None
|
||
# REQ-028 CHG-20260608-01 (BAT-022): operator-facing per-slot state list,
|
||
# additive and independent of strategy-facing `assets[]` (universe). Lets
|
||
# `trex-core slots` show the daemon's real slot table (idle/holding/pending)
|
||
# instead of universe portfolios. Always a list ([] when no slots).
|
||
slots: list[dict] = field(default_factory=list)
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"assets": list(self.assets),
|
||
"view": self.view,
|
||
"available_slots": self.available_slots,
|
||
"degraded": self.degraded,
|
||
"reason": self.reason,
|
||
"message": self.message,
|
||
"slots": list(self.slots),
|
||
}
|
||
|
||
|
||
def _asset_to_dict(item: PortfolioItem) -> dict:
|
||
"""REQ-027 AC-1 / REQ-028 AC-3 — strict no-price field shape."""
|
||
return {
|
||
"symbol": item.symbol,
|
||
"asset_id": item.portfolio_id,
|
||
"cash_balance": item.cash_balance,
|
||
"asset_quantity": item.asset_quantity,
|
||
"active_strategy_id": item.active_strategy_id,
|
||
"strategy_params": dict(item.strategy_params or {}),
|
||
}
|
||
|
||
|
||
def slot_to_dict(s) -> dict:
|
||
"""Canonical operator-facing per-slot state shape (REQ-028 AC-4, BAT-022).
|
||
|
||
Shared by the BAT-008 envelope (`build_snapshot_envelope`) and the BAT-007
|
||
stub-shape handler so `trex-core slots` sees an identical slot dict on either
|
||
path. Unlike `_asset_to_dict` (strategy universe, no-price), this is
|
||
operator-only and exposes prices. External read field name `entry_price`
|
||
is kept for contract compatibility (dcy-012 / NFR-3 — internally backed by
|
||
the renamed `limit_price`, the pending order limit cleared once holding).
|
||
"""
|
||
return {
|
||
"slot_id": str(s.slot_id),
|
||
"trading_date": s.trading_date.isoformat(),
|
||
"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,
|
||
"allocated_capital": float(s.allocated_capital) if s.allocated_capital is not None else None,
|
||
"strategy_id": s.strategy_id,
|
||
"sell_strategy_id": s.sell_strategy_id,
|
||
"portfolio_id": s.portfolio_id,
|
||
"forced_close_at": s.forced_close_at.isoformat() if s.forced_close_at else None,
|
||
}
|
||
|
||
|
||
def _sanitize(message: str) -> str:
|
||
"""Strip credentials/URLs/tokens — REQ-028 AC-2 sanitization."""
|
||
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
|
||
# Truncate to keep response payload bounded.
|
||
return out[:200]
|
||
|
||
|
||
def build_snapshot_envelope(
|
||
*,
|
||
store: SnapshotStore,
|
||
pause_store: PauseStore,
|
||
slot_repo,
|
||
available_slots: int,
|
||
view: View = View.ALL,
|
||
strategy_key: Optional[str] = None,
|
||
trading_date=None,
|
||
killswitch_store: "Optional[KillSwitchStore]" = None,
|
||
) -> SnapshotEnvelope:
|
||
"""Compose base(view) − mask(pending·warning·pause·killswitch) − strategy_key.
|
||
|
||
`slot_repo` is the SlotRepository used to read holding/pending state.
|
||
`trading_date` defaults to UTC today (matches the daemon's preallocate).
|
||
`killswitch_store` (REQ-046) adds the day-scoped kill-switch exclusion to
|
||
the mask: a `GLOBAL_KILLSWITCH` ("*") entry masks the whole universe.
|
||
"""
|
||
# Read core authoritative slot state once.
|
||
day = trading_date or datetime.now(tz=timezone.utc).date()
|
||
slots = slot_repo.list_slots(day)
|
||
holding_pids: set[str] = set()
|
||
pending_pids: set[str] = set()
|
||
holding_qty: dict[str, float] = {}
|
||
for s in slots:
|
||
if s.portfolio_id is None:
|
||
continue
|
||
if s.status.value == "holding":
|
||
holding_pids.add(s.portfolio_id)
|
||
holding_qty[s.portfolio_id] = float(s.qty or 0.0)
|
||
elif s.status.value in ("pending_buy", "pending_sell"):
|
||
pending_pids.add(s.portfolio_id)
|
||
|
||
assets = store.list_assets()
|
||
paused = pause_store.list_paused()
|
||
warned = {a.item.portfolio_id for a in assets if a.warning}
|
||
# REQ-046: day-scoped kill-switch exclusion. '*' sentinel = whole universe.
|
||
excluded = killswitch_store.list_excluded(day) if killswitch_store is not None else set()
|
||
killswitch_global = GLOBAL_KILLSWITCH in excluded
|
||
|
||
# Build the base set (view).
|
||
by_pid = {a.item.portfolio_id: a for a in assets}
|
||
base_pids: list[str]
|
||
if view is View.HELD:
|
||
base_pids = sorted(holding_pids)
|
||
elif view is View.BUYABLE:
|
||
if available_slots <= 0:
|
||
base_pids = []
|
||
else:
|
||
# universe − occupied slots (pending OR holding occupy)
|
||
occupied = holding_pids | pending_pids
|
||
base_pids = [a.item.portfolio_id for a in assets
|
||
if a.item.portfolio_id not in occupied]
|
||
else: # ALL
|
||
base_pids = list({a.item.portfolio_id for a in assets} | holding_pids)
|
||
base_pids.sort()
|
||
|
||
# Apply masks: pending OR warning OR pause OR kill-switch day-exclusion.
|
||
masked = pending_pids | warned | paused | excluded
|
||
|
||
# Filter and shape.
|
||
out: list[dict] = []
|
||
for pid in base_pids:
|
||
if killswitch_global or pid in masked:
|
||
continue
|
||
if pid in by_pid:
|
||
item = by_pid[pid].item
|
||
elif pid in holding_pids:
|
||
# holding slot for an asset NOT in the current universe (e.g. fetch
|
||
# gap) — synthesize a minimal item using slot data so the response
|
||
# still carries authoritative position info (REQ-029 slot overlay
|
||
# priority over portfolio-hub presence).
|
||
slot = next((s for s in slots if s.portfolio_id == pid), None)
|
||
item = PortfolioItem(
|
||
portfolio_id=pid,
|
||
symbol=(slot.symbol if slot else "") or "",
|
||
cash_balance=0.0,
|
||
asset_quantity=float((slot.qty if slot else 0.0) or 0.0),
|
||
active_strategy_id=(slot.strategy_id if slot else None),
|
||
strategy_params={},
|
||
)
|
||
else:
|
||
continue
|
||
if strategy_key is not None and item.active_strategy_id != strategy_key:
|
||
continue
|
||
d = _asset_to_dict(item)
|
||
# holding slot quantity wins for held assets (slot-overlay priority).
|
||
if pid in holding_qty:
|
||
d["asset_quantity"] = holding_qty[pid]
|
||
out.append(d)
|
||
|
||
# REQ-028 CHG-20260608-01 (BAT-022): operator-facing per-slot state — the
|
||
# full daemon slot table for `day`, independent of the asset view filter.
|
||
slot_states = [slot_to_dict(s) for s in slots]
|
||
|
||
reason_code, message = store.degraded_state()
|
||
return SnapshotEnvelope(
|
||
assets=out,
|
||
view=view.value,
|
||
available_slots=max(0, int(available_slots)),
|
||
degraded=reason_code is not None,
|
||
reason=reason_code.value if reason_code is not None else None,
|
||
message=message,
|
||
slots=slot_states,
|
||
)
|