kis_bot/kis_trader/engine/strategy_eod.py

#!/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:20") -> 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, 20


def is_live_eod_now(
    enabled: bool,
    eod_hm: str,
    now: dt,
    *,
    default_hm: str = "15:20",
) -> 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", ""),
    # 정규장 15:30 마감 — 최소 10분 전 강제청산 (순차매도 여유)
    "MOMENTUM": ("MOMENTUM_EOD_ENABLED", "MOMENTUM_EOD_HM", True, "15:20", "MOMENTUM_FORCE_EOD_EXIT"),
    "TAIL": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:20", "force_eod_exit"),
    "SHORT": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:20", "force_eod_exit"),
    # 실매 scalping.py 기존 하드코딩 15:25 와 동일 (장마감청산)
    "SCALP": ("SCALP_EOD_ENABLED", "SCALP_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:20"
    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:20",
) -> 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 eod_bar_time_key(
    day_yyyymmdd: str,
    eod_hm: str,
    *,
    default_hm: str = "15:20",
) -> str:
    """당일 EOD 시각을 봉 키(YYYYMMDDHHMM)로. 백테 벽시계 EOD sell_time 용."""
    day = str(day_yyyymmdd or "").strip()[:8]
    if len(day) != 8 or not day.isdigit():
        return ""
    hh, mm = parse_eod_hm(eod_hm, default_hm)
    return "%s%02d%02d" % (day, hh, 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)