trex-core/trex_core/application/snapshot_scheduler.py

"""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,
        slot_repo=None,
        instance_key: str = "default",
        reconcile_enabled: bool = False,
        trading_date_fn: Optional[Callable[[], object]] = 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
        # dcy-020 / REQ-058 (BAT-028): asset-centric slot reconcile after each
        # universe fetch (only when enabled = asset-centric mode).
        self._slot_repo = slot_repo
        self._instance_key = instance_key
        self._reconcile_enabled = reconcile_enabled
        self._trading_date_fn = trading_date_fn
        self._last_universe_size: Optional[int] = 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")
        if self._reconcile_enabled and self._slot_repo is not None:
            try:
                self._reconcile_slots(items)
            except Exception:  # noqa: BLE001
                _log.exception("slot reconcile pass failed")
        self._last_fired_date = self._clock()
        return True

    # ---- asset-centric slot reconcile (dcy-020/REQ-058, BAT-028) ----------
    def _reconcile_slots(self, items: list) -> None:
        universe = {it.portfolio_id for it in items if getattr(it, "portfolio_id", None)}
        if not universe:
            _log.warning("slot reconcile skipped: empty universe (degrade guard)")
            return
        # mass-disappearance guard (D3): retire 보류 on N→0 (above) or >=50% shrink
        allow_retire = True
        prior = self._last_universe_size
        if prior is not None and len(universe) < 0.5 * prior:
            allow_retire = False
            _log.warning(
                "slot reconcile: universe shrank %d->%d (>=50%%) — retire held",
                prior, len(universe),
            )
        if self._trading_date_fn is not None:
            trading_date = self._trading_date_fn()
        else:
            trading_date = self._clock().date()
        result = self._slot_repo.reconcile_universe(
            self._instance_key, trading_date, universe, allow_retire=allow_retire,
        )
        self._last_universe_size = len(universe)
        if any(result.values()):
            _log.info("slot reconcile: %s (instance=%s)", result, self._instance_key)