"""Resident scheduler: catch-up on start, then one-shot briefing jobs computed from the exchange calendar.""" from __future__ import annotations import logging import signal import threading from collections.abc import Callable from datetime import date, datetime, timedelta from typing import Any, Protocol from zoneinfo import ZoneInfo from apscheduler.events import EVENT_JOB_MISSED, JobExecutionEvent from apscheduler.schedulers import SchedulerNotRunningError from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.date import DateTrigger from apscheduler.triggers.interval import IntervalTrigger from us2kr.config import KST, Settings from us2kr.deliver import build_deliverers from us2kr.market_time import MarketClock from us2kr.pipeline import run_briefing from us2kr.store import Store log = logging.getLogger(__name__) BRIEFING_JOB_ID = "briefing" RECOVERY_JOB_ID = "briefing-recover" PREMARKET_JOB_ID = "premarket" HEARTBEAT_JOB_ID = "heartbeat" # None => APScheduler runs the one-shot briefing job whenever it becomes due, even hours late (machine # sleep, SIGSTOP, clock jump). With a finite grace APScheduler *drops* a late job instead of running it, # and a dropped one-shot job never reaches _briefing_job, whose `finally: schedule_next()` is what keeps # the chain of briefings alive. _briefing_job enforces the catch-up window itself. BRIEFING_MISFIRE_GRACE: int | None = None # A briefing whose run time is older than this and which is still recorded as "next" was either lost by # the scheduler or is running right now; the heartbeat re-arms the schedule in both cases (idempotent). HEARTBEAT_STALE_AFTER = timedelta(minutes=5) class SchedulerLike(Protocol): """The subset of APScheduler's BaseScheduler the daemon relies on (fakeable in tests).""" def add_job(self, func: Callable[..., Any], trigger: Any = None, **kwargs: Any) -> Any: ... def add_listener(self, callback: Callable[[Any], None], mask: int = ...) -> None: ... def shutdown(self, wait: bool = True) -> None: ... class Daemon: """Wires clock, store and pipeline into a scheduler. Every collaborator is injectable.""" def __init__( self, settings: Settings, *, scheduler: SchedulerLike, clock: MarketClock, store: Store, run_fn: Callable[[date], Any], now_fn: Callable[[], datetime] | None = None, deliverers_fn: Callable[[Settings], list[Any]] | None = None, premarket_fn: Callable[[], Any] | None = None, ) -> None: self.settings = settings self.scheduler = scheduler self.clock = clock self.store = store self.run_fn = run_fn self.now_fn = now_fn or (lambda: datetime.now(ZoneInfo(KST))) self.deliverers_fn = deliverers_fn or build_deliverers # Extension point: plug a real pre-market update here (PLAN.md P6). self.premarket_fn = premarket_fn self.next_run: tuple[date, datetime] | None = None self.stop_requested = False # Serialises pipeline runs (scheduled job / catch-up / heartbeat recovery) across executor threads. self._run_lock = threading.Lock() # ----------------------------------------------------------------- lifecycle def start(self) -> None: """Catch up a missed briefing, then schedule the next one plus auxiliary jobs.""" self.scheduler.add_listener(self._on_job_missed, EVENT_JOB_MISSED) self.run_catchup() if self.stop_requested: # signal arrived during catch-up, before the scheduler was running log.info("shutdown requested during catch-up - not scheduling jobs") return self.schedule_next() self.schedule_premarket() self.schedule_heartbeat() def install_signal_handlers(self) -> None: def _handler(signum: int, _frame: Any) -> None: log.info("signal %s received - shutting down", signal.Signals(signum).name) if not self.shutdown(): # The scheduler is not running yet (start-up catch-up in progress): the only way to stop # the synchronous briefing is to unwind the main thread. run_catchup and the pipeline # catch Exception only, so KeyboardInterrupt propagates up to run_daemon. raise KeyboardInterrupt try: signal.signal(signal.SIGTERM, _handler) signal.signal(signal.SIGINT, _handler) except ValueError: # not in the main thread log.warning("signal handlers not installed (not main thread)") def shutdown(self) -> bool: """Request a stop. Returns True when a running scheduler was told to shut down, False otherwise.""" self.stop_requested = True try: self.scheduler.shutdown(wait=False) except SchedulerNotRunningError: log.debug("scheduler not running - nothing to shut down") return False except Exception: # shutting down twice must not raise log.debug("scheduler shutdown raised", exc_info=True) return False return True # ------------------------------------------------------------------ catch-up def run_catchup(self) -> bool: """Run a briefing immediately if one was missed within the catch-up window. Returns True if run.""" now = self.now_fn() candidate = self.clock.catchup_candidate(now, self.settings.catchup_window_hours) if candidate is None: log.info("no catch-up candidate at %s", now.isoformat(timespec="minutes")) return False krx_date = self.clock.krx_date_for(candidate) if self.store.has_report(krx_date): log.info("catch-up: briefing for %s (KRX %s) already sent", candidate, krx_date) return False if not self._run_lock.acquire(blocking=False): log.info("catch-up: a briefing is already running - skipping") return False try: log.info("catch-up: running missed briefing for session %s", candidate) self.run_fn(candidate) except Exception: # daemon must keep going log.exception("catch-up briefing for %s failed", candidate) self._notify_error("US2KR catch-up 브리핑 실패", f"session={candidate}") finally: self._run_lock.release() return True # ---------------------------------------------------------------- scheduling def schedule_next(self) -> tuple[date, datetime] | None: """Compute the next briefing time from the calendar and (re)schedule the one-shot job.""" now = self.now_fn() try: session, run_at = self.clock.next_briefing(now) except Exception: # calendar failure: retry via heartbeat rather than die log.exception("could not compute next briefing time") self.next_run = None return None self.scheduler.add_job( self._briefing_job, DateTrigger(run_date=run_at), args=[session, run_at], id=BRIEFING_JOB_ID, name=f"briefing {session}", replace_existing=True, misfire_grace_time=BRIEFING_MISFIRE_GRACE, ) self.next_run = (session, run_at) log.info("next briefing: US session %s at %s", session, run_at.isoformat(timespec="minutes")) return self.next_run def schedule_premarket(self) -> None: if not self.settings.premarket_update_enabled: return hour, minute = _parse_hhmm(self.settings.premarket_time_kst) self.scheduler.add_job( self._premarket_job, CronTrigger(day_of_week="mon-fri", hour=hour, minute=minute, timezone=ZoneInfo(KST)), id=PREMARKET_JOB_ID, name="premarket update", replace_existing=True, misfire_grace_time=600, ) log.info("premarket update scheduled Mon-Fri %02d:%02d KST", hour, minute) def schedule_heartbeat(self) -> None: self.scheduler.add_job( self._heartbeat_job, IntervalTrigger(hours=1), id=HEARTBEAT_JOB_ID, name="heartbeat", replace_existing=True, ) # --------------------------------------------------------------------- jobs def _briefing_job(self, session: date, run_at: datetime | None = None) -> None: """Scheduled briefing. Late fires (sleep/wake) honour the catch-up window; ALWAYS reschedules.""" try: if run_at is not None: late = self.now_fn() - run_at if late >= timedelta(hours=self.settings.catchup_window_hours): log.warning( "briefing for %s scheduled at %s fired %s late - outside catch-up window, skipping", session, run_at.isoformat(timespec="minutes"), late, ) return if late > HEARTBEAT_STALE_AFTER: log.warning("briefing for %s fired %s late - running (inside window)", session, late) log.info("scheduled briefing starting for session %s", session) with self._run_lock: self.run_fn(session) except Exception as exc: # never let a job kill the scheduler log.exception("scheduled briefing for %s failed", session) self._notify_error("US2KR 브리핑 실패", f"session={session}: {type(exc).__name__}: {exc}"[:500]) finally: self.schedule_next() def _premarket_job(self) -> None: if self.premarket_fn is None: log.info("premarket update not yet implemented") return try: self.premarket_fn() except Exception: log.exception("premarket update failed") def _heartbeat_job(self) -> None: """Hourly liveness log; also the backstop that re-arms a missing or stale briefing schedule.""" now = self.now_fn() if self.next_run is None: log.warning("heartbeat: no briefing scheduled - retrying schedule") self.schedule_next() return session, run_at = self.next_run if now - run_at > HEARTBEAT_STALE_AFTER: # The one-shot job should have fired by now. Either it is running (run_catchup then finds # the lock held and does nothing) or the scheduler lost it: catch up inside the window and # re-arm. schedule_next() is idempotent, so re-arming during a run is harmless. log.warning( "heartbeat: briefing for %s at %s is stale - recovering schedule", session, run_at.isoformat(timespec="minutes"), ) self._recover_schedule() return log.info("heartbeat: next briefing for %s at %s", session, run_at.isoformat(timespec="minutes")) def _on_job_missed(self, event: JobExecutionEvent) -> None: """EVENT_JOB_MISSED listener: a dropped one-shot briefing job must be replaced. Cannot happen while BRIEFING_MISFIRE_GRACE is None, but kept as defence in depth. APScheduler removes the exhausted one-shot job *after* dispatching this event (possibly on this very thread, inside its job-processing loop), so re-adding a job with the same id here would be deleted immediately. Defer the recovery to a separately named one-shot job instead. """ if getattr(event, "job_id", None) != BRIEFING_JOB_ID: return log.warning( "briefing job scheduled at %s was missed - queuing schedule recovery", getattr(event, "scheduled_run_time", "?"), ) try: self.scheduler.add_job( self._recover_schedule, DateTrigger(run_date=self.now_fn()), id=RECOVERY_JOB_ID, name="briefing schedule recovery", replace_existing=True, misfire_grace_time=None, ) except Exception: # the heartbeat is the last backstop log.exception("could not queue schedule recovery") def _recover_schedule(self) -> None: """Catch-up policy (window + store idempotency), then reschedule. Never raises.""" try: self.run_catchup() except Exception: # clock/store failure must not kill the scheduler thread log.exception("catch-up during schedule recovery failed") finally: self.schedule_next() # ------------------------------------------------------------------ helpers def _notify_error(self, subject: str, detail: str) -> None: try: for deliverer in self.deliverers_fn(self.settings): try: deliverer.notify_error(subject, detail) except Exception: log.exception("notify_error via %s failed", getattr(deliverer, "name", deliverer)) except Exception: # building deliverers itself may fail log.exception("could not build deliverers for error notification") def _parse_hhmm(value: str) -> tuple[int, int]: try: hh, mm = value.strip().split(":") hour, minute = int(hh), int(mm) except ValueError as exc: raise ValueError(f"invalid HH:MM time: {value!r}") from exc if not (0 <= hour <= 23 and 0 <= minute <= 59): raise ValueError(f"invalid HH:MM time: {value!r}") return hour, minute def run_daemon(settings: Settings) -> None: """Build the real daemon and block until SIGTERM/SIGINT.""" scheduler = BlockingScheduler(timezone=ZoneInfo(KST)) clock = MarketClock(settings.run_offset_minutes, extra_krx_holidays=settings.krx_extra_holidays) store = Store(settings.db_path) daemon = Daemon( settings, scheduler=scheduler, clock=clock, store=store, run_fn=lambda session: run_briefing(session, settings), ) daemon.install_signal_handlers() log.info( "us2kr daemon starting (offset=%dmin, catch-up window=%dh)", settings.run_offset_minutes, settings.catchup_window_hours, ) try: daemon.start() if daemon.stop_requested: log.info("stop requested during start-up - not starting scheduler") else: scheduler.start() except (KeyboardInterrupt, SystemExit): daemon.shutdown() finally: store.close() log.info("us2kr daemon stopped")