kis_bot/kis_trader/engine/momentum_chase_patterns.py

#!/usr/bin/env python3
"""
momentum_chase_patterns.py — 모멘텀 V3 추세추격 TRIGGER 패턴 (OR)

꼬리잡기 ``eval_tail_reversal_pattern`` 과 동일 패턴:
  - env ``MOMENTUM_PATTERN_*`` 로 ON/OFF
  - 켜진 패턴 중 **하나라도** 충족 시 진입 후보

패턴:
  - breakout        : N분 고가 돌파 + 양봉 + 거래량
  - pullback_rebreak: 스윙 고점 → 저거래 눌림(음봉) → 재돌파 + 거래량 폭발
"""
from __future__ import annotations

from typing import Any, Callable, Dict, List, Tuple

from kis_trader.utils.env import get_env_float, get_env_int


def _to_bool(v: Any, default: bool = False) -> bool:
    if v is None:
        return default
    if isinstance(v, bool):
        return v
    s = str(v).strip().lower()
    if s in ("1", "true", "t", "y", "yes", "on"):
        return True
    if s in ("0", "false", "f", "n", "no", "off", ""):
        return False
    return default


def chase_pattern_defaults() -> Dict[str, Any]:
    """V2 패턴 파라미터 기본값 — get_env_* (하드코딩 금지)."""
    return {
        "pattern_breakout": True,
        "pattern_pullback": True,
        "chase_lookback_min": get_env_int("MOMENTUM_CHASE_LOOKBACK_MIN", 10),
        "pullback_lookback_min": get_env_int("MOMENTUM_PULLBACK_LOOKBACK_MIN", 15),
        "pullback_min_pct": get_env_float("MOMENTUM_PULLBACK_MIN_PCT", 0.3),
        "pullback_max_pct": get_env_float("MOMENTUM_PULLBACK_MAX_PCT", 3.0),
        # V3 눌림 setup — 저거래 횡보·음봉 (0=OFF)
        "setup_vol_max_mult": get_env_float("MOMENTUM_SETUP_VOL_MAX_MULT", 0.8),
        "setup_bear_bars_min": get_env_int("MOMENTUM_SETUP_BEAR_BARS_MIN", 1),
    }


def _volume_spike_ok(candles: List[Dict], i: int, params: Dict[str, Any]) -> Tuple[bool, str]:
    vol_mult = float(params.get("mom_vol_mult", 1.5))
    vol_win = int(params.get("mom_vol_win", 5))
    if vol_mult <= 0:
        return True, ""
    vol = float(candles[i].get("volume", 0) or 0)
    win = max(1, min(vol_win, i))
    vols = [float(candles[k].get("volume", 0) or 0) for k in range(i - win, i)]
    if not vols or sum(vols) <= 0:
        return False, "거래량창없음"
    avg = sum(vols) / len(vols)
    if avg <= 0 or vol < avg * vol_mult:
        ratio = vol / avg if avg > 0 else 0.0
        return False, "%.2fx < %.1fx" % (ratio, vol_mult)
    return True, ""


def _pullback_setup_ok(
    candles: List[Dict],
    swing_idx: int,
    i: int,
    params: Dict[str, Any],
) -> Tuple[bool, str]:
    """
    V3 눌림 setup — 스윙 고점 이후~신호봉 직전 구간이 저거래·음봉인지 검사.
    ``setup_vol_max_mult`` 0 이하 또는 ``setup_bear_bars_min`` 0 이면 해당 축 OFF.
    """
    setup_vol_mult = float(params.get("setup_vol_max_mult", 0) or 0)
    setup_bear_min = int(params.get("setup_bear_bars_min", 0) or 0)
    if setup_vol_mult <= 0 and setup_bear_min <= 0:
        return True, ""

    pull_zone = candles[swing_idx + 1 : i]
    if not pull_zone:
        return False, "눌림봉없음"

    if setup_bear_min > 0:
        bear_cnt = sum(
            1 for x in pull_zone
            if float(x.get("close", 0) or 0) < float(x.get("open", 0) or 0)
        )
        if bear_cnt < setup_bear_min:
            return False, "음봉%d<%d" % (bear_cnt, setup_bear_min)

    if setup_vol_mult > 0:
        pull_vols = [float(x.get("volume", 0) or 0) for x in pull_zone]
        pull_avg = sum(pull_vols) / len(pull_vols) if pull_vols else 0.0
        vol_win = int(params.get("mom_vol_win", 7))
        base_start = max(0, swing_idx - vol_win)
        base_vols = [
            float(candles[k].get("volume", 0) or 0)
            for k in range(base_start, swing_idx)
        ]
        base_avg = sum(base_vols) / len(base_vols) if base_vols else pull_avg
        if base_avg > 0 and pull_avg > base_avg * setup_vol_mult:
            return False, "눌림Vol %.0f>%.0f×%.2f" % (pull_avg, base_avg, setup_vol_mult)

    return True, ""


