kis_bot/kis_trader/engine/range_break_engine.py

#!/usr/bin/env python3
"""
kis_trader/engine/range_break_engine.py — 박스권 돌파(RANGE_BREAK) 공통 엔진
====================================================================
[SCAN] HTS momentum 조건(F·G·J) — 오늘 거래 활발 종목 풀
[TRIGGER] 횡보 박스 형성 → 거래량 폭발 양봉으로 박스 상단 돌파 시 진입

매도: 박스 실패(상단 재이탈) → 익절 → 어깨 → 손절 → 트레일 → EOD
(청산 본체는 breakout 과 동일 우선순위 — ``check_sell_signal_breakout_live`` 재사용)
"""
from __future__ import annotations

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

from kis_trader.engine.range_break_env_keys import (
    range_break_env_bool,
    range_break_env_float,
    range_break_env_int,
)
from kis_trader.strategies.breakout import (
    check_sell_signal_breakout_live,
    normalize_breakout_max_loss_krw,
)

RANGE_BREAK_STRATEGY_ID = "RANGE_BREAK"


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 _confirmed_candles_only(candles: List[Dict]) -> List[Dict]:
    confirmed = [c for c in candles if _to_bool(c.get("is_confirmed", 1), True)]
    return confirmed if confirmed else list(candles)


def _t2dt(t: str) -> datetime:
    return datetime.strptime(str(t)[:12], "%Y%m%d%H%M")


