kis_bot/kis_trader/engine/momentum_hts_logic.py

#!/usr/bin/env python3
"""
momentum_hts_logic.py — HTS momentum 조건식(E∧F∧H∧I) 정합 TRIGGER·청산
====================================================================

[역할 분담 — SCAN vs TRIGGER vs 청산]
- SCAN (키움 WS ``momentum`` 조건검색): E∧F∧H∧I 일봉 필터 → target_candidates_history
  · E: 종가 > 전일 시가
  · F: 2천~20만원
  · H: 5일 평균거래량 105% 이상
  · I: 거래량증감률 상위 400
- TRIGGER (본 모듈): SCAN 후 **진입 타이밍**만 검사
  · ``MOMENTUM_SKIP_HTS_SCAN_DUPES=true`` (kiwoom 기본): E·양봉·분봉거래량·RSI 중복 생략
  · false: E 유지·양봉·거래량 펄스(선택) — 당일 모멘텀 살아있음 확인
- 청산 (본 모듈): 래칫·어깨·트레일·손절·시간컷 (A안 돌파 추격)
"""
from __future__ import annotations

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

from kis_trader.engine.momentum_env_keys import momentum_env_bool, momentum_env_float, momentum_env_int
from kis_trader.engine.orderbook_filter import orderbook_reject_for_entry
from kis_trader.engine.program_filter import program_reject_for_entry
from kis_trader.engine.whipsaw_filter import whipsaw_reject_for_signal
from kis_trader.utils.env import get_env_from_db
from kis_trader.utils.trade_time import parse_trade_datetime as _t2dt


def _to_bool(v: Any, default: bool = True) -> 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 resolve_prev_trading_day_open(
    candles: List[Dict],
    i: int,
    day: str,
) -> Optional[float]:
    """
    전일(직전 거래일) 시가 — HTS momentum E 조건 ``close > prev_open`` 확인용.

    1분봉에서 전일 첫 봉 open = 일봉 시가.
    """
    prev_day: Optional[str] = None
    prev_open: Optional[float] = None
    for j in range(i - 1, -1, -1):
        d = str(candles[j].get("candle_time", ""))[:8]
        if not d or d >= day:
            continue
        if prev_day is None:
            prev_day = d
        if d != prev_day:
            break
        op = float(candles[j].get("open", 0) or 0)
        if op > 0:
            prev_open = op
    return prev_open


def _volume_pulse_ok(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
) -> Tuple[bool, str]:
    """H/I 정신: 당일 거래량 펄스 — 직전 N봉 평균 × 배수 이상."""
    vol_mult = float(params.get("mom_vol_mult", 1.0))
    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 _parse_ratchet_tiers(params: Dict[str, Any]) -> List[Tuple[float, float]]:
    raw = params.get("ratchet_tiers")
    if raw is None:
        raw = get_env_from_db("MOMENTUM_RATCHET_TIERS", "")
    if isinstance(raw, (list, tuple)):
        pairs = list(raw)
    else:
        s = str(raw or "").strip()
        if not s:
            return []
        pairs = []
        for chunk in s.split(","):
            chunk = chunk.strip()
            if not chunk or ":" not in chunk:
                continue
            g, c = chunk.split(":", 1)
            pairs.append((g, c))
    tiers: List[Tuple[float, float]] = []
    for g, c in pairs:
        try:
            gain = abs(float(g)) / 100.0
            cut = abs(float(c)) / 100.0
        except (TypeError, ValueError):
            continue
        if gain <= 0 or cut <= 0:
            continue
        tiers.append((gain, cut))
    tiers.sort(key=lambda x: x[0])
    return tiers


def _shoulder_ratios(params: Dict[str, Any]) -> Tuple[float, float]:
    """어깨 발동·컷 비율 — params(비율) 또는 legacy 퍼센트."""
    smh = float(params.get("shoulder_min_high", 0.005))
    sc = float(params.get("shoulder_cut_pct", 0.003))
    return max(0.0, smh), max(0.0, sc)


def _minutes_held(position: Dict[str, Any], candle: Dict[str, Any]) -> Optional[int]:
    try:
        e = _t2dt(position.get("entry_time") or position.get("buy_time", ""))
        n = _t2dt(candle.get("candle_time", ""))
        return max(0, int((n - e).total_seconds() / 60))
    except Exception:
        return None


def resolve_effective_tp_pct(tp_pct: float, tp_max_pct: float) -> float:
    tp = abs(float(tp_pct))
    cap = abs(float(tp_max_pct))
    if cap > 0:
        return min(tp, cap)
    return tp