def _detect_breakout_pattern(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """단순 돌파 — 직전 lookback 봉 고가 돌파 + 양봉 + 거래량."""
    lookback = int(params.get("chase_lookback_min", 10))
    if i < max(lookback, 2):
        return False, {}

    c = candles[i]
    cl = float(c.get("close", 0) or 0)
    op = float(c.get("open", 0) or 0)
    if cl <= 0 or cl <= op:
        return False, {}

    highs = [float(x.get("high", 0) or 0) for x in candles[i - lookback : i]]
    if not highs:
        return False, {}
    resistance = max(highs)
    if resistance <= 0 or cl <= resistance:
        return False, {}

    ok, vmsg = _volume_spike_ok(candles, i, params)
    if not ok:
        return False, {}

    gap_pct = (cl - resistance) / resistance * 100.0
    return True, {
        "pattern": "breakout",
        "resistance": resistance,
        "break_gap_pct": round(gap_pct, 3),
        "vol_ok": vmsg or "ok",
    }


def _detect_pullback_rebreak_pattern(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """눌림 후 재돌파 — 스윙 고점 대비 눌림 % 후 고점 재돌파."""
    lookback = int(params.get("pullback_lookback_min", 15))
    pb_min = float(params.get("pullback_min_pct", 0.3))
    pb_max = float(params.get("pullback_max_pct", 3.0))
    if i < lookback + 2:
        return False, {}

    start = max(0, i - lookback)
    swing_idx = start
    swing_high = float(candles[start].get("high", 0) or 0)
    for k in range(start + 1, i):
        h = float(candles[k].get("high", 0) or 0)
        if h >= swing_high:
            swing_high = h
            swing_idx = k

    if swing_high <= 0 or swing_idx >= i - 1:
        return False, {}

    pull_low = swing_high
    for k in range(swing_idx + 1, i + 1):
        lo = float(candles[k].get("low", 0) or 0)
        if lo > 0:
            pull_low = min(pull_low, lo)

    if pull_low >= swing_high:
        return False, {}

    pullback_pct = (swing_high - pull_low) / swing_high * 100.0
    if pullback_pct < pb_min or pullback_pct > pb_max:
        return False, {}

    setup_ok, setup_msg = _pullback_setup_ok(candles, swing_idx, i, params)
    if not setup_ok:
        return False, {}

    c = candles[i]
    cl = float(c.get("close", 0) or 0)
    op = float(c.get("open", 0) or 0)
    if cl <= swing_high or cl <= op:
        return False, {}

    ok, vmsg = _volume_spike_ok(candles, i, params)
    if not ok:
        return False, {}

    return True, {
        "pattern": "pullback_rebreak",
        "swing_high": swing_high,
        "pullback_pct": round(pullback_pct, 3),
        "setup_ok": setup_msg or "ok",
        "vol_ok": vmsg or "ok",
    }


def eval_momentum_chase_pattern(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
) -> Tuple[bool, str, Dict[str, Any]]:
    """
    추세추격 패턴 OR 평가.

    ``MOMENTUM_PATTERN_*`` 가 모두 OFF 이면 breakout 만 검사.
    """
    checks: List[Tuple[str, Callable]] = []
    if _to_bool(params.get("pattern_breakout"), True):
        checks.append(("breakout", _detect_breakout_pattern))
    if _to_bool(params.get("pattern_pullback"), True):
        checks.append(("pullback_rebreak", _detect_pullback_rebreak_pattern))
    if not checks:
        checks.append(("breakout", _detect_breakout_pattern))

    for name, fn in checks:
        ok, metrics = fn(candles, i, params)
        if ok:
            return True, name, metrics

    active = ",".join(n for n, _ in checks)
    return False, active, {}