feat(signal_v2): scheduler + 5 unit tests

Time-window dispatcher: pre-market (07:00-09:00, 5min), market
(09:00-15:30, 1min), post-market (15:30-20:00, 5min), overnight skip
to next market day 07:00. Weekend + holiday detection via holidays.json.

Stub holidays.json with 13 dates. Task 6 will sync from
web-backend/stock/app/holidays.json.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-16 03:44:24 +09:00
parent 90235497ae
commit fdabc69004
3 changed files with 118 additions and 0 deletions

15
signal_v2/holidays.json Normal file
View File

@@ -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"
]

62
signal_v2/scheduler.py Normal file
View File

@@ -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

View File

@@ -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