QuantEngineByItz/src/quant_engine/lib_trading_calendar.py

#!/usr/bin/env python3
"""
lib_trading_calendar.py
───────────────────────────────────────────────────────────────────────────────
KRX 거래일 기반 데이터 신선도 판정 모듈 (결정론)

배경: 한국 주식시장은 주말·공휴일 휴장한다. 금요일 종가 데이터는 토·일 내내
변하지 않으므로, 단순 "24시간 경과" SLA는 토요일 오후만 돼도 거짓 신선도 위반을
일으킨다. 신선도는 "캡처한 종가가 여전히 최신 종가인가"로 판정해야 한다.

핵심 규칙:
  데이터가 STALE인 시점 = 캡처 시각 이후 도래하는 첫 거래일 개장(09:00 KST).
  → 금요일 15:35 캡처 → 다음 개장 = 월요일 09:00. 그 전까지 FRESH(페널티 0).

KST(UTC+9), KRX 정규장: 09:00 개장 / 15:30 마감.
공휴일: spec/krx_holidays.yaml 에서 로드. 없으면 주말만 비거래일.
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone, date, time
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[2]
HOLIDAYS_PATH = ROOT / "spec" / "krx_holidays.yaml"

KST = timezone(timedelta(hours=9))
MARKET_OPEN_HOUR = 9     # 09:00 KST 개장
MARKET_OPEN_MINUTE = 0
MARKET_CLOSE_HOUR = 15   # 15:30 KST 마감
MARKET_CLOSE_MINUTE = 30


def _load_holidays() -> set[str]:
    """공휴일 YYYY-MM-DD 문자열 집합. 파일 없으면 빈 집합(주말만 적용)."""
    if not HOLIDAYS_PATH.exists():
        return set()
    try:
        data = yaml.safe_load(HOLIDAYS_PATH.read_text(encoding="utf-8")) or {}
        hols = data.get("krx_market_holidays") or data.get("holidays") or []
        result: set[str] = set()
        for h in hols:
            d = str(h.get("date") if isinstance(h, dict) else h or "").strip()
            if d:
                result.add(d[:10])
        return result
    except Exception:
        return set()


_HOLIDAYS_CACHE: set[str] | None = None


def _holidays() -> set[str]:
    global _HOLIDAYS_CACHE
    if _HOLIDAYS_CACHE is None:
        _HOLIDAYS_CACHE = _load_holidays()
    return _HOLIDAYS_CACHE


def is_trading_day(d: date) -> bool:
    """거래일 여부 — 주말(토5/일6) 및 공휴일 제외."""
    if d.weekday() >= 5:  # 5=토, 6=일
        return False
    if d.isoformat() in _holidays():
        return False
    return True


def next_trading_day(d: date) -> date:
    """d 다음(d 제외) 첫 거래일."""
    nxt = d + timedelta(days=1)
    for _ in range(30):  # 연속 휴장 한계 방어
        if is_trading_day(nxt):
            return nxt
        nxt += timedelta(days=1)
    return nxt


def market_open_dt(d: date) -> datetime:
    """거래일 d 의 개장 시각(KST aware)."""
    return datetime.combine(d, time(MARKET_OPEN_HOUR, MARKET_OPEN_MINUTE), tzinfo=KST)


def next_market_open_after(dt_utc: datetime) -> datetime:
    """주어진 시각(UTC aware) 이후 도래하는 첫 거래일 개장 시각(KST aware).

    예: 금요일 15:35 KST → 월요일 09:00 KST (토·일 건너뜀)
        금요일 08:00 KST → 금요일 09:00 KST
        월요일 10:00 KST → 화요일 09:00 KST
    """
    dt_kst = dt_utc.astimezone(KST)
    cur_date = dt_kst.date()

    # 오늘이 거래일이고 아직 개장(09:00) 전이면 → 오늘 개장
    if is_trading_day(cur_date):
        today_open = market_open_dt(cur_date)
        if dt_kst < today_open:
            return today_open

    nd = next_trading_day(cur_date)
    return market_open_dt(nd)


def is_data_stale(captured_at_iso: str, now_utc: datetime | None = None) -> dict:
    """거래일 기반 데이터 신선도 판정.

    Returns dict: stale(bool), captured_at_kst, stale_deadline_kst, now_kst,
                  hours_until_stale(양수=FRESH, 음수=STALE), reason.
    """
    if now_utc is None:
        now_utc = datetime.now(timezone.utc)

    try:
        dt = datetime.fromisoformat(captured_at_iso.replace("Z", "+00:00"))
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
    except Exception:
        return {
            "stale": True,
            "captured_at_kst": None,
            "stale_deadline_kst": None,
            "now_kst": now_utc.astimezone(KST).isoformat(),
            "hours_until_stale": None,
            "reason": "INVALID_CAPTURED_AT",
        }

    deadline = next_market_open_after(dt)
    now_kst = now_utc.astimezone(KST)
    stale = now_kst >= deadline
    hours_until = (deadline - now_kst).total_seconds() / 3600.0

    return {
        "stale": stale,
        "captured_at_kst": dt.astimezone(KST).isoformat(),
        "stale_deadline_kst": deadline.isoformat(),
        "now_kst": now_kst.isoformat(),
        "hours_until_stale": round(hours_until, 2),
        "reason": "STALE_NEW_SESSION_OPENED" if stale else "FRESH_WITHIN_TRADING_SESSION",
    }


if __name__ == "__main__":
    import sys
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
    # 셀프 테스트 (KST 기준: 금 15:35 = UTC 06:35)
    cases = [
        ("2026-05-29T06:35:00+00:00", "2026-05-30T05:00:00+00:00", "금15:35캡처 → 토14:00 KST: FRESH 기대"),
        ("2026-05-29T06:35:00+00:00", "2026-05-31T23:30:00+00:00", "금15:35캡처 → 월08:30 KST: FRESH 기대"),
        ("2026-05-29T06:35:00+00:00", "2026-06-01T00:30:00+00:00", "금15:35캡처 → 월09:30 KST: STALE 기대"),
    ]
    for cap, now, desc in cases:
        r = is_data_stale(cap, datetime.fromisoformat(now))
        print(f"{desc}\n  stale={r['stale']} deadline={r['stale_deadline_kst']} hours_until={r['hours_until_stale']}\n")