feat(task-watcher): mode.py — 시간대+휴장일 판정 (SP-10)

current_mode(now, holidays): 비휴장 평일 07:00–16:30 → trading, 그 외 free.
fetch_holidays(): NAS /api/stock/holidays 조회 (실패 시 빈 set = free 안전).
TRADING_START/END env로 윈도우 조정. idle 감지 생략 (박재오 결정).
6 tests (평일 장중/장전/장후, 주말, 휴장, 경계).
Plan-B-Infra Phase 2.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-22 01:42:36 +09:00
parent 53a0657027
commit 4b60ab34c3
3 changed files with 101 additions and 0 deletions

View File

@@ -0,0 +1,57 @@
"""시간대 + 휴장일 기반 모드 판정 (idle 감지 생략 — 박재오 결정 2026-05-22).
trading: 비휴장 평일 07:0016:30 (장중) → queue:paused SET
free: 그 외 (장 전/후, 주말, 휴장) → queue:paused DEL
"""
from __future__ import annotations
import datetime as dt
import logging
import os
from typing import Set
from zoneinfo import ZoneInfo
import httpx
logger = logging.getLogger(__name__)
KST = ZoneInfo("Asia/Seoul")
STOCK_BASE_URL = os.getenv("STOCK_BASE_URL", "http://192.168.45.54:18500")
# 트레이딩 윈도우 (HH:MM, KST). .env로 조정 가능.
TRADING_START = os.getenv("TRADING_START", "07:00")
TRADING_END = os.getenv("TRADING_END", "16:30")
def _parse_hhmm(s: str) -> dt.time:
hh, mm = s.split(":")
return dt.time(int(hh), int(mm))
def current_mode(now: dt.datetime, holidays: Set[str]) -> str:
"""now(KST aware) + holidays(ISO date set) → 'trading' | 'free'."""
# 주말 (토=5, 일=6)
if now.weekday() >= 5:
return "free"
# 휴장일
if now.date().isoformat() in holidays:
return "free"
# 트레이딩 윈도우 [start, end)
start = _parse_hhmm(TRADING_START)
end = _parse_hhmm(TRADING_END)
t = now.timetz().replace(tzinfo=None)
if start <= t < end:
return "trading"
return "free"
def fetch_holidays() -> Set[str]:
"""NAS stock /api/stock/holidays 조회. 실패 시 빈 set (안전 — free로 판정)."""
try:
r = httpx.get(f"{STOCK_BASE_URL}/api/stock/holidays", timeout=10.0)
if r.status_code == 200:
return set(r.json().get("holidays", []))
logger.warning("holidays fetch returned %d", r.status_code)
except Exception:
logger.exception("holidays fetch 실패")
return set()