def effective_tp_pct_from_params(params: Dict[str, Any]) -> float:
    return resolve_effective_tp_pct(
        params.get("tp_pct", 0.05),
        params.get("tp_max_pct", 0.08),
    )


def resolve_momentum_skip_hts_scan_dupes(r: Optional[Dict[str, Any]] = None) -> bool:
    """
    HTS momentum SCAN(kiwoom_condition) 사용 시 TRIGGER 중복 필터 생략 여부.

    - ``MOMENTUM_SKIP_HTS_SCAN_DUPES`` 명시 → 그대로
    - 미설정 → ``MOMENTUM_UNIVERSE_SOURCE`` 가 condition/kiwoom_condition 이면 True
    """
    if r is None:
        try:
            from kis_trader.utils.env import get_strategy_env_dict
            r = get_strategy_env_dict("MOMENTUM") or {}
        except Exception:
            r = {}
    raw = r.get("MOMENTUM_SKIP_HTS_SCAN_DUPES")
    if raw is not None and str(raw).strip() != "":
        return _to_bool(raw, True)
    # 엔진 defaults 에 이미 해석된 bool 이 있으면 universe fallback 금지
    if "skip_hts_scan_dupes" in r and r.get("skip_hts_scan_dupes") is not None:
        return _to_bool(r.get("skip_hts_scan_dupes"), False)
    src = str(r.get("MOMENTUM_UNIVERSE_SOURCE") or "condition").strip().lower()
    return src in ("kiwoom_condition", "condition")


def hts_trigger_defaults_from_row(r: Dict[str, Any]) -> Dict[str, Any]:
    """env 행에서 HTS TRIGGER 전용 플래그."""
    return {
        "trigger_e_confirm": momentum_env_bool(r, "MOMENTUM_TRIGGER_E_CONFIRM", True),
        "trigger_require_bull_bar": momentum_env_bool(r, "MOMENTUM_TRIGGER_REQUIRE_BULL_BAR", True),
        "use_vol_trigger": momentum_env_bool(r, "MOMENTUM_USE_VOL_TRIGGER", True),
        "use_rsi_filter": momentum_env_bool(r, "MOMENTUM_USE_RSI_FILTER", False),
    }