def _slot_key(candle_time: str, scan_interval_min: int = 1) -> str:
    date = candle_time[:8]
    hm = int(candle_time[8:12])
    total_min = (hm // 100) * 60 + (hm % 100)
    slot_min = (total_min // scan_interval_min) * scan_interval_min
    slot_hm = (slot_min // 60) * 100 + (slot_min % 60)
    return date + str(slot_hm).zfill(4)


def _day_running_high_low(candles: List[Dict], i: int, day: str) -> Tuple[float, float, float]:
    running_high = float(candles[i]["high"])
    running_low = float(candles[i]["low"])
    day_open = float(candles[i]["open"])
    for j in range(i, -1, -1):
        if candles[j]["candle_time"][:8] != day:
            break
        running_high = max(running_high, float(candles[j]["high"]))
        running_low = min(running_low, float(candles[j]["low"]))
        day_open = float(candles[j]["open"])
    return running_high, running_low, day_open


def _box_window(candles: List[Dict], i: int, box_lookback: int) -> List[Dict]:
  """신호봉 i 직전 box_lookback 개 확정봉 = 박스 구간."""
  start = max(0, i - box_lookback)
  return candles[start:i]


def eval_range_break_buy_at_index(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    day_open: Optional[float] = None,
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """박스권 돌파 매수 TRIGGER — 인덱스 i 신호봉."""
    box_lookback = int(params.get("box_lookback_min", 30))
    box_max_w = float(params.get("box_max_width_pct", 2.5))
    box_min_w = float(params.get("box_min_width_pct", 0.3))
    setup_vol_mult = float(params.get("setup_vol_max_mult", 0.8))
    setup_bear_min = int(params.get("setup_bear_bars_min", 1))
    vol_mult = float(params.get("vol_mult", 2.0))
    vol_win = int(params.get("vol_window", 7))
    break_margin = float(params.get("break_margin_pct", 0.0) or 0.0)
    body_min = float(params.get("body_min_pct", 0.0) or 0.0)
    time_start = int(params.get("time_start_hm", 1030))
    time_end = int(params.get("time_end_hm", 1520))
    min_price = float(params.get("min_price", 1000.0))
    max_daily_chg = float(params.get("max_daily_chg", 25.0))
    use_high_chase = _to_bool(params.get("use_high_chase_filter"), True)
    high_chase_thr = float(params.get("high_chase_thr", 0.96))
    vol_baseline_win = int(params.get("vol_baseline_win", 30))

    need = max(box_lookback, vol_win, vol_baseline_win) + 3
    if i < need or i >= len(candles):
        return ("탈락-봉부족", "need=%d i=%d" % (need, i), None)

    c = candles[i]
    ct = str(c.get("candle_time", ""))
    if len(ct) < 12:
        return ("탈락-시간없음", ct, None)
    hm = int(ct[8:12])
    if hm < time_start or hm >= time_end:
        return (None, None, None)

    try:
        op = float(c["open"])
        hi = float(c["high"])
        lo = float(c["low"])
        cl = float(c["close"])
        vol = float(c.get("volume", 0) or 0)
    except Exception as e:
        return ("탈락-캔들파싱", str(e), None)

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

    box_arr = _box_window(candles, i, box_lookback)
    if len(box_arr) < max(3, box_lookback // 2):
        return ("탈락-박스창없음", "len=%d" % len(box_arr), None)

    box_high = max(float(x["high"]) for x in box_arr)
    box_low = min(float(x["low"]) for x in box_arr)
    if box_low <= 0 or box_high <= box_low:
        return ("탈락-박스무효", "high=%.0f low=%.0f" % (box_high, box_low), None)

    box_width_pct = (box_high - box_low) / box_low * 100.0
    if box_width_pct > box_max_w:
        return ("탈락-박스넓음", "%.2f%% > %.2f%%" % (box_width_pct, box_max_w), None)
    if box_width_pct < box_min_w:
        return ("탈락-박스좁음", "%.2f%% < %.2f%%" % (box_width_pct, box_min_w), None)

    box_vols = [float(x.get("volume", 0) or 0) for x in box_arr]
    box_vol_avg = sum(box_vols) / len(box_vols) if box_vols else 0.0
    base_start = max(0, i - box_lookback - vol_baseline_win)
    base_vols = [
        float(candles[k].get("volume", 0) or 0)
        for k in range(base_start, i - box_lookback)
    ]
    base_avg = sum(base_vols) / len(base_vols) if base_vols else box_vol_avg
    if base_avg > 0 and box_vol_avg > base_avg * setup_vol_mult:
        return (
            "탈락-박스거래량과다",
            "%.0f > %.0f×%.2f" % (box_vol_avg, base_avg, setup_vol_mult),
            None,
        )

    bear_cnt = sum(
        1 for x in box_arr
        if float(x.get("close", 0) or 0) < float(x.get("open", 0) or 0)
    )
    if bear_cnt < setup_bear_min:
        return ("탈락-음봉부족", "%d < %d" % (bear_cnt, setup_bear_min), None)

    if cl <= op:
        return ("탈락-음봉", "close<=open", None)

    need_price = box_high * (1.0 + break_margin / 100.0)
    if cl < need_price:
        return (
            "탈락-박스미돌파",
            "close=%.0f < need=%.0f" % (cl, need_price),
            None,
        )

    if body_min > 0 and op > 0:
        body_pct = (cl - op) / op * 100.0
        if body_pct < body_min:
            return ("탈락-몸통부족", "%.2f%%" % body_pct, None)

    vol_ref = box_vol_avg if box_vol_avg > 0 else 0.0
    if vol_ref <= 0:
        win_arr = [float(candles[k].get("volume", 0) or 0) for k in range(i - vol_win, i)]
        vol_ref = sum(win_arr) / len(win_arr) if win_arr else 0.0
    if vol_ref <= 0 or vol < vol_ref * vol_mult:
        ratio = vol / vol_ref if vol_ref > 0 else 0.0
        return ("탈락-거래량부족", "%.2fx < %.1fx" % (ratio, vol_mult), None)

    day = ct[:8]
    running_high, running_low, d_open = _day_running_high_low(candles, i, day)
    if day_open is None:
        day_open = d_open
    if day_open > 0:
        daily_chg = (cl - day_open) / day_open * 100.0
        if daily_chg > max_daily_chg:
            return ("탈락-이격과열", "%.1f%%" % daily_chg, None)

    if use_high_chase and running_high > 0 and cl >= running_high * high_chase_thr:
        return ("탈락-고점추격", "%.0f" % cl, None)

    vol_ratio = vol / vol_ref if vol_ref > 0 else 0.0
    return (None, None, {
        "box_high": box_high,
        "box_low": box_low,
        "box_width_pct": round(box_width_pct, 3),
        "vol_ratio": vol_ratio,
        "bear_bars": bear_cnt,
        "signal_candle_time": ct,
    })


def check_buy_signal_range_break_live(
    candles: List[Dict],
    params: Dict[str, Any],
    state: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """실매·백테 공용 — align: 신호봉 확정 후 다음 봉 시가 진입."""
    live_align = _to_bool(params.get("live_backtest_align", True), True)
    lookback = max(1, int(params.get("live_signal_lookback_bars", 1)))
    confirmed = _confirmed_candles_only(candles)
    if len(confirmed) < 10:
        return ("탈락-봉부족", "확정봉 부족", None)

    cooldown_min = float(params.get("cooldown_min", 30))
    max_daily = int(params.get("max_daily", 1))
    last_exit_dt = state.get("last_exit_dt")
    if last_exit_dt is not None:
        elapsed = (_t2dt(confirmed[-1]["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)

    last_reject: Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]] = (
        None, None, None,
    )
    if live_align:
        entry_i = len(confirmed) - 1
        for k in range(lookback):
            signal_i = entry_i - 1 - k
            if signal_i < 1:
                break
            reject, msg, sig = eval_range_break_buy_at_index(
                confirmed, signal_i, params, None,
            )
            if reject:
                if k == 0:
                    last_reject = (reject, msg, None)
                continue
            if sig:
                ent = confirmed[entry_i]
                entry_open = float(ent.get("open", 0) or 0)
                if entry_open <= 0:
                    entry_open = float(ent.get("close", 0) or 0)
                sig["entry_price"] = entry_open
                sig["entry_bar_key"] = ent.get("candle_time")
                sig["box_stop_line"] = float(sig.get("box_high", 0) or 0)
                return (None, None, sig)
        return last_reject

    i = len(confirmed) - 1
    return eval_range_break_buy_at_index(confirmed, i, params, None)


def check_sell_signal_range_break_live(
    position: Dict[str, Any],
    current_candle: Dict[str, Any],
    params: Dict[str, Any],
    is_eod: bool = False,
) -> Optional[Tuple[str, float]]:
    """박스 상단 재이탈 우선 → breakout 청산 체인."""
    try:
        lo = float(current_candle.get("low", current_candle["close"]))
        cl = float(current_candle["close"])
    except Exception:
        return None

    box_line = float(position.get("box_stop_line", 0) or 0)
    if box_line > 0 and lo < box_line:
        return ("box_fail", min(cl, box_line))

    return check_sell_signal_breakout_live(position, current_candle, params, is_eod=is_eod)


def get_range_break_defaults_from_db(db=None) -> Dict[str, Any]:
    """env_config + config_range_break 병합."""
    own_db = None
    r: Dict[str, Any] = {}
    try:
        if db is None:
            from database import TradeDB
            own_db = TradeDB()
            db = own_db
        if hasattr(db, "get_merged_env_snapshot"):
            r = db.get_merged_env_snapshot()
        elif hasattr(db, "get_latest_env"):
            latest = db.get_latest_env()
            r = dict((latest or {}).get("snapshot") or {})
        else:
            row = db.conn.execute(
                "SELECT * FROM env_config ORDER BY id DESC LIMIT 1"
            ).fetchone()
            r = dict(row) if row else {}
    except Exception:
        r = {}
    finally:
        if own_db is not None:
            try:
                own_db.close()
            except Exception:
                pass

    sl = abs(range_break_env_float(r, "RANGE_BREAK_STOP_LOSS_PCT", 0.03))
    tp = range_break_env_float(r, "RANGE_BREAK_TAKE_PROFIT_PCT", 0.10)
    trail = range_break_env_float(r, "RANGE_BREAK_TRAIL_PCT", 0.015)
    trail_arm = range_break_env_float(r, "RANGE_BREAK_TRAIL_ARM_PCT", 0.015)
    shoulder_high = range_break_env_float(r, "RANGE_BREAK_SHOULDER_MIN_HIGH_PCT", 0.03)
    shoulder_cut = range_break_env_float(r, "RANGE_BREAK_SHOULDER_CUT_PCT", 0.005)
    cooldown_sec = range_break_env_int(r, "RANGE_BREAK_COOLDOWN_SEC", 1800)

    return {
        "box_lookback_min": range_break_env_int(r, "RANGE_BREAK_BOX_LOOKBACK_MIN", 30),
        "box_max_width_pct": range_break_env_float(r, "RANGE_BREAK_BOX_MAX_WIDTH_PCT", 2.5),
        "box_min_width_pct": range_break_env_float(r, "RANGE_BREAK_BOX_MIN_WIDTH_PCT", 0.3),
        "setup_vol_max_mult": range_break_env_float(r, "RANGE_BREAK_SETUP_VOL_MAX_MULT", 0.8),
        "setup_bear_bars_min": range_break_env_int(r, "RANGE_BREAK_SETUP_BEAR_BARS_MIN", 1),
        "vol_mult": range_break_env_float(r, "RANGE_BREAK_VOL_MULT", 2.0),
        "vol_window": range_break_env_int(r, "RANGE_BREAK_VOL_WIN", 7),
        "break_margin_pct": range_break_env_float(r, "RANGE_BREAK_BREAK_MARGIN_PCT", 0.0),
        "body_min_pct": range_break_env_float(r, "RANGE_BREAK_BODY_MIN_PCT", 0.0),
        "time_start_hm": range_break_env_int(r, "RANGE_BREAK_TIME_START", 1030),
        "time_end_hm": range_break_env_int(r, "RANGE_BREAK_TIME_END_HM", 1520),
        "sl_pct": sl,
        "tp_pct": tp,
        "trail_pct": trail,
        "trail_arm_pct": trail_arm,
        "shoulder_min_high": shoulder_high,
        "shoulder_cut_pct": shoulder_cut,
        "max_hold_bars": range_break_env_int(r, "RANGE_BREAK_MAX_HOLD_BARS", 0),
        "max_daily": range_break_env_int(r, "RANGE_BREAK_MAX_DAILY", 1),
        "cooldown_min": cooldown_sec / 60.0,
        "max_daily_chg": range_break_env_float(r, "RANGE_BREAK_MAX_DAILY_CHG", 25.0),
        "min_price": range_break_env_float(r, "RANGE_BREAK_MIN_PRICE", 1000.0),
        "high_chase_thr": range_break_env_float(r, "RANGE_BREAK_HIGH_CHASE_THR", 0.96),
        "use_high_chase_filter": range_break_env_bool(r, "RANGE_BREAK_USE_HIGH_CHASE_FILTER", True),
        "max_loss_krw": float(range_break_env_int(r, "RANGE_BREAK_MAX_LOSS_PER_TRADE_KRW", 200_000)),
        "slot_money": float(range_break_env_int(r, "RANGE_BREAK_SLOT_MONEY", 200_000)),
        "max_stocks": range_break_env_int(r, "RANGE_BREAK_MAX_STOCKS", 20),
        "total_budget_krw": range_break_env_int(r, "RANGE_BREAK_TOTAL_BUDGET_KRW", 4_000_000),
        "live_backtest_align": range_break_env_bool(r, "RANGE_BREAK_LIVE_BACKTEST_ALIGN", True),
        "live_signal_lookback_bars": range_break_env_int(r, "RANGE_BREAK_LIVE_SIGNAL_LOOKBACK_BARS", 1),
        "force_eod_exit": range_break_env_bool(r, "RANGE_BREAK_FORCE_EOD_EXIT", False),
        "vol_baseline_win": range_break_env_int(r, "RANGE_BREAK_VOL_BASELINE_WIN", 30),
    }


def range_break_ui_to_engine_params(ui: Dict[str, Any]) -> Dict[str, Any]:
    """웹 UI(%) → 엔진 params."""
    sl_ui = float(ui.get("sl_pct", 3.0))
    max_loss = normalize_breakout_max_loss_krw(ui.get("max_loss_krw", 200_000))
    slot_cap = float(ui.get("slot_money", 200_000))
    from kis_trader.strategies.breakout import breakout_invest_amount_krw
    slot = breakout_invest_amount_krw(max_loss, sl_ui, slot_cap)

    def _ui_bool(key: str, default: bool = False) -> bool:
        raw = ui.get(key)
        if raw is None or raw == "":
            return default
        if isinstance(raw, bool):
            return raw
        return str(raw).strip().lower() in ("1", "true", "t", "y", "yes", "on")

    return {
        "box_lookback_min": int(ui.get("box_lookback_min", 30)),
        "box_max_width_pct": float(ui.get("box_max_width_pct", 2.5)),
        "box_min_width_pct": float(ui.get("box_min_width_pct", 0.3)),
        "setup_vol_max_mult": float(ui.get("setup_vol_max_mult", 0.8)),
        "setup_bear_bars_min": int(ui.get("setup_bear_bars_min", 1)),
        "vol_mult": float(ui.get("vol_mult", 2.0)),
        "vol_window": int(ui.get("vol_window", 7)),
        "vol_baseline_win": int(ui.get("vol_baseline_win", 30)),
        "break_margin_pct": float(ui.get("break_margin_pct", 0.0) or 0.0),
        "body_min_pct": float(ui.get("body_min_pct", 0.0) or 0.0),
        "time_start_hm": int(ui.get("time_start_hm", 1030)),
        "time_end_hm": int(ui.get("time_end_hm", 1520)),
        "cooldown_min": float(ui.get("cooldown_min", 30)),
        "max_daily": int(ui.get("max_daily", 1)),
        "max_daily_chg": float(ui.get("max_daily_chg", 25.0)),
        "min_price": float(ui.get("min_price", 1000.0)),
        "high_chase_thr": float(ui.get("high_chase_thr", 0.96)),
        "use_high_chase_filter": _ui_bool("use_high_chase_filter", True),
        "max_loss_krw": max_loss,
        "slot_money": slot,
        "stop_loss_pct": -abs(sl_ui) / 100.0,
        "take_profit_pct": abs(float(ui.get("tp_pct", 10.0))) / 100.0,
        "trail_pct": abs(float(ui.get("trail_pct", 1.5))) / 100.0,
        "trail_arm_pct": abs(float(ui.get("trail_arm_pct", 1.5) or 0.0)) / 100.0,
        "shoulder_min_high": abs(float(ui.get("shoulder_min_high_pct", 3.0))) / 100.0,
        "shoulder_cut_pct": abs(float(ui.get("shoulder_cut_pct", 0.5))) / 100.0,
        "max_hold_bars": int(float(ui.get("max_hold_bars", 0) or 0)),
        "fee_rate": float(ui.get("fee_rate_pct", 0.015) or 0.015) / 100.0,
        "sell_tax": float(ui.get("sell_tax_pct", 0.18) or 0.18) / 100.0,
        "live_backtest_align": _ui_bool("live_backtest_align", True),
        "live_signal_lookback_bars": int(ui.get("live_signal_lookback_bars", 1)),
        "force_eod_exit": _ui_bool("force_eod_exit", False),
        "portfolio_mode": True,
    }


def range_break_min_bars_required(params: Dict[str, Any]) -> int:
    """백테 최소 봉 수 — 박스 lookback + 거래량 기준창."""
    box_lb = int(params.get("box_lookback_min", 30))
    vol_win = int(params.get("vol_window", 7))
    vol_base = int(params.get("vol_baseline_win", 30))
    align_extra = 1 if _to_bool(params.get("live_backtest_align", True), True) else 0
    return max(box_lb, vol_win, vol_base) + 3 + align_extra


def range_break_scan_buy_at_bar(
    candles: List[Dict],
    bar_index: int,
    params: Dict[str, Any],
    state: Optional[Dict[str, Any]] = None,
    day_open: Optional[float] = None,
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]], float, str]:
    """
    백테 단일 봉 매수 스캔 — align: 신호봉 확정 후 다음 봉 시가 진입.

    Returns:
        reason, msg, signal, entry_price, entry_time(candle_time)
    """
    i = bar_index
    st = state or {}
    sub = candles[: i + 1]
    reason, msg, sig = check_buy_signal_range_break_live(sub, params, st)
    if not sig:
        return reason, msg, None, 0.0, ""
    if i + 1 >= len(candles):
        return ("탈락-진입봉없음", "다음 봉 없음", None, 0.0, "")
    next_c = candles[i + 1]
    if next_c["candle_time"][:8] != sub[-1]["candle_time"][:8]:
        return ("탈락-일자넘김", "다음봉 일자 불일치", None, 0.0, "")
    ep = float(next_c.get("open") or 0)
    if ep <= 0:
        ep = float(sig.get("entry_price") or sig.get("close") or 0)
    sig["box_stop_line"] = float(sig.get("box_high", 0) or 0)
    return None, None, sig, ep, str(next_c.get("candle_time") or "")


def run_range_break_backtest(
    codes_candles: Dict[str, List[Dict]],
    params: Dict[str, Any],
    universe_by_slot: Optional[Dict[str, List[str]]] = None,
) -> List[Dict]:
    from kis_trader.backtest.range_break_portfolio_backtest import (
        run_range_break_backtest_portfolio,
    )
    return run_range_break_backtest_portfolio(
        codes_candles, params, universe_by_slot=universe_by_slot,
    )