Hwang
ad0ad308d1
Changes: - Updated import paths for `compute_atr_series` and `is_strategy_eod_bar` to reflect new module structure. - Removed the unused `compute_atr_series` function from `tail_engine.py`, streamlining the codebase. Impact: - These changes enhance code organization and maintainability by ensuring that only necessary components are imported and utilized, while also eliminating redundant code.
133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
kis_trader/engine/strategy_eod.py — 전략 EOD 시각 판정 (실매·백테·엔진 공통)
|
|
=====================================================================
|
|
strategies.base 와 engine 간 순환 import 방지용 — 엔진은 이 모듈만 import.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime as dt
|
|
from typing import Any, Dict, Tuple
|
|
|
|
|
|
def parse_eod_hm(raw: str, default: str = "15:25") -> Tuple[int, int]:
|
|
"""EOD 시각 문자열 → (시, 분). ``1515`` / ``15:15`` 모두 허용."""
|
|
s = str(raw or default).strip()
|
|
if not s or s.lower() == "none":
|
|
s = default
|
|
if ":" in s:
|
|
parts = s.split(":", 1)
|
|
try:
|
|
return int(parts[0]), int(parts[1])
|
|
except (ValueError, TypeError):
|
|
pass
|
|
if len(s) == 4 and s.isdigit():
|
|
return int(s[:2]), int(s[2:])
|
|
try:
|
|
hh, mm = [int(x) for x in s.split(":")]
|
|
return hh, mm
|
|
except Exception:
|
|
return 15, 25
|
|
|
|
|
|
def is_live_eod_now(
|
|
enabled: bool,
|
|
eod_hm: str,
|
|
now: dt,
|
|
*,
|
|
default_hm: str = "15:25",
|
|
) -> bool:
|
|
"""실매 EOD 당일청산 시각 도달 여부."""
|
|
if not enabled:
|
|
return False
|
|
eod_hh, eod_mm = parse_eod_hm(eod_hm, default_hm)
|
|
return (now.hour > eod_hh) or (now.hour == eod_hh and now.minute >= eod_mm)
|
|
|
|
|
|
# 전략별 EOD env 키 — 실매·백테·파라서치 공통
|
|
_STRATEGY_EOD_SPEC: Dict[str, Tuple[str, str, bool, str, str]] = {
|
|
"BREAKOUT": ("BREAKOUT_EOD_ENABLED", "BREAKOUT_EOD_HM", True, "15:15", ""),
|
|
"MOMENTUM": ("MOMENTUM_EOD_ENABLED", "MOMENTUM_EOD_HM", True, "15:25", "MOMENTUM_FORCE_EOD_EXIT"),
|
|
"TAIL": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:25", "force_eod_exit"),
|
|
"SHORT": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:25", "force_eod_exit"),
|
|
}
|
|
|
|
|
|
def _params_truthy_bool(val: Any, default: bool) -> bool:
|
|
if val is None or val == "" or val == "None":
|
|
return default
|
|
if isinstance(val, bool):
|
|
return val
|
|
return str(val).strip().lower() in ("1", "true", "t", "y", "yes", "on")
|
|
|
|
|
|
def resolve_strategy_eod_params(
|
|
params: Dict[str, Any],
|
|
strategy_id: str,
|
|
) -> Tuple[bool, str]:
|
|
"""params → (eod_enabled, eod_hm). UI ``eod_enabled``/``eod_hm`` 우선, 없으면 env 키."""
|
|
sid = str(strategy_id or "").strip().upper()
|
|
if sid == "SHORT":
|
|
sid = "TAIL"
|
|
spec = _STRATEGY_EOD_SPEC.get(sid)
|
|
if spec is None:
|
|
return False, "15:25"
|
|
en_key, hm_key, def_en, def_hm, leg_key = spec
|
|
|
|
if "eod_enabled" in params:
|
|
enabled = _params_truthy_bool(params.get("eod_enabled"), def_en)
|
|
elif en_key in params:
|
|
enabled = _params_truthy_bool(params.get(en_key), def_en)
|
|
elif leg_key and leg_key in params:
|
|
enabled = _params_truthy_bool(params.get(leg_key), def_en)
|
|
else:
|
|
enabled = def_en
|
|
|
|
raw_hm = params.get("eod_hm")
|
|
if raw_hm not in (None, "", "None"):
|
|
eod_hm = str(raw_hm).strip()
|
|
elif params.get(hm_key) not in (None, "", "None"):
|
|
eod_hm = str(params.get(hm_key)).strip()
|
|
else:
|
|
eod_hm = def_hm
|
|
return enabled, eod_hm
|
|
|
|
|
|
def is_backtest_eod_bar(
|
|
candle_time: str,
|
|
enabled: bool,
|
|
eod_hm: str,
|
|
*,
|
|
default_hm: str = "15:25",
|
|
) -> bool:
|
|
"""백테 1분봉/스캔키 — 실매 ``is_live_eod_now`` 와 동일 시각 기준."""
|
|
if not enabled:
|
|
return False
|
|
eod_hh, eod_mm = parse_eod_hm(eod_hm, default_hm)
|
|
t = str(candle_time).strip()
|
|
if len(t) < 12:
|
|
return False
|
|
try:
|
|
bar_hh = int(t[8:10])
|
|
bar_mm = int(t[10:12])
|
|
except (ValueError, TypeError):
|
|
return False
|
|
return (bar_hh > eod_hh) or (bar_hh == eod_hh and bar_mm >= eod_mm)
|
|
|
|
|
|
def is_strategy_eod_bar(
|
|
candle_time: str,
|
|
params: Dict[str, Any],
|
|
strategy_id: str,
|
|
) -> bool:
|
|
"""전략 params + 봉시각 → EOD 청산 여부 (실매와 동일 키·시각)."""
|
|
sid = str(strategy_id or "").strip().upper()
|
|
if sid == "SHORT":
|
|
sid = "TAIL"
|
|
spec = _STRATEGY_EOD_SPEC.get(sid)
|
|
if spec is None:
|
|
return False
|
|
_, _, _, def_hm, _ = spec
|
|
enabled, eod_hm = resolve_strategy_eod_params(params, sid)
|
|
return is_backtest_eod_bar(candle_time, enabled, eod_hm, default_hm=def_hm)
|