copilot-swe-agent[bot]andgenichin
1f728f4d7d
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
475 lines
21 KiB
Python
475 lines
21 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.
|
|
|
|
BAT-050 / REQ-103: reconcile now accepts warn_store for durable reconcile-warning.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import threading
|
|
from collections.abc import Callable
|
|
from datetime import UTC, date, datetime, time, timedelta, timezone
|
|
|
|
from trex_core.application.alarm_emitter import AlarmEmitter
|
|
from trex_core.application.balance_health import BalanceHealthTracker
|
|
from trex_core.application.broker_port import BrokerBalanceError, BrokerPort
|
|
from trex_core.application.portfolio import (
|
|
BrokerCredentialPayload,
|
|
DegradedReason,
|
|
PortfolioHubReadPort,
|
|
PortfolioHubUnavailableError,
|
|
ReconcileNotificationStore,
|
|
ReconcileWarningStore,
|
|
SnapshotStore,
|
|
TradingContext,
|
|
UnmanagedPositionStore,
|
|
)
|
|
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,
|
|
core_holdings_broker: BrokerPort | None = None,
|
|
schedule_kst: time = time(hour=7, minute=30),
|
|
clock: Callable[[], datetime] | None = None,
|
|
slot_repo=None,
|
|
slot_service=None,
|
|
instance_key: str = "default",
|
|
alarm_emitter: AlarmEmitter | None = None,
|
|
warn_store: ReconcileWarningStore | None = None,
|
|
notification_store: ReconcileNotificationStore | None = None,
|
|
unmanaged_position_store: UnmanagedPositionStore | None = None,
|
|
credential_fetch_enabled: bool = False,
|
|
kis_reinit_callback: Callable[[BrokerCredentialPayload], None] | None = None,
|
|
credential_refresh: Callable[[], object] | None = None,
|
|
) -> None:
|
|
self._read = read_port
|
|
self._store = store
|
|
self._broker = broker
|
|
self._core_holdings_broker = core_holdings_broker or broker
|
|
self._schedule_kst = schedule_kst
|
|
self._clock = clock or (lambda: datetime.now(tz=UTC))
|
|
self._daemon_id: str | None = None
|
|
self._last_fired_date: datetime | None = 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: int | None = 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
|
|
# req-0007 / ac-0024: durable notification deduplication store (Optional).
|
|
self._notification_store = notification_store
|
|
self._unmanaged_position_store = unmanaged_position_store
|
|
# dcy-036 / REQ-113/114/115: credential fetch opt-in + break-glass detection.
|
|
# credential_fetch_enabled=False (default) → env-only, no portfolio-hub fetch.
|
|
self._credential_fetch_enabled: bool = credential_fetch_enabled
|
|
# dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: optional callback to reinitialize
|
|
# KISBroker with the fetched day-start credential. None = no reinit (paper
|
|
# broker or credential_fetch_enabled=False). Set by daemon.py assembly.
|
|
self._kis_reinit_callback: Callable[[BrokerCredentialPayload], None] | None = (
|
|
kis_reinit_callback
|
|
)
|
|
# REQ-119: provider-owned refresh supersedes the legacy raw fetch hook.
|
|
self._credential_refresh = credential_refresh
|
|
# REQ-113: detect break-glass at init time (env var set by operator).
|
|
self._break_glass: bool = (
|
|
os.environ.get("TREX_KIS_BROKER_BREAK_GLASS", "").lower()
|
|
in ("1", "true", "yes")
|
|
)
|
|
# REQ-115: day-scoped credential cache. None until first successful fetch.
|
|
self._cached_credential: BrokerCredentialPayload | None = None
|
|
self._credential_date: date | None = None
|
|
# REQ-114: entry gate. True = new entries allowed (default).
|
|
# credential_fetch_enabled=False keeps this always True (no gate).
|
|
self._entry_allowed: bool = True
|
|
self._balance_health = BalanceHealthTracker(clock=self._clock)
|
|
|
|
# ---- 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:
|
|
self._maybe_retry_balance_workers()
|
|
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()
|
|
|
|
# ---- entry-allowed gate (REQ-114) ------------------------------------
|
|
def get_entry_allowed(self) -> bool:
|
|
"""Return current day-start entry gate state (REQ-114).
|
|
|
|
True = new entries allowed (default, or credential_fetch_enabled=False).
|
|
False = day-start load failed (fail-closed); BuyIntent admission blocked.
|
|
Thread-safe: _entry_allowed is set only in run_once() (scheduler thread).
|
|
"""
|
|
return self._entry_allowed
|
|
|
|
# ---- credential fetch hook (REQ-112/113/114/115) ----------------------
|
|
def _maybe_fetch_credential(self) -> None:
|
|
"""Attempt credential fetch at day-start checkpoint (REQ-114).
|
|
|
|
Called at the start of run_once() when credential_fetch_enabled=True.
|
|
Applies priority:
|
|
1. break-glass (TREX_KIS_BROKER_BREAK_GLASS=true): skip fetch, log INFO.
|
|
2. day-scoped cache hit (already fetched today): skip re-fetch (REQ-115 AC-3).
|
|
3. fetch from portfolio-hub (REQ-112): cache on success, fail-closed on failure.
|
|
|
|
Side effects:
|
|
- Sets self._entry_allowed (True on success/break-glass, False on failure).
|
|
- Sets self._cached_credential / self._credential_date on success.
|
|
- Emits CRITICAL alarm DAY_START_LOAD_FAILED on fail-closed (REQ-114).
|
|
- NFR-2: app_secret is never logged in plaintext.
|
|
"""
|
|
if self._break_glass:
|
|
_log.info(
|
|
"credential fetch: break-glass active (TREX_KIS_BROKER_BREAK_GLASS=true) "
|
|
"— using env KIS config; fetch skipped (instance=%s)",
|
|
self._instance_key,
|
|
)
|
|
# break-glass: treat as env-only, entry allowed
|
|
self._entry_allowed = True
|
|
return
|
|
|
|
today = self._clock().astimezone(_KST).date()
|
|
if self._credential_date is not None and self._credential_date == today:
|
|
# REQ-115 AC-3: day-scoped cache hit — no re-fetch
|
|
_log.debug(
|
|
"credential fetch: cache hit for %s (instance=%s)",
|
|
today, self._instance_key,
|
|
)
|
|
return
|
|
|
|
# Attempt fetch
|
|
try:
|
|
ctx: TradingContext = self._read.fetch_broker_credential()
|
|
except PortfolioHubUnavailableError as exc:
|
|
_log.error(
|
|
"DAY_START_LOAD_FAILED: portfolio-hub unavailable at day-start "
|
|
"(instance=%s): %s",
|
|
self._instance_key, exc,
|
|
)
|
|
self._entry_allowed = False
|
|
self._emit_day_start_alarm(reason="portfolio_hub_unavailable")
|
|
return
|
|
except Exception as exc: # noqa: BLE001 - unexpected failure = fail-closed
|
|
_log.error(
|
|
"DAY_START_LOAD_FAILED: unexpected error fetching broker credential "
|
|
"(instance=%s): %s",
|
|
self._instance_key, exc.__class__.__name__,
|
|
)
|
|
self._entry_allowed = False
|
|
self._emit_day_start_alarm(reason="unexpected_error")
|
|
return
|
|
|
|
# Evaluate fail-closed conditions (REQ-114)
|
|
if ctx.broker is None:
|
|
_log.error(
|
|
"DAY_START_LOAD_FAILED: broker credential absent (broker=null) "
|
|
"(instance=%s)",
|
|
self._instance_key,
|
|
)
|
|
self._entry_allowed = False
|
|
self._emit_day_start_alarm(reason="broker_null")
|
|
return
|
|
|
|
if not ctx.trading_enabled:
|
|
_log.error(
|
|
"DAY_START_LOAD_FAILED: trading_enabled=False at day-start "
|
|
"(instance=%s)",
|
|
self._instance_key,
|
|
)
|
|
self._entry_allowed = False
|
|
self._emit_day_start_alarm(reason="trading_disabled")
|
|
return
|
|
|
|
# Success: cache + allow entry
|
|
self._cached_credential = ctx.broker
|
|
self._credential_date = today
|
|
self._entry_allowed = True
|
|
# NFR-2: log account_ref only, never app_secret
|
|
_log.info(
|
|
"DAY_START credential loaded: account_ref=%s trading_enabled=%s "
|
|
"credential_version=%s (instance=%s, source=fetch)",
|
|
ctx.broker.account_ref,
|
|
ctx.trading_enabled,
|
|
ctx.broker.credential_version,
|
|
self._instance_key,
|
|
)
|
|
# dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: reinitialize KISBroker with
|
|
# fetched credential so intraday KIS API calls use the day-start creds.
|
|
# Failure does NOT flip entry_allowed — the broker runs on existing env
|
|
# credentials until next checkpoint (graceful degrade, not fail-closed).
|
|
if self._kis_reinit_callback is not None:
|
|
try:
|
|
self._kis_reinit_callback(ctx.broker)
|
|
_log.info(
|
|
"DAY_START KISBroker reinitialized with fetched credential "
|
|
"(account_ref=%s, instance=%s)",
|
|
ctx.broker.account_ref,
|
|
self._instance_key,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
_log.error(
|
|
"DAY_START KISBroker reinitialize failed: %s (instance=%s) "
|
|
"— trading continues with existing env credential",
|
|
exc.__class__.__name__,
|
|
self._instance_key,
|
|
)
|
|
|
|
def _emit_day_start_alarm(self, *, reason: str) -> None:
|
|
"""Publish CRITICAL DAY_START_LOAD_FAILED alarm (REQ-114)."""
|
|
if self._alarm_emitter is None:
|
|
return
|
|
self._alarm_emitter.publish_alarm(
|
|
"DAY_START_LOAD_FAILED",
|
|
"그날-시작 KIS 인증정보 로드 실패: 신규 진입 차단",
|
|
fields={"reason": reason, "instance_key": self._instance_key},
|
|
severity="critical",
|
|
)
|
|
|
|
# ---- the actual refresh -----------------------------------------------
|
|
def run_once(self) -> bool:
|
|
"""Fetch + swap + reconcile. Returns True on success, False if degraded."""
|
|
# dcy-036 / REQ-114: day-start credential fetch hook (opt-in).
|
|
# Must run before portfolio snapshot so entry_allowed is set before
|
|
# the first BuyIntent can arrive from the strategy daemon.
|
|
if self._credential_fetch_enabled:
|
|
if self._credential_refresh is None:
|
|
self._maybe_fetch_credential()
|
|
else:
|
|
try:
|
|
self._credential_refresh()
|
|
self._entry_allowed = True
|
|
except Exception as exc: # noqa: BLE001 - provider exposes only sanitized failure
|
|
_log.error("DAY_START_LOAD_FAILED: broker credential refresh failed: %s (instance=%s)",
|
|
exc.__class__.__name__, self._instance_key)
|
|
self._entry_allowed = False
|
|
self._emit_day_start_alarm(reason="broker_credential_unavailable")
|
|
|
|
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.
|
|
self._run_position_reconciliation_worker()
|
|
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:
|
|
# BAT-050 / REQ-103: pass warn_store for B-case durable warning.
|
|
self._run_core_holding_reconciliation_worker()
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("core-holding divergence reconcile failed")
|
|
self._last_fired_date = self._clock()
|
|
return True
|
|
|
|
def _run_position_reconciliation_worker(self) -> None:
|
|
self._run_balance_worker(
|
|
worker="position_reconciliation",
|
|
stale_after_sec=24 * 60 * 60,
|
|
run=lambda: reconcile_positions(
|
|
store=self._store,
|
|
broker=self._broker,
|
|
slot_repo=self._slot_repo,
|
|
instance_key=self._instance_key,
|
|
warn_store=self._warn_store,
|
|
alarm_emitter=self._alarm_emitter,
|
|
notification_store=self._notification_store,
|
|
),
|
|
)
|
|
|
|
def _run_core_holding_reconciliation_worker(self) -> None:
|
|
from trex_core.application.reconciliation import reconcile_core_holdings
|
|
|
|
self._run_balance_worker(
|
|
worker="core_holding_reconciliation",
|
|
stale_after_sec=24 * 60 * 60,
|
|
run=lambda: reconcile_core_holdings(
|
|
slot_repo=self._slot_repo,
|
|
slot_service=self._slot_service,
|
|
broker=self._core_holdings_broker,
|
|
instance_key=self._instance_key,
|
|
alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100
|
|
warn_store=self._warn_store, # BAT-050 / REQ-103
|
|
notification_store=self._notification_store, # req-0007 / ac-0024
|
|
unmanaged_position_store=self._unmanaged_position_store,
|
|
),
|
|
)
|
|
|
|
def _has_pending_balance_failure(self, worker: str) -> bool:
|
|
status = self._balance_health.status(worker, stale_after_sec=24 * 60 * 60)
|
|
return int(status.get("consecutive_failures", 0)) > 0
|
|
|
|
def _maybe_retry_balance_workers(self) -> None:
|
|
if self._has_pending_balance_failure("position_reconciliation"):
|
|
try:
|
|
self._run_position_reconciliation_worker()
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("reconciliation retry pass failed")
|
|
if (
|
|
self._slot_repo is not None
|
|
and self._slot_service is not None
|
|
and self._has_pending_balance_failure("core_holding_reconciliation")
|
|
):
|
|
try:
|
|
self._run_core_holding_reconciliation_worker()
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("core-holding divergence retry failed")
|
|
|
|
def _run_balance_worker(
|
|
self,
|
|
*,
|
|
worker: str,
|
|
stale_after_sec: float,
|
|
run: Callable[[], object],
|
|
) -> None:
|
|
ready, backoff_left = self._balance_health.ready(worker)
|
|
if not ready:
|
|
_log.info(
|
|
"broker balance worker deferred: worker=%s instance_key=%s backoff_left_sec=%.1f",
|
|
worker,
|
|
self._instance_key,
|
|
backoff_left,
|
|
)
|
|
return
|
|
try:
|
|
run()
|
|
except BrokerBalanceError as exc:
|
|
lifecycle, backoff = self._balance_health.record_failure(worker, code=exc.code)
|
|
if lifecycle is not None:
|
|
state = self._balance_health.status(worker, stale_after_sec=stale_after_sec)
|
|
_log.warning(
|
|
"broker balance unavailable: lifecycle=%s worker=%s instance_key=%s "
|
|
"failure_code=%s consecutive_failures=%s backoff_sec=%.1f",
|
|
lifecycle,
|
|
worker,
|
|
self._instance_key,
|
|
exc.code,
|
|
state["consecutive_failures"],
|
|
backoff,
|
|
)
|
|
raise
|
|
lifecycle = self._balance_health.record_success(worker)
|
|
if lifecycle == "recovered":
|
|
state = self._balance_health.status(worker, stale_after_sec=stale_after_sec)
|
|
_log.info(
|
|
"broker balance recovered: worker=%s instance_key=%s last_success_at=%s",
|
|
worker,
|
|
self._instance_key,
|
|
state["last_success_at"],
|
|
)
|
|
|
|
def balance_worker_health(
|
|
self,
|
|
*,
|
|
stale_after_sec: float = 24 * 60 * 60,
|
|
) -> dict[str, dict[str, object]]:
|
|
return {
|
|
"position_reconciliation": self._balance_health.status(
|
|
"position_reconciliation", stale_after_sec=stale_after_sec
|
|
),
|
|
"core_holding_reconciliation": self._balance_health.status(
|
|
"core_holding_reconciliation", stale_after_sec=stale_after_sec
|
|
),
|
|
}
|
|
|
|
# ---- 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)
|