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.

BAT-050 / REQ-103: reconcile now accepts warn_store for durable reconcile-warning.
"""
from __future__ import annotations

import logging
import threading
from datetime import datetime, time, timedelta, timezone
from typing import Callable, Optional

from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.broker_port import BrokerPort
from trex_core.application.portfolio import (
    DegradedReason,
    PortfolioHubReadPort,
    PortfolioHubUnavailableError,
    ReconcileWarningStore,
    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,
        slot_service=None,
        instance_key: str = "default",
        alarm_emitter: Optional[AlarmEmitter] = None,
        warn_store: Optional[ReconcileWarningStore] = 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-030): asset-centric slot reconcile after each
        # universe fetch — always on when slot_repo is wired (the reconcile is
        # what creates/retires the permanent asset slots).
        self._slot_repo = slot_repo
        self._slot_service = slot_service
        self._instance_key = instance_key
        self._last_universe_size: Optional[int] = None
        # dcy-033 / REQ-100: best-effort alarm emitter.
        self._alarm_emitter = alarm_emitter
        # BAT-050 / REQ-103: durable reconcile-warning store (Optional).
        self._warn_store = warn_store

    # ---- 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)
                # dcy-033 / REQ-100: CRITICAL alarm — portfolio hub unavailable.
                if self._alarm_emitter is not None:
                    self._alarm_emitter.publish_alarm(
                        "portfolio_hub_unavailable",
                        "portfolio-hub whoami 실패: 포트폴리오 정보 부실화",
                        fields={"reason": "whoami_failed"},
                    )
                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)
            # dcy-033 / REQ-100: CRITICAL alarm — portfolio hub unavailable.
            if self._alarm_emitter is not None:
                self._alarm_emitter.publish_alarm(
                    "portfolio_hub_unavailable",
                    "portfolio-hub get_portfolios 실패: 포트폴리오 정보 부실화",
                    fields={"daemon_id": self._daemon_id, "reason": "get_portfolios_failed"},
                )
            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:
            # BAT-050 / REQ-103: pass warn_store + slot_repo for C-case durable warning.
            reconcile_positions(
                store=self._store,
                broker=self._broker,
                slot_repo=self._slot_repo,
                instance_key=self._instance_key,
                warn_store=self._warn_store,
            )
        except Exception:  # noqa: BLE001
            _log.exception("reconciliation pass failed")
        if self._slot_repo is not None:
            try:
                self._reconcile_slots(items)
            except Exception:  # noqa: BLE001
                _log.exception("slot reconcile pass failed")
            if self._slot_service is not None:
                try:
                    from trex_core.application.reconciliation import reconcile_core_holdings
                    # BAT-050 / REQ-103: pass warn_store for B-case durable warning.
                    reconcile_core_holdings(
                        slot_repo=self._slot_repo, slot_service=self._slot_service,
                        broker=self._broker, instance_key=self._instance_key,
                        alarm_emitter=self._alarm_emitter,  # dcy-033 / REQ-100
                        warn_store=self._warn_store,  # BAT-050 / REQ-103
                    )
                except Exception:  # noqa: BLE001
                    _log.exception("core-holding divergence reconcile 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: it.symbol
            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),
            )
        result = self._slot_repo.reconcile_universe(
            self._instance_key, 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)