feat: add portfolio snapshot read layer
This commit is contained in:
@@ -16,6 +16,7 @@ from typing import Optional
|
||||
import httpx
|
||||
|
||||
from trex_core.application.broker_port import (
|
||||
BalanceItem,
|
||||
FillStatus,
|
||||
OrderAck,
|
||||
OrderRequest,
|
||||
@@ -196,6 +197,43 @@ class KISBroker:
|
||||
return FillStatus(state="partially_filled", filled_qty=ccld_qty,
|
||||
avg_price=float(avg) if avg else None, raw=row)
|
||||
|
||||
def get_balance(self) -> list[BalanceItem]:
|
||||
"""KIS 주식잔고조회 (`TTTC8434R`/`VTTC8434R`).
|
||||
|
||||
Aggregates `output1[*].pdno` + `hldg_qty`, drops zero-qty rows
|
||||
(D-2 leftover, Design Decision 7). Network/credential errors map to
|
||||
an empty list (caller treats as no broker positions).
|
||||
"""
|
||||
cano, product = self._account()
|
||||
tr_id = "VTTC8434R" if self._mock else "TTTC8434R"
|
||||
params = {
|
||||
"CANO": cano, "ACNT_PRDT_CD": product,
|
||||
"AFHR_FLPR_YN": "N", "OFL_YN": "",
|
||||
"INQR_DVSN": "02", "UNPR_DVSN": "01",
|
||||
"FUND_STTL_ICLD_YN": "N", "FNCG_AMT_AUTO_RDPT_YN": "N",
|
||||
"PRCS_DVSN": "01",
|
||||
"CTX_AREA_FK100": "", "CTX_AREA_NK100": "",
|
||||
}
|
||||
try:
|
||||
resp = httpx.get(
|
||||
f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-balance",
|
||||
headers=self._headers(tr_id), params=params, timeout=self._timeout,
|
||||
)
|
||||
payload = resp.json()
|
||||
except (httpx.HTTPError, ValueError):
|
||||
return []
|
||||
if str(payload.get("rt_cd")) != "0":
|
||||
return []
|
||||
rows = payload.get("output1") or []
|
||||
out: list[BalanceItem] = []
|
||||
for row in rows:
|
||||
symbol = str(row.get("pdno") or "").strip()
|
||||
qty = _safe_int(row.get("hldg_qty")) or 0
|
||||
if not symbol or qty <= 0:
|
||||
continue
|
||||
out.append(BalanceItem(symbol=symbol, hldg_qty=qty))
|
||||
return out
|
||||
|
||||
def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int) -> bool:
|
||||
cano, product = self._account()
|
||||
parts = broker_order_id.split(":")
|
||||
|
||||
@@ -11,6 +11,7 @@ import threading
|
||||
from typing import Optional
|
||||
|
||||
from trex_core.application.broker_port import (
|
||||
BalanceItem,
|
||||
FillStatus,
|
||||
OrderAck,
|
||||
OrderRequest,
|
||||
@@ -94,3 +95,19 @@ class PaperBroker:
|
||||
return False
|
||||
o.cancelled = True
|
||||
return True
|
||||
|
||||
def get_balance(self) -> list[BalanceItem]:
|
||||
"""Deterministic per-symbol fill aggregate (REQ-031 AC-3).
|
||||
|
||||
Aggregates buys − sells across all filled (or partially filled)
|
||||
paper orders, drops zero-qty rows (Design Decision 7).
|
||||
"""
|
||||
with self._lock:
|
||||
balances: dict[str, int] = {}
|
||||
for o in self._orders.values():
|
||||
if o.filled <= 0:
|
||||
continue
|
||||
sign = 1 if o.side == "buy" else -1
|
||||
balances[o.symbol] = balances.get(o.symbol, 0) + sign * int(o.filled)
|
||||
return [BalanceItem(symbol=sym, hldg_qty=qty)
|
||||
for sym, qty in balances.items() if qty != 0]
|
||||
|
||||
@@ -32,6 +32,7 @@ from trex_core.adapters.messaging.topology import (
|
||||
EVENTS_DLQ_QUEUE,
|
||||
EVENTS_EXCHANGE,
|
||||
INTENT_REQ_QUEUE,
|
||||
PAUSE_REQ_QUEUE,
|
||||
SNAPSHOT_REQ_QUEUE,
|
||||
STRATEGY_QUEUE_ARGS,
|
||||
)
|
||||
@@ -136,6 +137,11 @@ class RabbitMQTransport:
|
||||
durable=True,
|
||||
arguments={"x-dead-letter-exchange": EVENTS_DLQ_EXCHANGE},
|
||||
)
|
||||
ch.queue_declare(
|
||||
PAUSE_REQ_QUEUE,
|
||||
durable=True,
|
||||
arguments={"x-dead-letter-exchange": EVENTS_DLQ_EXCHANGE},
|
||||
)
|
||||
|
||||
def declare_strategy_queue(self, queue: str, binding_pattern: str) -> None:
|
||||
"""Per-strategy durable queue (REQ-023 AC-4) bound to events exchange."""
|
||||
|
||||
@@ -15,6 +15,7 @@ from __future__ import annotations
|
||||
# durable for simplicity since the queue itself is shared) ---
|
||||
SNAPSHOT_REQ_QUEUE = "trex.snapshot.req"
|
||||
INTENT_REQ_QUEUE = "trex.intent.req"
|
||||
PAUSE_REQ_QUEUE = "trex.pause.req" # BAT-008 / REQ-032 control RPC
|
||||
|
||||
# --- Events ---
|
||||
EVENTS_EXCHANGE = "trex.events" # topic, durable
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
-- BAT-008 / REQ-032: durable manual pause overlay for snapshot read responses.
|
||||
-- pause unit = portfolio_id (= asset_id). Survives daemon restart and the
|
||||
-- daily snapshot refresh; only CLI / RPC control messages mutate it.
|
||||
|
||||
CREATE TABLE IF NOT EXISTS snapshot_pause (
|
||||
portfolio_id TEXT PRIMARY KEY,
|
||||
actor TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT (now() AT TIME ZONE 'UTC'),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT (now() AT TIME ZONE 'UTC')
|
||||
);
|
||||
@@ -0,0 +1,6 @@
|
||||
"""portfolio-hub read adapters (REQ-027, dcy-005)."""
|
||||
from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter
|
||||
from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore
|
||||
from trex_core.adapters.portfolio.stub_read import StubPortfolioRead
|
||||
|
||||
__all__ = ["PortfolioHubReadAdapter", "PostgreSQLPauseStore", "StubPortfolioRead"]
|
||||
@@ -0,0 +1,107 @@
|
||||
"""portfolio-hub read HTTP adapter (REQ-027 AC-1/AC-4).
|
||||
|
||||
Mirrors portfolio-hub `list_portfolios_v1`:
|
||||
GET /v1/portfolios?managed_by_daemon_id=<id>
|
||||
GET /v1/auth/whoami (returns {"daemon_id": "..."})
|
||||
|
||||
Credentials reused from BAT-006 (TREX_PORTFOLIO_HUB_*). All failures lift
|
||||
to PortfolioHubUnavailableError so the snapshot store can set the degraded
|
||||
signal (REQ-028 AC-2) without callers seeing raw httpx exceptions.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from trex_core.application.portfolio import (
|
||||
PortfolioHubUnavailableError,
|
||||
PortfolioItem,
|
||||
)
|
||||
|
||||
|
||||
class PortfolioHubReadAdapter:
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
*,
|
||||
token: Optional[str] = None,
|
||||
timeout: float = 5.0,
|
||||
client: Optional[httpx.Client] = None,
|
||||
) -> None:
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._token = token
|
||||
self._timeout = timeout
|
||||
self._client = client
|
||||
|
||||
def _headers(self) -> dict[str, str]:
|
||||
h = {"accept": "application/json"}
|
||||
if self._token:
|
||||
h["authorization"] = f"Bearer {self._token}"
|
||||
return h
|
||||
|
||||
def _get(self, url: str, *, params: Optional[dict] = None) -> httpx.Response:
|
||||
if self._client is not None:
|
||||
return self._client.get(url, params=params, headers=self._headers(),
|
||||
timeout=self._timeout)
|
||||
return httpx.get(url, params=params, headers=self._headers(),
|
||||
timeout=self._timeout)
|
||||
|
||||
def whoami(self) -> str:
|
||||
url = f"{self._base_url}/v1/auth/whoami"
|
||||
try:
|
||||
resp = self._get(url)
|
||||
except httpx.HTTPError as exc:
|
||||
raise PortfolioHubUnavailableError(
|
||||
f"whoami network error: {exc.__class__.__name__}"
|
||||
) from exc
|
||||
if not (200 <= resp.status_code < 300):
|
||||
raise PortfolioHubUnavailableError(
|
||||
f"whoami {resp.status_code}: {resp.text[:120]}"
|
||||
)
|
||||
try:
|
||||
body = resp.json()
|
||||
except ValueError as exc:
|
||||
raise PortfolioHubUnavailableError("whoami response not JSON") from exc
|
||||
did = body.get("daemon_id") if isinstance(body, dict) else None
|
||||
if not did:
|
||||
raise PortfolioHubUnavailableError("whoami missing daemon_id")
|
||||
return str(did)
|
||||
|
||||
def get_portfolios(self, daemon_id: str) -> list[PortfolioItem]:
|
||||
url = f"{self._base_url}/v1/portfolios"
|
||||
try:
|
||||
resp = self._get(url, params={"managed_by_daemon_id": daemon_id})
|
||||
except httpx.HTTPError as exc:
|
||||
raise PortfolioHubUnavailableError(
|
||||
f"get_portfolios network error: {exc.__class__.__name__}"
|
||||
) from exc
|
||||
if not (200 <= resp.status_code < 300):
|
||||
raise PortfolioHubUnavailableError(
|
||||
f"get_portfolios {resp.status_code}: {resp.text[:120]}"
|
||||
)
|
||||
try:
|
||||
body = resp.json()
|
||||
except ValueError as exc:
|
||||
raise PortfolioHubUnavailableError("get_portfolios not JSON") from exc
|
||||
|
||||
items_raw = body.get("items", body) if isinstance(body, dict) else body
|
||||
if not isinstance(items_raw, list):
|
||||
raise PortfolioHubUnavailableError("get_portfolios shape unexpected")
|
||||
|
||||
out: list[PortfolioItem] = []
|
||||
for raw in items_raw:
|
||||
if not isinstance(raw, dict):
|
||||
continue
|
||||
pid = str(raw.get("portfolio_id") or raw.get("asset_id") or "")
|
||||
if not pid:
|
||||
continue
|
||||
out.append(PortfolioItem(
|
||||
portfolio_id=pid,
|
||||
symbol=str(raw.get("symbol") or ""),
|
||||
cash_balance=float(raw.get("cash_balance") or 0.0),
|
||||
asset_quantity=float(raw.get("asset_quantity") or 0.0),
|
||||
active_strategy_id=raw.get("active_strategy_id") or None,
|
||||
strategy_params=dict(raw.get("strategy_params") or {}),
|
||||
))
|
||||
return out
|
||||
@@ -0,0 +1,42 @@
|
||||
"""PostgreSQL-backed durable pause store (REQ-032 AC-2)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from trex_core.adapters.persistence.postgresql import _require_psycopg
|
||||
|
||||
|
||||
class PostgreSQLPauseStore:
|
||||
"""`snapshot_pause` table CRUD. portfolio_id PK is the gate."""
|
||||
|
||||
def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
|
||||
self._psycopg = _require_psycopg(psycopg_module)
|
||||
self._dsn = dsn
|
||||
|
||||
def list_paused(self) -> set[str]:
|
||||
with self._psycopg.connect(self._dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT portfolio_id FROM snapshot_pause")
|
||||
return {row[0] for row in cur.fetchall()}
|
||||
|
||||
def pause(self, portfolio_id: str, *, actor: str = "operator") -> None:
|
||||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"INSERT INTO snapshot_pause (portfolio_id, actor) "
|
||||
"VALUES (%s, %s) "
|
||||
"ON CONFLICT (portfolio_id) DO UPDATE "
|
||||
"SET actor = EXCLUDED.actor, updated_at = %s",
|
||||
(portfolio_id, actor, datetime.now(tz=timezone.utc)),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def resume(self, portfolio_id: str) -> None:
|
||||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"DELETE FROM snapshot_pause WHERE portfolio_id = %s",
|
||||
(portfolio_id,),
|
||||
)
|
||||
conn.commit()
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Deterministic stub for PortfolioHubReadPort (REQ-027 AC-2)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from trex_core.application.portfolio import PortfolioItem
|
||||
|
||||
|
||||
class StubPortfolioRead:
|
||||
"""Returns a fixed universe — no network required.
|
||||
|
||||
`whoami()` returns the configured daemon_id; tests can override the
|
||||
universe by mutating `self.items`.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
daemon_id: str = "stub-daemon",
|
||||
items: list[PortfolioItem] | None = None,
|
||||
) -> None:
|
||||
self._daemon_id = daemon_id
|
||||
self.items: list[PortfolioItem] = list(items or [
|
||||
PortfolioItem(
|
||||
portfolio_id="pf-005930",
|
||||
symbol="005930",
|
||||
cash_balance=10_000_000.0,
|
||||
asset_quantity=0.0,
|
||||
active_strategy_id="target-value-v1",
|
||||
strategy_params={"target_value": 1_000_000},
|
||||
),
|
||||
PortfolioItem(
|
||||
portfolio_id="pf-000660",
|
||||
symbol="000660",
|
||||
cash_balance=5_000_000.0,
|
||||
asset_quantity=10.0,
|
||||
active_strategy_id="trend-follow-v1",
|
||||
strategy_params={"window": 20},
|
||||
),
|
||||
])
|
||||
|
||||
def get_portfolios(self, daemon_id: str) -> list[PortfolioItem]:
|
||||
# Stub does not filter by daemon_id — every test universe is scoped.
|
||||
return list(self.items)
|
||||
|
||||
def whoami(self) -> str:
|
||||
return self._daemon_id
|
||||
@@ -29,6 +29,18 @@ class OrderAck:
|
||||
raw: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BalanceItem:
|
||||
"""One symbol in a broker account balance (REQ-031, BAT-008).
|
||||
|
||||
KIS `TTTC8434R`/`VTTC8434R` returns per-symbol `pdno` + `hldg_qty`. The
|
||||
paper broker returns a deterministic in-memory ledger keyed by symbol.
|
||||
"""
|
||||
symbol: str # KIS `pdno`
|
||||
hldg_qty: int # held quantity (integer shares; D-2 zero-qty rows are
|
||||
# treated as not-held — design Decision 7)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FillStatus:
|
||||
state: str # accepted | partially_filled | filled | rejected | cancelled | unknown
|
||||
@@ -69,6 +81,15 @@ class BrokerPort(Protocol):
|
||||
"""Best-effort cancel. Returns True if the broker accepted the cancel."""
|
||||
...
|
||||
|
||||
def get_balance(self) -> list[BalanceItem]:
|
||||
"""Per-symbol balance for the broker's single account (REQ-031).
|
||||
|
||||
Returns symbol+hldg_qty rows. Adapters MUST drop zero-qty rows so
|
||||
D-2 leftover entries (KIS keeps `hldg_qty=0` for two days after a
|
||||
full-sell) are reported as not-held (Design Decision 7).
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
# --- shared fee model (REQ-018, orbit calculate_fee pattern) -----------------
|
||||
# Conservative KR equities approximation (bps). Brokers can override, but we
|
||||
|
||||
@@ -14,6 +14,12 @@ from uuid import UUID
|
||||
|
||||
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
|
||||
from trex_core.application.messaging import RpcRequest, RpcResponse
|
||||
from trex_core.application.portfolio import (
|
||||
PauseStore,
|
||||
SnapshotStore,
|
||||
View,
|
||||
build_snapshot_envelope,
|
||||
)
|
||||
from trex_core.domain.slot import SlotRepository, SlotService
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
@@ -89,32 +95,68 @@ def make_intent_rpc_handler(intent_service: IntentService):
|
||||
|
||||
# ---- snapshot RPC handler ---------------------------------------------------
|
||||
|
||||
def make_snapshot_rpc_handler(slot_repo: SlotRepository, slot_service: SlotService):
|
||||
"""Build the RPC handler for `trex.snapshot.req`.
|
||||
def make_snapshot_rpc_handler(
|
||||
slot_repo: SlotRepository,
|
||||
slot_service: SlotService,
|
||||
*,
|
||||
snapshot_store: Optional[SnapshotStore] = None,
|
||||
pause_store: Optional[PauseStore] = None,
|
||||
default_total_slots: int = 10,
|
||||
):
|
||||
"""Build the RPC handler for `trex.snapshot.req` (REQ-028).
|
||||
|
||||
Payload shape (stub for BAT-007 — dcy-005 logical contract is not yet
|
||||
REQ-normalized; this returns a transport-shape evidence equivalent of
|
||||
`GET /v1/slots` + `GET /v1/capacity`):
|
||||
When `snapshot_store` + `pause_store` are provided (production wiring),
|
||||
returns the BAT-008 envelope:
|
||||
{
|
||||
"strategy_id": "<filter or null>",
|
||||
"view": "open" | "all" (default "all"),
|
||||
"trading_date": "YYYY-MM-DD" (optional, default today UTC)
|
||||
"assets": list of {symbol, asset_id, cash_balance,
|
||||
asset_quantity, active_strategy_id,
|
||||
strategy_params},
|
||||
"view": "all" | "buyable" | "held",
|
||||
"available_slots": int,
|
||||
"degraded": bool,
|
||||
"reason": DegradedReason value or null,
|
||||
"message": sanitized text or null
|
||||
}
|
||||
|
||||
Without those, falls back to the BAT-007 slot+capacity stub shape so
|
||||
older callers and existing tests don't break.
|
||||
"""
|
||||
|
||||
def handler(req: RpcRequest) -> RpcResponse:
|
||||
body = req.payload or {}
|
||||
view = str(body.get("view") or "all")
|
||||
strategy_filter = _opt_str(body.get("strategy_id"))
|
||||
day = _parse_date(body.get("trading_date")) if body.get("trading_date") else _today()
|
||||
|
||||
# --- BAT-008 envelope path (snapshot store wired) ---
|
||||
if snapshot_store is not None and pause_store is not None:
|
||||
requested_view = str(body.get("view") or "all").lower()
|
||||
try:
|
||||
view = View(requested_view)
|
||||
except ValueError:
|
||||
view = View.ALL
|
||||
strategy_key = _opt_str(body.get("strategy_key"))
|
||||
cap = slot_service.compute_slot_capacity(day, default_total_slots)
|
||||
envelope = build_snapshot_envelope(
|
||||
store=snapshot_store,
|
||||
pause_store=pause_store,
|
||||
slot_repo=slot_repo,
|
||||
available_slots=cap.idle_count,
|
||||
view=view,
|
||||
strategy_key=strategy_key,
|
||||
trading_date=day,
|
||||
)
|
||||
return RpcResponse(payload=envelope.to_dict())
|
||||
|
||||
# --- BAT-007 legacy stub-shape path ---
|
||||
view_raw = str(body.get("view") or "all")
|
||||
strategy_filter = _opt_str(body.get("strategy_id"))
|
||||
slots = slot_repo.list_slots(day)
|
||||
if view == "open":
|
||||
if view_raw == "open":
|
||||
slots = [s for s in slots if s.status.value in ("pending_buy", "holding", "pending_sell")]
|
||||
if strategy_filter is not None:
|
||||
slots = [s for s in slots if s.strategy_id == strategy_filter]
|
||||
cap = slot_service.compute_slot_capacity(day, max(1, len(slot_repo.list_slots(day))))
|
||||
return RpcResponse(payload={
|
||||
"view": view,
|
||||
"view": view_raw,
|
||||
"strategy_id": strategy_filter,
|
||||
"trading_date": day.isoformat(),
|
||||
"slots": [_slot_to_dict(s) for s in slots],
|
||||
@@ -130,6 +172,48 @@ def make_snapshot_rpc_handler(slot_repo: SlotRepository, slot_service: SlotServi
|
||||
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})
|
||||
|
||||
|
||||
# ---- event payload builder --------------------------------------------------
|
||||
|
||||
def slot_transition_event_payload(slot, *, event_type: str) -> dict:
|
||||
|
||||
@@ -0,0 +1,361 @@
|
||||
"""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)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 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
|
||||
|
||||
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,
|
||||
}
|
||||
|
||||
|
||||
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 _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,
|
||||
) -> SnapshotEnvelope:
|
||||
"""Compose base(view) − mask(pending·warning·pause) − strategy_key (REQ-029).
|
||||
|
||||
`slot_repo` is the SlotRepository used to read holding/pending state.
|
||||
`trading_date` defaults to UTC today (matches the daemon's preallocate).
|
||||
"""
|
||||
# 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}
|
||||
|
||||
# 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.
|
||||
masked = pending_pids | warned | paused
|
||||
|
||||
# Filter and shape.
|
||||
out: list[dict] = []
|
||||
for pid in base_pids:
|
||||
if 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)
|
||||
|
||||
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,
|
||||
)
|
||||
@@ -0,0 +1,59 @@
|
||||
"""Position reconciliation between portfolio-hub and broker (REQ-031).
|
||||
|
||||
The broker account is single (invariant B) so portfolio-hub quantities are
|
||||
aggregated to `symbol` before being compared with `BrokerPort.get_balance()`.
|
||||
Mismatches flip the asset's `warning` overlay; a subsequent successful match
|
||||
clears it automatically (in-memory only — pause is the durable channel for
|
||||
manual holds).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from trex_core.application.broker_port import BrokerPort
|
||||
from trex_core.application.portfolio import SnapshotStore
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def reconcile_positions(*, store: SnapshotStore, broker: BrokerPort) -> dict[str, str]:
|
||||
"""Run one reconciliation pass.
|
||||
|
||||
Returns a `{portfolio_id: 'warning' | 'cleared' | 'ok'}` map for telemetry.
|
||||
The map is also reflected onto `store` via `set_warning(...)`.
|
||||
"""
|
||||
broker_qty: dict[str, int] = {
|
||||
item.symbol: int(item.hldg_qty) for item in broker.get_balance()
|
||||
}
|
||||
|
||||
# Group portfolio-hub quantities by symbol so we can compare to broker's
|
||||
# per-symbol aggregate.
|
||||
by_symbol: dict[str, list[tuple[str, float]]] = {}
|
||||
assets = store.list_assets()
|
||||
for a in assets:
|
||||
sym = a.item.symbol
|
||||
by_symbol.setdefault(sym, []).append((a.item.portfolio_id, a.item.asset_quantity))
|
||||
|
||||
decisions: dict[str, str] = {}
|
||||
for symbol, entries in by_symbol.items():
|
||||
hub_total = int(sum(qty for _, qty in entries))
|
||||
broker_total = broker_qty.get(symbol, 0)
|
||||
match = hub_total == broker_total
|
||||
for pid, _ in entries:
|
||||
asset = store.get(pid)
|
||||
was_warned = bool(asset and asset.warning)
|
||||
if not match:
|
||||
if not was_warned:
|
||||
_log.warning(
|
||||
"position warning raised: portfolio_id=%s symbol=%s "
|
||||
"hub=%d broker=%d", pid, symbol, hub_total, broker_total,
|
||||
)
|
||||
store.set_warning(pid, True)
|
||||
decisions[pid] = "warning"
|
||||
else:
|
||||
if was_warned:
|
||||
store.set_warning(pid, False)
|
||||
decisions[pid] = "cleared"
|
||||
else:
|
||||
decisions[pid] = "ok"
|
||||
return decisions
|
||||
@@ -0,0 +1,103 @@
|
||||
"""Daily snapshot refresh scheduler (REQ-030, dcy-005 FR-6).
|
||||
|
||||
Wakes at the configured KST time, fetches universe from PortfolioHubReadPort,
|
||||
swaps the SnapshotStore, then runs reconciliation. Option D `*_after` writes
|
||||
(REQ-030 AC-2/AC-3) are wired through `SnapshotStore.update_cash_position`
|
||||
but disabled at the ledger layer until portfolio-hub#8 ships.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from datetime import datetime, time, timedelta, timezone
|
||||
from typing import Callable, Optional
|
||||
|
||||
from trex_core.application.broker_port import BrokerPort
|
||||
from trex_core.application.portfolio import (
|
||||
DegradedReason,
|
||||
PortfolioHubReadPort,
|
||||
PortfolioHubUnavailableError,
|
||||
SnapshotStore,
|
||||
)
|
||||
from trex_core.application.reconciliation import reconcile_positions
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
_KST = timezone(timedelta(hours=9))
|
||||
|
||||
|
||||
class SnapshotScheduler:
|
||||
"""Background-thread-friendly scheduler. Use `run_once()` for tests."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
read_port: PortfolioHubReadPort,
|
||||
store: SnapshotStore,
|
||||
broker: BrokerPort,
|
||||
schedule_kst: time = time(hour=7, minute=30),
|
||||
clock: Optional[Callable[[], datetime]] = None,
|
||||
) -> None:
|
||||
self._read = read_port
|
||||
self._store = store
|
||||
self._broker = broker
|
||||
self._schedule_kst = schedule_kst
|
||||
self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
|
||||
self._daemon_id: Optional[str] = None
|
||||
self._last_fired_date: Optional[datetime] = None
|
||||
|
||||
# ---- callable from a daemon thread loop --------------------------------
|
||||
def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None:
|
||||
# Boot fetch: do an immediate refresh so the first snapshot is alive
|
||||
# before the scheduled time, then enter the cadence.
|
||||
self.run_once()
|
||||
while not stop.wait(tick_sec):
|
||||
try:
|
||||
self._maybe_tick()
|
||||
except Exception: # noqa: BLE001 - loop must survive transient errors
|
||||
_log.exception("snapshot scheduler tick failed")
|
||||
|
||||
def _maybe_tick(self) -> None:
|
||||
now_kst = self._clock().astimezone(_KST)
|
||||
target = datetime.combine(now_kst.date(), self._schedule_kst, tzinfo=_KST)
|
||||
if now_kst < target:
|
||||
return
|
||||
last = self._last_fired_date
|
||||
if last is not None and last.date() == now_kst.date():
|
||||
return # already fired today
|
||||
self.run_once()
|
||||
|
||||
# ---- the actual refresh -----------------------------------------------
|
||||
def run_once(self) -> bool:
|
||||
"""Fetch + swap + reconcile. Returns True on success, False if degraded."""
|
||||
if self._daemon_id is None:
|
||||
try:
|
||||
self._daemon_id = self._read.whoami()
|
||||
except PortfolioHubUnavailableError as exc:
|
||||
self._store.set_degraded(
|
||||
DegradedReason.PORTFOLIO_HUB_UNAVAILABLE, str(exc)
|
||||
)
|
||||
_log.warning("whoami failed: %s", exc)
|
||||
return False
|
||||
try:
|
||||
items = self._read.get_portfolios(self._daemon_id)
|
||||
except PortfolioHubUnavailableError as exc:
|
||||
self._store.set_degraded(
|
||||
DegradedReason.PORTFOLIO_HUB_UNAVAILABLE, str(exc)
|
||||
)
|
||||
_log.warning("get_portfolios failed: %s", exc)
|
||||
return False
|
||||
except Exception as exc: # noqa: BLE001 - shape/parsing failure
|
||||
self._store.set_degraded(
|
||||
DegradedReason.SNAPSHOT_FETCH_FAILED, exc.__class__.__name__
|
||||
)
|
||||
_log.exception("snapshot fetch shape failure")
|
||||
return False
|
||||
|
||||
self._store.swap(items)
|
||||
try:
|
||||
reconcile_positions(store=self._store, broker=self._broker)
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("reconciliation pass failed")
|
||||
self._last_fired_date = self._clock()
|
||||
return True
|
||||
@@ -0,0 +1,47 @@
|
||||
"""trex-core CLI entrypoints (REQ-032).
|
||||
|
||||
Subcommands:
|
||||
pause <portfolio_id> — durably pause a portfolio (Postgres)
|
||||
resume <portfolio_id> — clear the pause
|
||||
paused — list paused portfolio_ids
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore
|
||||
from trex_core.runtime.config import DaemonConfig
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(prog="trex-core", description="trex-core admin CLI")
|
||||
sub = parser.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p_pause = sub.add_parser("pause", help="Pause a portfolio_id")
|
||||
p_pause.add_argument("portfolio_id")
|
||||
p_pause.add_argument("--actor", default="cli")
|
||||
|
||||
p_resume = sub.add_parser("resume", help="Resume a portfolio_id")
|
||||
p_resume.add_argument("portfolio_id")
|
||||
|
||||
sub.add_parser("paused", help="List paused portfolio_ids")
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
cfg = DaemonConfig.from_env()
|
||||
store = PostgreSQLPauseStore(cfg.dsn)
|
||||
|
||||
if args.cmd == "pause":
|
||||
store.pause(args.portfolio_id, actor=args.actor)
|
||||
print(f"paused {args.portfolio_id}")
|
||||
elif args.cmd == "resume":
|
||||
store.resume(args.portfolio_id)
|
||||
print(f"resumed {args.portfolio_id}")
|
||||
elif args.cmd == "paused":
|
||||
for pid in sorted(store.list_paused()):
|
||||
print(pid)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
sys.exit(main())
|
||||
@@ -48,6 +48,14 @@ class RabbitMQConfig:
|
||||
timeout: float = 5.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SnapshotScheduleConfig:
|
||||
"""Daily snapshot refresh schedule (REQ-030)."""
|
||||
|
||||
kst_time: str = "07:30"
|
||||
tick_sec: float = 60.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DaemonConfig:
|
||||
dsn: str
|
||||
@@ -62,6 +70,8 @@ class DaemonConfig:
|
||||
# REQ-022 (BAT-007): RabbitMQ transport config. Required by the new
|
||||
# sync+thread daemon — HTTP API has been removed (REQ-025).
|
||||
rabbitmq: Optional[RabbitMQConfig] = None
|
||||
# REQ-030 (BAT-008): daily snapshot refresh.
|
||||
snapshot_schedule: SnapshotScheduleConfig = SnapshotScheduleConfig()
|
||||
|
||||
@staticmethod
|
||||
def from_env() -> "DaemonConfig":
|
||||
@@ -98,6 +108,10 @@ class DaemonConfig:
|
||||
url=mq_url,
|
||||
timeout=float(os.environ.get("TREX_RABBITMQ_TIMEOUT", "5")),
|
||||
)
|
||||
snapshot_schedule = SnapshotScheduleConfig(
|
||||
kst_time=os.environ.get("TREX_SNAPSHOT_SCHEDULE_KST", "07:30").strip() or "07:30",
|
||||
tick_sec=float(os.environ.get("TREX_SNAPSHOT_TICK_SEC", "60")),
|
||||
)
|
||||
return DaemonConfig(
|
||||
dsn=dsn,
|
||||
total_slots=int(os.environ.get("TREX_TOTAL_SLOTS", "10")),
|
||||
@@ -107,4 +121,5 @@ class DaemonConfig:
|
||||
portfolio_hub=portfolio_hub,
|
||||
ledger_interval_sec=float(os.environ.get("TREX_LEDGER_INTERVAL_SEC", "1.0")),
|
||||
rabbitmq=rabbitmq,
|
||||
snapshot_schedule=snapshot_schedule,
|
||||
)
|
||||
|
||||
@@ -20,6 +20,8 @@ import signal
|
||||
import threading
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from datetime import time as time_cls
|
||||
|
||||
from trex_core.adapters.broker.paper import PaperBroker
|
||||
from trex_core.adapters.ledger import InMemoryLedger, PortfolioHubLedger
|
||||
from trex_core.adapters.messaging import (
|
||||
@@ -33,6 +35,11 @@ from trex_core.adapters.persistence.postgresql import (
|
||||
PostgreSQLLedgerOutbox,
|
||||
PostgreSQLSlotRepository,
|
||||
)
|
||||
from trex_core.adapters.portfolio import (
|
||||
PortfolioHubReadAdapter,
|
||||
PostgreSQLPauseStore,
|
||||
StubPortfolioRead,
|
||||
)
|
||||
from trex_core.application.broker_port import BrokerPort
|
||||
from trex_core.application.execution import ExecutionService
|
||||
from trex_core.application.intents import IntentService
|
||||
@@ -41,9 +48,15 @@ from trex_core.application.ledger_port import LedgerPort
|
||||
from trex_core.application.messaging import MessagingTransport
|
||||
from trex_core.application.messaging_handlers import (
|
||||
make_intent_rpc_handler,
|
||||
make_pause_rpc_handler,
|
||||
make_snapshot_rpc_handler,
|
||||
slot_transition_event_payload,
|
||||
)
|
||||
from trex_core.application.portfolio import (
|
||||
PortfolioHubReadPort,
|
||||
SnapshotStore,
|
||||
)
|
||||
from trex_core.application.snapshot_scheduler import SnapshotScheduler
|
||||
from trex_core.domain.slot import SlotService
|
||||
from trex_core.runtime.config import DaemonConfig
|
||||
|
||||
@@ -86,6 +99,23 @@ def _build_transport(config: DaemonConfig) -> MessagingTransport:
|
||||
return RabbitMQTransport(mq.url, timeout=mq.timeout)
|
||||
|
||||
|
||||
def _build_portfolio_read(config: DaemonConfig) -> PortfolioHubReadPort:
|
||||
"""portfolio-hub HTTP read adapter when configured; otherwise stub (REQ-027)."""
|
||||
pf = config.portfolio_hub
|
||||
if pf is None:
|
||||
return StubPortfolioRead()
|
||||
return PortfolioHubReadAdapter(
|
||||
pf.base_url, token=pf.token or None, timeout=pf.timeout,
|
||||
)
|
||||
|
||||
|
||||
def _parse_kst_time(spec: str) -> time_cls:
|
||||
parts = spec.split(":")
|
||||
h = int(parts[0])
|
||||
m = int(parts[1]) if len(parts) > 1 else 0
|
||||
return time_cls(hour=h, minute=m)
|
||||
|
||||
|
||||
class Daemon:
|
||||
"""Thread-orchestrating daemon (no HTTP)."""
|
||||
|
||||
@@ -104,6 +134,17 @@ class Daemon:
|
||||
self.outbox = PostgreSQLLedgerOutbox(config.dsn)
|
||||
self.ledger_writer = LedgerWriter(self.outbox, _build_ledger(config))
|
||||
|
||||
# BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
|
||||
self.snapshot_store = SnapshotStore()
|
||||
self.pause_store = PostgreSQLPauseStore(config.dsn)
|
||||
self.portfolio_read = _build_portfolio_read(config)
|
||||
self.snapshot_scheduler = SnapshotScheduler(
|
||||
read_port=self.portfolio_read,
|
||||
store=self.snapshot_store,
|
||||
broker=self.broker,
|
||||
schedule_kst=_parse_kst_time(config.snapshot_schedule.kst_time),
|
||||
)
|
||||
|
||||
# Event publisher transport — one dedicated instance for the
|
||||
# execution-loop thread (pika is not thread-safe).
|
||||
self.event_publisher: MessagingTransport = _build_transport(config)
|
||||
@@ -160,9 +201,32 @@ class Daemon:
|
||||
transport.declare_topology()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
handler = make_snapshot_rpc_handler(self.slot_repo, self.slot_service)
|
||||
handler = make_snapshot_rpc_handler(
|
||||
self.slot_repo, self.slot_service,
|
||||
snapshot_store=self.snapshot_store,
|
||||
pause_store=self.pause_store,
|
||||
default_total_slots=self.config.total_slots,
|
||||
)
|
||||
transport.consume_rpc_forever(topology.SNAPSHOT_REQ_QUEUE, handler, self.stop)
|
||||
|
||||
def _pause_rpc_loop(self) -> None:
|
||||
transport = _build_transport(self.config)
|
||||
self._transports.append(transport)
|
||||
try:
|
||||
transport.declare_topology()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
handler = make_pause_rpc_handler(self.pause_store)
|
||||
transport.consume_rpc_forever(topology.PAUSE_REQ_QUEUE, handler, self.stop)
|
||||
|
||||
def _snapshot_scheduler_loop(self) -> None:
|
||||
try:
|
||||
self.snapshot_scheduler.run_forever(
|
||||
self.stop, tick_sec=self.config.snapshot_schedule.tick_sec,
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("snapshot scheduler loop crashed")
|
||||
|
||||
# ------------------------------------------------------------ lifecycle
|
||||
def setup(self) -> None:
|
||||
apply_migrations(self.config.dsn)
|
||||
@@ -181,6 +245,8 @@ class Daemon:
|
||||
("trex-ledger", self._ledger_loop),
|
||||
("trex-intent-rpc", self._intent_rpc_loop),
|
||||
("trex-snapshot-rpc", self._snapshot_rpc_loop),
|
||||
("trex-pause-rpc", self._pause_rpc_loop),
|
||||
("trex-snapshot-scheduler", self._snapshot_scheduler_loop),
|
||||
]
|
||||
for name, target in workers:
|
||||
t = threading.Thread(target=target, name=name, daemon=True)
|
||||
|
||||
Reference in New Issue
Block a user