#!/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, {}