diff --git a/signal_v2/holidays.json b/signal_v2/holidays.json new file mode 100644 index 0000000..9053896 --- /dev/null +++ b/signal_v2/holidays.json @@ -0,0 +1,15 @@ +[ + "2026-01-01", + "2026-02-16", + "2026-02-17", + "2026-03-01", + "2026-05-05", + "2026-06-06", + "2026-08-15", + "2026-09-25", + "2026-09-26", + "2026-09-27", + "2026-10-03", + "2026-10-09", + "2026-12-25" +] diff --git a/signal_v2/scheduler.py b/signal_v2/scheduler.py new file mode 100644 index 0000000..952cabe --- /dev/null +++ b/signal_v2/scheduler.py @@ -0,0 +1,62 @@ +"""Polling scheduler — 시간대별 분기 + 휴장일 처리.""" +from __future__ import annotations +import json +import logging +from datetime import datetime, timedelta, time +from pathlib import Path +from zoneinfo import ZoneInfo + +logger = logging.getLogger(__name__) + +KST = ZoneInfo("Asia/Seoul") +_HOLIDAYS_PATH = Path(__file__).parent / "holidays.json" +_HOLIDAYS: set[str] = set(json.loads(_HOLIDAYS_PATH.read_text(encoding="utf-8"))) + +# Market windows +_PRE_OPEN = time(7, 0) +_OPEN = time(9, 0) +_CLOSE = time(15, 30) +_POST_END = time(20, 0) + + +def _is_market_day(now: datetime) -> bool: + """평일 + 휴장일 아닌 날.""" + if now.weekday() >= 5: # Sat/Sun + return False + return now.strftime("%Y-%m-%d") not in _HOLIDAYS + + +def _is_polling_window(now: datetime) -> bool: + """현재 시각이 폴링 윈도우 (07:00-20:00) 안인가.""" + return _PRE_OPEN <= now.time() < _POST_END + + +def _next_interval(now: datetime) -> float: + """다음 폴링까지 sleep 초수.""" + if not _is_market_day(now): + return _seconds_until_next_market_open(now) + + t = now.time() + if _PRE_OPEN <= t < _OPEN: + return 300.0 + elif _OPEN <= t < _CLOSE: + return 60.0 + elif _CLOSE <= t < _POST_END: + return 300.0 + else: + return _seconds_until_next_market_open(now) + + +def _seconds_until_next_market_open(now: datetime) -> float: + """다음 영업일의 07:00 KST 까지 초수.""" + candidate = now.replace(hour=7, minute=0, second=0, microsecond=0) + if candidate <= now: + candidate += timedelta(days=1) + + for _ in range(14): # safety bound (max 2 weeks of holidays) + if _is_market_day(candidate): + return (candidate - now).total_seconds() + candidate += timedelta(days=1) + + logger.warning("could not find next market day within 14 days; using +1 day") + return 86400.0 diff --git a/signal_v2/tests/test_scheduler.py b/signal_v2/tests/test_scheduler.py new file mode 100644 index 0000000..e04e3dd --- /dev/null +++ b/signal_v2/tests/test_scheduler.py @@ -0,0 +1,41 @@ +"""Tests for scheduler interval logic.""" +from datetime import datetime + +import pytest + +from signal_v2.scheduler import _next_interval, _is_market_day, KST + + +def _kst(year, month, day, hour, minute=0): + return datetime(year, month, day, hour, minute, tzinfo=KST) + + +def test_next_interval_pre_market_5min(): + now = _kst(2026, 5, 18, 8, 30) # Monday 08:30 + assert _next_interval(now) == 300 + + +def test_next_interval_market_open_1min(): + now = _kst(2026, 5, 18, 10, 0) # Monday 10:00 + assert _next_interval(now) == 60 + + +def test_next_interval_post_market_5min(): + now = _kst(2026, 5, 18, 17, 0) # Monday 17:00 + assert _next_interval(now) == 300 + + +def test_next_interval_overnight_skip_to_next_morning(): + now = _kst(2026, 5, 18, 22, 0) # Monday 22:00 + interval = _next_interval(now) + # Next polling: Tuesday 07:00 (9 hours away) + assert 9 * 3600 - 60 < interval < 9 * 3600 + 60 + + +def test_next_interval_holiday_skip(): + # 2026-05-05 어린이날 (Tuesday holiday) + now = _kst(2026, 5, 5, 10, 0) + assert _is_market_day(now) is False + interval = _next_interval(now) + # Next: 2026-05-06 (Wed) 07:00, ~21h away + assert 20 * 3600 < interval < 22 * 3600