def eval_momentum_hts_buy_at_index(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    state: Dict[str, Any],
    *,
    compute_rsi_series_fn=None,
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """
    HTS momentum 조건식 정합 TRIGGER.

    SCAN(E∧F∧H∧I)은 이미 통과한 종목만 후보 — 여기서는 진입 타이밍만 검사.
    """
    if i < 1 or i >= len(candles):
        return ("탈락-봉부족", "인덱스 부적절 (i=%d)" % i, None)

    time_start_hm = int(params.get("time_start_hm", 900))
    time_end_hm = int(params.get("mom_time_end_hm", params.get("time_end_hm", 1530)))
    cooldown_min = float(params.get("cooldown_min", 10))
    max_daily = int(params.get("max_daily", 5))
    min_price = float(params.get("min_price", 1000.0))
    use_defense = _to_bool(params.get("use_defense_filters"), True)
    use_high_chase_f = _to_bool(params.get("use_high_chase_filter"), False)
    high_chase_thr = float(params.get("high_chase_thr", 0.96))
    trigger_e_confirm = _to_bool(params.get("trigger_e_confirm"), True)
    trigger_bull_bar = _to_bool(params.get("trigger_require_bull_bar"), True)
    use_vol_trigger = _to_bool(params.get("use_vol_trigger"), True)
    use_rsi_filter = _to_bool(params.get("use_rsi_filter"), False)
    rsi_period = int(params.get("rsi_period", 3))
    rsi_min = float(params.get("mom_rsi_min", 50.0))
    rsi_max = float(params.get("mom_rsi_max", 80.0))

    c = candles[i]
    day = c["candle_time"][:8]
    hm = int(c["candle_time"][8:12])
    op = float(c.get("open", 0) or 0)
    cl = float(c.get("close", 0) or 0)
    hi = float(c.get("high", cl) or cl)

    if hm < time_start_hm or hm >= time_end_hm:
        return (None, None, None)

    if use_defense and cl < min_price:
        return ("탈락-최소가격", "%.0f < %.0f" % (cl, min_price), None)

    last_exit_dt = state.get("last_exit_dt")
    if last_exit_dt is not None:
        elapsed = (_t2dt(c["candle_time"]) - last_exit_dt).total_seconds() / 60
        if elapsed < cooldown_min:
            return (None, None, None)

    if state.get("daily_cnt", 0) >= max_daily:
        return (None, None, None)

    skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), False)
    prev_open = resolve_prev_trading_day_open(candles, i, day)

    # kiwoom momentum SCAN — E∧F∧H∧I 이미 통과, TRIGGER 는 타이밍·호가·휩쏘만
    if skip_hts:
        sig_hts: Dict[str, Any] = {
            "signal": True,
            "mode": "momentum_hts",
            "pattern": "momentum_hts_scan",
            "signal_candle_time": c.get("candle_time"),
            "prev_day_open": prev_open,
        }
        ws_rej, ws_msg = whipsaw_reject_for_signal(
            params, "MOMENTUM", signal_bar=c, current_price=cl,
        )
        if ws_rej:
            return (ws_rej, ws_msg, None)
        ob_rej, ob_msg = orderbook_reject_for_entry(params, "MOMENTUM", current_price=cl)
        if ob_rej:
            return (ob_rej, ob_msg, None)
        prog_rej, prog_msg = program_reject_for_entry(params, "MOMENTUM", current_price=cl)
        if prog_rej:
            return (prog_rej, prog_msg, None)
        return (None, None, sig_hts)

    # ── E: 전일 시가 돌파 유지 (HTS momentum 핵심) ─────────────────────
    if trigger_e_confirm:
        if prev_open is None or prev_open <= 0:
            return ("탈락-전일시가없음", "전일 시가 미확인", None)
        if cl <= prev_open:
            return (
                "탈락-E미충족",
                "종가 %.0f ≤ 전일시가 %.0f" % (cl, prev_open),
                None,
            )

    # ── 양봉: 당일 매수세 확인 (선택) ─────────────────────────────────
    if trigger_bull_bar and op > 0 and cl <= op:
        return ("탈락-음봉", "양봉 미충족", None)

    # ── 거래량 펄스: H/I 정신 (선택) ───────────────────────────────────
    if use_vol_trigger:
        vol_ok, vol_msg = _volume_pulse_ok(candles, i, params)
        if not vol_ok:
            return ("탈락-거래량", vol_msg, None)

    # ── 고점추격 방지 (선택, 기본 OFF) ───────────────────────────────
    if use_high_chase_f:
        running_high = hi
        for j in range(i, -1, -1):
            if candles[j]["candle_time"][:8] != day:
                break
            running_high = max(running_high, float(candles[j].get("high", 0) or 0))
        if running_high > 0 and cl >= running_high * high_chase_thr:
            return (
                "탈락-고점추격",
                "현재가 %.0f ≥ 고가 %.0f × %.2f" % (cl, running_high, high_chase_thr),
                None,
            )

    # ── RSI 필터 (선택, 기본 OFF — SCAN 이후 보조만) ───────────────────
    rsi_val: Optional[float] = None
    if use_rsi_filter and compute_rsi_series_fn is not None:
        closes = [float(x["close"]) for x in candles]
        ic = params.get("_indicator_cache")
        if ic is not None and hasattr(ic, "rsi_at"):
            rsi_val = ic.rsi_at(i, rsi_period)
        else:
            rsis = compute_rsi_series_fn(closes, rsi_period)
            rsi_val = rsis[i] if i < len(rsis) else None
        if rsi_val is None:
            return ("탈락-RSI없음", "RSI 미계산", None)
        if rsi_val < rsi_min:
            return ("탈락-RSI약함", "RSI=%.1f < %.0f" % (rsi_val, rsi_min), None)
        if rsi_val > rsi_max:
            return ("탈락-RSI과열", "RSI=%.1f > %.0f" % (rsi_val, rsi_max), None)

    sig: Dict[str, Any] = {
        "signal": True,
        "mode": "momentum_hts",
        "pattern": "hts_e_confirm",
        "signal_candle_time": c.get("candle_time"),
        "prev_day_open": prev_open,
    }
    if rsi_val is not None:
        sig["rsi"] = rsi_val

    ws_rej, ws_msg = whipsaw_reject_for_signal(
        params, "MOMENTUM",
        signal_bar=c,
        current_price=cl,
    )
    if ws_rej:
        return (ws_rej, ws_msg, None)
    ob_rej, ob_msg = orderbook_reject_for_entry(
        params, "MOMENTUM", current_price=cl,
    )
    if ob_rej:
        return (ob_rej, ob_msg, None)
    prog_rej, prog_msg = program_reject_for_entry(
        params, "MOMENTUM", current_price=cl,
    )
    if prog_rej:
        return (prog_rej, prog_msg, None)
    return (None, None, sig)


