kis_bot/kis_trader/backtest/optuna_breakout_tpe_space.py

#!/usr/bin/env python3
"""
optuna_breakout_tpe_space.py — 돌파 Optuna 연속(TPE) 탐색 공간
기존 ``_breakout_grids()`` categorical 경로는 유지. ``skip_hts`` 항상 False.

호가 ON/OFF 는 trial 축이 아님 — CLI ``--orderbook-filter on|off`` 스터디 스위치.
(fixed/atr 손절과 같이 한 스터디에 섞지 않음 → 최대 fixed|atr × off|on = 4순차)
"""
from __future__ import annotations

from typing import Any, Dict, List, Optional

import optuna

from kis_trader.backtest.optuna_tpe_common import (
    ORDERBOOK_TPE_THRESHOLD_KEYS,
    RATCHET_TPE_AXIS_KEYS,
    r1,
    r2,
    suggest_orderbook_thresholds_tpe,
    suggest_ratchet_tiers_pct,
)

# 매수 시간창(time_end_hm) 은 TPE 제외 — 운영 DB/골든타임 고정. 다단래칫·청산축만 탐색.
_BREAKOUT_BASE_KEYS: List[str] = [
    "max_daily_chg",
    "vol_mult",
    "vol_window",
    "min_turnover_1m_pct",
    "prev_chg_min",
    "prev_chg_max",
    "min_price",
    "tp_pct",
    "sl_mode",
    "trail_pct",
    "trail_arm_pct",
    "shoulder_min_high_pct",
    "shoulder_cut_pct",
    "lookback_min",
    "confirm_margin_pct",
    "body_min_pct",
    "max_hold_bars",
]
# 돌파: 본축 + 래칫. 호가 ON/OFF·임계는 스터디 스위치(아래 breakout_tpe_axis_keys).
# 휩쏘는 전략 특성상 TPE 스킵 — 기존과 동일.
BREAKOUT_TPE_AXIS_KEYS: List[str] = (
    list(_BREAKOUT_BASE_KEYS) + list(RATCHET_TPE_AXIS_KEYS)
)


def normalize_tpe_breakout_sl_mode(raw: Optional[Any] = None) -> str:
    """TPE는 손절모드를 탐색하지 않고 스터디마다 고정. fixed | atr."""
    s = str(raw or "").strip().lower()
    if s in ("atr", "atr_sl", "dynamic"):
        return "atr"
    return "fixed"


def normalize_tpe_breakout_ob_mode(raw: Optional[Any] = None) -> str:
    """돌파 TPE 호가 스터디 스위치. off | on (auto→on). trial 축 아님."""
    s = str(raw or "off").strip().lower()
    if s in ("on", "1", "true", "yes", "auto"):
        return "on"
    return "off"


def breakout_tpe_study_extra(
    sl_mode: Optional[str] = None,
    orderbook_filter: Optional[str] = None,
) -> str:
    """study 이름용 — 예: fixed_ob_off / atr_ob_on (한 스터디에 손절·호가 안 섞음)."""
    sm = normalize_tpe_breakout_sl_mode(sl_mode)
    om = normalize_tpe_breakout_ob_mode(orderbook_filter)
    return f"{sm}_ob_{om}"


def breakout_tpe_axis_keys(
    sl_mode: Optional[str] = None,
    orderbook_filter: Optional[str] = None,
) -> List[str]:
    """fixed=sl_pct 만, atr=atr_sl_mult 만. 호가 ON 스터디만 임계값 축 추가."""
    keys = list(BREAKOUT_TPE_AXIS_KEYS)
    if normalize_tpe_breakout_sl_mode(sl_mode) == "atr":
        keys.append("atr_sl_mult")
    else:
        keys.append("sl_pct")
    if normalize_tpe_breakout_ob_mode(orderbook_filter) == "on":
        keys.extend(list(ORDERBOOK_TPE_THRESHOLD_KEYS))
    return keys


