fix: 스케줄러 단계별 타임아웃 — KIS 무응답 시 EOD 청산 누락 방지
로그 분석(2026-09-22): 15:03:46 진입 스캔 이후 데이트레이딩 스케줄러가 16:52 재배포까지 1시간 49분간 아무 로그도 남기지 않음 = KIS API 무응답(500/Server disconnected 연속)으로 await가 반환되지 않아 루프 정지 → 15:20 EOD 강제청산 미실행 → 포지션 익일 보유. - _guard(): 각 단계를 asyncio.wait_for로 감싸 초과 시 취소 후 다음 사이클 진행 - 단계별 상한: MDD 30s / watchlist 120s / ORB·EOD청산 300s / 청산·진입 180s - MDD 판단 불가 시 보수적으로 신규 진입 차단 - 테스트 3건 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -39,6 +39,23 @@ from app.services.daytrading.exit_service import check_and_exit_positions, force
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# 단계별 최대 실행 시간(초). KIS API가 응답 없이 멈추면(2026-09-22 15:03 이후 1시간 49분
|
||||||
|
# 스케줄러 정지 → 15:20 EOD 청산 누락) 루프 전체가 얼어붙으므로 단계마다 상한을 둔다.
|
||||||
|
STEP_TIMEOUT = 180
|
||||||
|
|
||||||
|
|
||||||
|
async def _guard(coro, step: str, timeout: int = STEP_TIMEOUT):
|
||||||
|
"""단계 실행을 타임아웃으로 보호. 초과 시 취소하고 None 반환(루프는 계속)."""
|
||||||
|
try:
|
||||||
|
return await asyncio.wait_for(coro, timeout=timeout)
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
logger.error(
|
||||||
|
f"[DayTrading] ⏱️ '{step}' {timeout}초 초과 — 취소하고 다음 사이클로 진행 "
|
||||||
|
f"(KIS API 무응답 가능성)"
|
||||||
|
)
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
# watchlist 자동 갱신 타임스탬프 (epoch seconds)
|
# watchlist 자동 갱신 타임스탬프 (epoch seconds)
|
||||||
# 30분(1800초)마다 stock_analysis에서 재조회
|
# 30분(1800초)마다 stock_analysis에서 재조회
|
||||||
_last_wl_sync: float = 0.0
|
_last_wl_sync: float = 0.0
|
||||||
@@ -90,7 +107,7 @@ async def daytrading_scheduler() -> None:
|
|||||||
# 개장 3분 전이면 watchlist를 미리 채워 09:00부터 ORB 수집 가능
|
# 개장 3분 전이면 watchlist를 미리 채워 09:00부터 ORB 수집 가능
|
||||||
if _until_open is not None and _until_open <= 180 and not get_watchlist():
|
if _until_open is not None and _until_open <= 180 and not get_watchlist():
|
||||||
try:
|
try:
|
||||||
count = await populate_daytrading_watchlist()
|
count = await _guard(populate_daytrading_watchlist(), "watchlist 구성", 120)
|
||||||
if count:
|
if count:
|
||||||
logger.info(f"[DayTrading] 개장 전 watchlist 선구성: {count}개")
|
logger.info(f"[DayTrading] 개장 전 watchlist 선구성: {count}개")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -104,12 +121,18 @@ async def daytrading_scheduler() -> None:
|
|||||||
continue
|
continue
|
||||||
|
|
||||||
# MDD 체크 (CLAUDE.md 5.3)
|
# MDD 체크 (CLAUDE.md 5.3)
|
||||||
short_term_ok = await mdd_manager.is_short_term_strategy_enabled()
|
short_term_ok = await _guard(
|
||||||
|
mdd_manager.is_short_term_strategy_enabled(), "MDD 체크", 30
|
||||||
|
)
|
||||||
|
if short_term_ok is None:
|
||||||
|
short_term_ok = False # 판단 불가 → 보수적으로 신규 진입 차단
|
||||||
|
|
||||||
# EOD 강제 청산 (15:20)
|
# EOD 강제 청산 (15:20)
|
||||||
if is_eod_flat_time():
|
if is_eod_flat_time():
|
||||||
if day_trading_state.positions:
|
if day_trading_state.positions:
|
||||||
exits = await force_close_all(reason="EOD 강제 청산 (15:20)")
|
exits = await _guard(
|
||||||
|
force_close_all(reason="EOD 강제 청산 (15:20)"), "EOD 강제청산", 300
|
||||||
|
)
|
||||||
if exits:
|
if exits:
|
||||||
logger.info(f"[DayTrading] EOD 청산: {len(exits)}건")
|
logger.info(f"[DayTrading] EOD 청산: {len(exits)}건")
|
||||||
await asyncio.sleep(30)
|
await asyncio.sleep(30)
|
||||||
@@ -124,7 +147,7 @@ async def daytrading_scheduler() -> None:
|
|||||||
logger.info(f"[DayTrading] 자동 watchlist: {count}개")
|
logger.info(f"[DayTrading] 자동 watchlist: {count}개")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"[DayTrading] watchlist 자동설정 실패: {e}")
|
logger.warning(f"[DayTrading] watchlist 자동설정 실패: {e}")
|
||||||
await collect_opening_ranges()
|
await _guard(collect_opening_ranges(), "ORB 수집", 300)
|
||||||
await asyncio.sleep(30)
|
await asyncio.sleep(30)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
@@ -148,13 +171,13 @@ async def daytrading_scheduler() -> None:
|
|||||||
)
|
)
|
||||||
|
|
||||||
# 청산 체크 (MDD 무관 — 기존 포지션 보호)
|
# 청산 체크 (MDD 무관 — 기존 포지션 보호)
|
||||||
exits = await check_and_exit_positions()
|
exits = await _guard(check_and_exit_positions(), "청산 체크")
|
||||||
if exits:
|
if exits:
|
||||||
logger.info(f"[DayTrading] 청산: {len(exits)}건")
|
logger.info(f"[DayTrading] 청산: {len(exits)}건")
|
||||||
|
|
||||||
# 진입 (MDD 허용 시만)
|
# 진입 (MDD 허용 시만)
|
||||||
if short_term_ok:
|
if short_term_ok:
|
||||||
entries = await scan_and_enter()
|
entries = await _guard(scan_and_enter(), "진입 스캔")
|
||||||
if entries:
|
if entries:
|
||||||
logger.info(f"[DayTrading] 진입: {len(entries)}건")
|
logger.info(f"[DayTrading] 진입: {len(entries)}건")
|
||||||
else:
|
else:
|
||||||
@@ -165,7 +188,7 @@ async def daytrading_scheduler() -> None:
|
|||||||
|
|
||||||
# 그 외 시간 — 보유 포지션 청산 체크만
|
# 그 외 시간 — 보유 포지션 청산 체크만
|
||||||
if day_trading_state.positions:
|
if day_trading_state.positions:
|
||||||
exits = await check_and_exit_positions()
|
exits = await _guard(check_and_exit_positions(), "장외 청산 체크")
|
||||||
if exits:
|
if exits:
|
||||||
logger.info(f"[DayTrading] 장외 청산: {len(exits)}건")
|
logger.info(f"[DayTrading] 장외 청산: {len(exits)}건")
|
||||||
|
|
||||||
|
|||||||
42
backend/tests/test_daytrading_scheduler_watchdog.py
Normal file
42
backend/tests/test_daytrading_scheduler_watchdog.py
Normal file
@@ -0,0 +1,42 @@
|
|||||||
|
"""스케줄러 단계 타임아웃 가드.
|
||||||
|
|
||||||
|
2026-09-22 15:03~16:52: KIS API 무응답으로 데이트레이딩 스케줄러가 1시간 49분 멈춰
|
||||||
|
15:20 EOD 강제청산을 통째로 건너뜀. 단계마다 타임아웃을 둬 루프가 계속 돌아야 한다.
|
||||||
|
"""
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.services.daytrading import scheduler as sch
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_guard_returns_value_on_success():
|
||||||
|
async def ok():
|
||||||
|
return "done"
|
||||||
|
assert await sch._guard(ok(), "step") == "done"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_guard_times_out_and_cancels_hung_step():
|
||||||
|
cancelled = asyncio.Event()
|
||||||
|
|
||||||
|
async def hang():
|
||||||
|
try:
|
||||||
|
await asyncio.sleep(60)
|
||||||
|
except asyncio.CancelledError:
|
||||||
|
cancelled.set()
|
||||||
|
raise
|
||||||
|
|
||||||
|
result = await sch._guard(hang(), "행 스텝", timeout=0.05)
|
||||||
|
assert result is None
|
||||||
|
await asyncio.sleep(0) # 취소 전파
|
||||||
|
assert cancelled.is_set(), "타임아웃 시 멈춘 단계는 취소되어야 함"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_guard_does_not_swallow_other_errors():
|
||||||
|
async def boom():
|
||||||
|
raise RuntimeError("API 오류")
|
||||||
|
with pytest.raises(RuntimeError):
|
||||||
|
await sch._guard(boom(), "step")
|
||||||
Reference in New Issue
Block a user