def check_sell_signal_momentum_hts_live(
    position: Dict[str, Any],
    current_candle: Dict[str, Any],
    params: Dict[str, Any],
    is_eod: bool = False,
) -> Optional[Tuple[str, float]]:
    """
    HTS momentum 추세추격 청산 — 어깨·래칫·트레일 병행.

    [청산 우선순위]
      1순위 래칫컷 (설정 시)
      2순위 어깨컷 (고점 대비 되돌림)
      3순위 손절 (하드 스탑)
      4순위 트레일컷 (추세 이익 보호)
      5순위 시간컷
      6순위 금액손실컷 (어깨·래칫 미발동 시)
      7순위 익절 (tp_max 상한, 하드 캡)
      8순위 장마감청산
    """
    sl_pct = -abs(float(params.get("sl_pct", params.get("stop_loss_pct", 0.03))))
    tp_pct = effective_tp_pct_from_params(params)
    trail_pct = abs(float(params.get("trail_pct", 0.0) or 0.0))
    trail_arm_pct = abs(float(params.get("trail_arm_pct", 0.0) or 0.0))
    shoulder_min_high, shoulder_cut_pct = _shoulder_ratios(params)
    ratchet_tiers = _parse_ratchet_tiers(params)
    max_hold_bars = int(params.get("max_hold_bars", 0) or 0)
    max_loss_krw = float(params.get("max_loss_krw", 200_000.0))
    min_hold_sec = float(params.get("min_hold_sec", 30.0))
    min_drop_pct = float(params.get("min_drop_pct_for_loss_cut", 0.015))

    try:
        hi = float(current_candle.get("high", current_candle["close"]))
        lo = float(current_candle.get("low", current_candle["close"]))
        cl = float(current_candle["close"])
    except Exception:
        return None

    candle_time = current_candle.get("candle_time", "")
    if candle_time and position.get("entry_time"):
        try:
            if (_t2dt(candle_time) - _t2dt(position["entry_time"])).total_seconds() < min_hold_sec:
                return None
        except Exception:
            pass

    max_price = max(float(position.get("max_price", position["entry_price"])), hi)
    position["max_price"] = max_price

    entry = float(position["entry_price"])
    qty = int(position.get("qty", 1) or 1)
    sl_line = entry * (1 + sl_pct)
    tp_line = entry * (1 + tp_pct)

    # 1순위 래칫
    if ratchet_tiers and entry > 0:
        peak_gain = (max_price - entry) / entry
        cut_ratio = 0.0
        for gain, cut in ratchet_tiers:
            if peak_gain >= gain:
                cut_ratio = cut
        if cut_ratio > 0.0:
            ratchet_line = max_price * (1.0 - cut_ratio)
            if lo <= ratchet_line:
                return ("래칫컷", ratchet_line)

    # 2순위 어깨
    shoulder_armed = entry > 0 and max_price >= entry * (1.0 + shoulder_min_high)
    if shoulder_armed and shoulder_cut_pct > 0:
        shoulder_line = max_price * (1.0 - shoulder_cut_pct)
        if lo <= shoulder_line:
            return ("어깨컷", shoulder_line)

    # 3순위 손절
    if lo <= sl_line:
        return ("손절", sl_line)

    # 4순위 트레일
    if trail_pct > 0 and max_price > entry:
        trail_arm_line = entry * (1.0 + trail_arm_pct)
        if trail_arm_pct <= 0 or max_price >= trail_arm_line:
            trail_line = max_price * (1.0 - trail_pct)
            if lo <= trail_line:
                return ("트레일컷", trail_line)

    # 5순위 시간컷
    if max_hold_bars > 0:
        held = _minutes_held(position, current_candle)
        if held is not None and held >= max_hold_bars:
            return ("시간컷", cl)

    # 6순위 금액손실컷
    profit_val = (lo - entry) * qty
    drop_pct = (entry - lo) / entry if entry > 0 else 0.0
    if (
        not shoulder_armed
        and not ratchet_tiers
        and profit_val <= -max_loss_krw
        and drop_pct >= min_drop_pct
    ):
        exit_px = entry - (max_loss_krw / qty) if qty > 0 else lo
        return ("금액손실컷", exit_px)

    # 7순위 익절 (하드 캡)
    if hi >= tp_line:
        return ("익절", tp_line)

    # 8순위 장마감
    if is_eod:
        return ("장마감청산", cl)

    return None