feat(task-watcher): watcher.py — 30초 loop + queue:paused 토글 (SP-10)
trading → SET queue:paused 1 EX 600 / free → DEL. holidays 1시간마다 refresh. PAUSED_TTL 600s (watcher 죽어도 자동 해제 — 안전). mode 전환 시에만 로그. Plan-B-Infra Phase 2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
59
services/task-watcher/watcher.py
Normal file
59
services/task-watcher/watcher.py
Normal file
@@ -0,0 +1,59 @@
|
||||
"""30초마다 current_mode 판정 → queue:paused 토글.
|
||||
|
||||
trading → SET queue:paused 1 EX 600 (10분 TTL — watcher 죽어도 자동 해제)
|
||||
free → DEL queue:paused
|
||||
holidays는 1시간마다 refresh (매 loop fetch 부하 회피).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime as dt
|
||||
import logging
|
||||
import os
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import redis.asyncio as aioredis
|
||||
|
||||
from mode import current_mode, fetch_holidays, KST
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
REDIS_URL = os.getenv("REDIS_URL", "redis://192.168.45.54:6379")
|
||||
PAUSED_KEY = "queue:paused"
|
||||
LOOP_INTERVAL = 30 # 초
|
||||
HOLIDAYS_REFRESH = 3600 # 1시간
|
||||
PAUSED_TTL = 600 # 10분 (watcher 죽어도 자동 해제)
|
||||
|
||||
|
||||
async def watcher_loop():
|
||||
redis = aioredis.from_url(REDIS_URL, decode_responses=False)
|
||||
holidays = fetch_holidays()
|
||||
last_holiday_refresh = dt.datetime.now(KST)
|
||||
last_mode = None
|
||||
logger.info("task-watcher started (trading window 토글)")
|
||||
|
||||
while True:
|
||||
try:
|
||||
now = dt.datetime.now(KST)
|
||||
# holidays 주기적 refresh
|
||||
if (now - last_holiday_refresh).total_seconds() >= HOLIDAYS_REFRESH:
|
||||
holidays = fetch_holidays()
|
||||
last_holiday_refresh = now
|
||||
|
||||
mode = current_mode(now, holidays)
|
||||
if mode == "trading":
|
||||
await redis.set(PAUSED_KEY, b"1", ex=PAUSED_TTL)
|
||||
else:
|
||||
await redis.delete(PAUSED_KEY)
|
||||
|
||||
if mode != last_mode:
|
||||
logger.info("mode 전환: %s → %s (paused=%s)", last_mode, mode, mode == "trading")
|
||||
last_mode = mode
|
||||
|
||||
await asyncio.sleep(LOOP_INTERVAL)
|
||||
except asyncio.CancelledError:
|
||||
logger.info("watcher_loop cancelled")
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("watcher_loop iteration 실패, 30초 후 재시도")
|
||||
await asyncio.sleep(LOOP_INTERVAL)
|
||||
Reference in New Issue
Block a user