def suggest_breakout_params_tpe(
    trial: optuna.Trial,
    sl_mode: Optional[str] = None,
    orderbook_filter: Optional[str] = None,
) -> Dict[str, Any]:
    combo: Dict[str, Any] = {}
    combo["max_daily_chg"] = r1(trial.suggest_float("max_daily_chg", 20.0, 55.0, step=1.0))
    combo["vol_mult"] = r2(trial.suggest_float("vol_mult", 1.0, 5.0, step=0.1))  # 확장: 1.5→1.0, 4.0→5.0
    combo["vol_window"] = trial.suggest_int("vol_window", 1, 15)
    combo["min_turnover_1m_pct"] = r2(
        trial.suggest_float("min_turnover_1m_pct", 0.05, 0.5, step=0.05),
    )
    combo["prev_chg_min"] = r2(trial.suggest_float("prev_chg_min", 0.2, 1.0, step=0.1))
    combo["prev_chg_max"] = r1(trial.suggest_float("prev_chg_max", 5.0, 30.0, step=0.5))
    if combo["prev_chg_min"] >= combo["prev_chg_max"]:
        raise optuna.TrialPruned("prev_chg invalid")

    combo["min_price"] = trial.suggest_int("min_price", 1000, 5000, step=500)
    combo["tp_pct"] = r2(trial.suggest_float("tp_pct", 1.0, 25.0, step=0.5))  # 확장: 2.0→1.0, 18.0→25.0
    # 한 스터디=한 손절모드. categorical 혼입 금지(웹 체크 2개면 잡 2개 순차).
    combo["sl_mode"] = normalize_tpe_breakout_sl_mode(sl_mode)
    if combo["sl_mode"] == "atr":
        combo["atr_sl_mult"] = r2(trial.suggest_float("atr_sl_mult", 1.5, 3.5, step=0.1))
    else:
        combo["sl_pct"] = r2(trial.suggest_float("sl_pct", 1.0, 8.0, step=0.1))  # 확장: 2.0→1.0, 6.0→8.0
    combo["trail_pct"] = r2(trial.suggest_float("trail_pct", 0.0, 6.0, step=0.1))  # 확장: 1.0→0.0, 4.0→6.0
    combo["trail_arm_pct"] = r2(trial.suggest_float("trail_arm_pct", 0.0, 6.0, step=0.1))  # 확장: 1.0→0.0, 4.0→6.0
    combo["shoulder_min_high_pct"] = r2(
        trial.suggest_float("shoulder_min_high_pct", 1.0, 6.0, step=0.1),
    )
    combo["shoulder_cut_pct"] = r2(
        trial.suggest_float("shoulder_cut_pct", 0.3, 1.5, step=0.1),
    )
    combo.update(
        suggest_ratchet_tiers_pct(
            trial,
            off_token="",
            n_max=3,
            gain_low=2.0,
            gain_high=15.0,
            gain_step=0.5,
            cut_low=0.5,
            cut_high=3.0,
            cut_step=0.1,
        ),
    )
    combo["lookback_min"] = trial.suggest_int("lookback_min", 1, 10)
    combo["confirm_margin_pct"] = r2(
        trial.suggest_float("confirm_margin_pct", 0.0, 1.0, step=0.1),
    )
    combo["body_min_pct"] = r2(trial.suggest_float("body_min_pct", 0.0, 0.5, step=0.1))
    combo["max_hold_bars"] = trial.suggest_int("max_hold_bars", 0, 180, step=10)

    combo["skip_hts_scan_dupes"] = False
    # 호가: 스터디 스위치 (trial categorical ON/OFF 금지 — TPE가 OFF만 편애하던 구멍)
    ob_on = normalize_tpe_breakout_ob_mode(orderbook_filter) == "on"
    combo["_orderbook_filter_enabled"] = ob_on
    combo["ob_filter_enabled"] = ob_on
    if ob_on:
        combo.update(suggest_orderbook_thresholds_tpe(trial))
    return combo