104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
"""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
|