kis_bot/kis_trader/backtest/optuna_tpe_common.py

#!/usr/bin/env python3
"""optuna_tpe_common.py — 전략 공통 TPE 헬퍼 (유효 HHMM · 소수 반올림 · 래칫 숫자축)."""
from __future__ import annotations

from typing import Any, Dict, List

import optuna


def hm_to_minutes(hm: int) -> int:
    h = int(hm) // 100
    m = int(hm) % 100
    return h * 60 + m


def minutes_to_hm(mins: int) -> int:
    h = int(mins) // 60
    m = int(mins) % 60
    return h * 100 + m


def hm_choices(lo_hm: int, hi_hm: int, step_min: int = 10) -> List[int]:
    """
    유효 HHMM 목록 (1460 같은 가짜 시각 방지).
    예: hm_choices(1400, 1530, 10) → 1400,1410,…,1530
    """
    step = max(1, int(step_min))
    a = hm_to_minutes(lo_hm)
    b = hm_to_minutes(hi_hm)
    if b < a:
        a, b = b, a
    out: List[int] = []
    for m in range(a, b + 1, step):
        out.append(minutes_to_hm(m))
    if out and out[-1] != minutes_to_hm(b):
        out.append(minutes_to_hm(b))
    # 중복 제거·정렬
    return sorted(set(out))


def r1(x: float) -> float:
    return round(float(x), 1)


def r2(x: float) -> float:
    return round(float(x), 2)


def r3(x: float) -> float:
    return round(float(x), 3)


def r4(x: float) -> float:
    return round(float(x), 4)


# 래칫 숫자축 키 (JSON axis / mode_combo 빈도용) — 엔진에는 ratchet_tiers 문자열만 전달
RATCHET_TPE_AXIS_KEYS: List[str] = [
    "ratchet_on",
    "ratchet_n",
    "ratchet_gain_1",
    "ratchet_cut_1",
    "ratchet_gain_2",
    "ratchet_cut_2",
    "ratchet_gain_3",
    "ratchet_cut_3",
    "ratchet_tiers",  # 조립 결과(엔진·apply용)
]


def format_ratchet_tiers_string(pairs: List[tuple]) -> str:
    """[(gain%, cut%), ...] → \"5:2,10:1\" (엔진 % 문자열)."""
    parts: List[str] = []
    for g, c in pairs:
        # 불필요 소수 꼬리 제거
        gs = f"{float(g):g}"
        cs = f"{float(c):g}"
        parts.append(f"{gs}:{cs}")
    return ",".join(parts)


def assemble_ratchet_tiers_from_combo(
    combo: Dict[str, Any],
    *,
    prefix: str = "ratchet",
    off_token: str = "",
) -> str:
    """
    mode_combo 최빈값 등에서 엔진용 ratchet_tiers 재조립.
    ratchet_on=False / n<=0 → off_token (모멘텀·돌파 \"\" / 꼬리 \"off\").
    """
    on = combo.get(f"{prefix}_on")
    if on is False or on in (0, "0", "False", "false", "off", "OFF"):
        return str(off_token)
    try:
        n = int(combo.get(f"{prefix}_n") or 0)
    except (TypeError, ValueError):
        n = 0
    if n <= 0:
        # 숫자축 없으면 기존 문자열 유지
        existing = combo.get("ratchet_tiers")
        if existing is not None and str(existing).strip() != "":
            return str(existing).strip()
        return str(off_token)
    pairs: List[tuple] = []
    prev = 0.0
    for i in range(1, n + 1):
        g = combo.get(f"{prefix}_gain_{i}")
        c = combo.get(f"{prefix}_cut_{i}")
        if g is None or c is None:
            continue
        try:
            gf = float(g)
            cf = float(c)
        except (TypeError, ValueError):
            continue
        if gf <= prev + 1e-9:
            continue
        pairs.append((gf, cf))
        prev = gf
    if not pairs:
        return str(off_token)
    return format_ratchet_tiers_string(pairs)


def finalize_ratchet_combo(
    combo: Dict[str, Any],
    *,
    prefix: str = "ratchet",
    off_token: str = "",
) -> Dict[str, Any]:
    """combo 에 ratchet_tiers 를 숫자축 기준으로 덮어쓴다 (evaluate/apply 직전)."""
    if not isinstance(combo, dict):
        return combo
    if f"{prefix}_on" not in combo and f"{prefix}_n" not in combo:
        return combo
    out = dict(combo)
    out["ratchet_tiers"] = assemble_ratchet_tiers_from_combo(
        out, prefix=prefix, off_token=off_token,
    )
    return out


def suggest_ratchet_tiers_pct(
    trial: optuna.Trial,
    *,
    prefix: str = "ratchet",
    off_allowed: bool = True,
    off_token: str = "",
    n_max: int = 3,
    gain_low: float = 2.0,
    gain_high: float = 15.0,
    gain_step: float = 0.5,
    cut_low: float = 0.5,
    cut_high: float = 3.0,
    cut_step: float = 0.1,
) -> Dict[str, Any]:
    """
    다단 래칫을 숫자 축으로 suggest → 엔진용 문자열 조립.

    반환 dict:
      ratchet_on, ratchet_n, ratchet_gain_i, ratchet_cut_i, ratchet_tiers
    ratchet_tiers 예: \"5:2,10:1\" / OFF 시 off_token (기본 \"\").

    multivariate TPE: ON/OFF·단수와 무관하게 **항상 동일 키·동일 분포**로
    suggest 한다 (조건부 분포 → independent sampling 경고·성능저하 방지).
    실제 사용은 on=True 일 때 앞쪽 ratchet_n단만. gain 비오름차순이면 TrialPruned.
    """
    n_max = max(1, min(3, int(n_max)))
    out: Dict[str, Any] = {}

    if off_allowed:
        on = bool(trial.suggest_categorical(f"{prefix}_on", [False, True]))
    else:
        on = True
    out[f"{prefix}_on"] = on

    # OFF여도 n·gain·cut 전부 suggest (키 공간·분포 고정)
    n = int(trial.suggest_int(f"{prefix}_n", 1, n_max))
    gains: List[float] = []
    cuts: List[float] = []
    for i in range(1, n_max + 1):
        g = r1(trial.suggest_float(
            f"{prefix}_gain_{i}", gain_low, gain_high, step=gain_step,
        ))
        c = r2(trial.suggest_float(
            f"{prefix}_cut_{i}", cut_low, cut_high, step=cut_step,
        ))
        gains.append(g)
        cuts.append(c)
        out[f"{prefix}_gain_{i}"] = g
        out[f"{prefix}_cut_{i}"] = c

    if not on:
        out[f"{prefix}_n"] = 0
        out["ratchet_tiers"] = str(off_token)
        out["_ratchet_ascending_ok"] = True
        return out

    out[f"{prefix}_n"] = n
    pairs: List[tuple] = []
    prev_gain = 0.0
    ascending_ok = True
    for i in range(n):
        g = float(gains[i])
        c = float(cuts[i])
        if g <= prev_gain + 1e-12:
            ascending_ok = False
            break
        pairs.append((g, c))
        prev_gain = g

    # TrialPruned 는 호출부에서 모든 suggest 끝난 뒤 (multivariate TPE 키 고정)
    out["_ratchet_ascending_ok"] = ascending_ok
    out["ratchet_tiers"] = (
        format_ratchet_tiers_string(pairs) if ascending_ok else str(off_token)
    )
    return out


# ── 호가·휩쏘 TPE 축 (본 trial 엔진 평가용, 사후「필터후」대체) ───────────────
# multivariate TPE: 키는 항상 suggest (OFF 여부와 무관). 범위는 OPTUNA_OB_ENTRY_* env.
# 임계값만 (스터디 스위치로 ON/OFF 고정할 때 — 돌파 등)
ORDERBOOK_TPE_THRESHOLD_KEYS: List[str] = [
    "max_spread_pct",
    "min_bid_ask_ratio",
    "ask_max_mult",
]
ORDERBOOK_TPE_AXIS_KEYS: List[str] = [
    "_orderbook_filter_enabled",
    *ORDERBOOK_TPE_THRESHOLD_KEYS,
]

WHIPSAW_TPE_AXIS_KEYS: List[str] = [
    "whipsaw_filter_enabled",
    "whipsaw_subbar_sec",
    "whipsaw_lookback_sec",
    "whipsaw_dip_pct",
    "whipsaw_recovery_tol_pct",
]


def optuna_tpe_include_orderbook() -> bool:
    """본 TPE에 호가 축 포함 (기본 ON). 끄면 구 타점-only 탐색."""
    from kis_trader.utils.env import get_env_bool

    return bool(get_env_bool("OPTUNA_TPE_INCLUDE_ORDERBOOK", True))


def optuna_tpe_include_whipsaw() -> bool:
    """본 TPE에 휩쏘 축 포함 (기본 ON). 꼬리·스캘·모멘텀."""
    from kis_trader.utils.env import get_env_bool

    return bool(get_env_bool("OPTUNA_TPE_INCLUDE_WHIPSAW", True))


def optuna_tpe_needs_orderbook_feed(mode: str, orderbook_filter: str = "off") -> bool:
    """TPE+호가축 또는 CLI 호가 ON → ws_orderbook 스냅 로드."""
    m = (mode or "").strip().lower()
    ob = (orderbook_filter or "off").strip().lower()
    if ob in ("on", "auto"):
        return True
    if m == "tpe" and optuna_tpe_include_orderbook():
        return True
    return False


def suggest_orderbook_thresholds_tpe(trial: optuna.Trial) -> Dict[str, Any]:
    """호가 임계값만 (ON/OFF 는 스터디 스위치·CLI). 돌파 TPE 등."""
    from kis_trader.utils.env import get_env_float

    lo_s = float(get_env_float("OPTUNA_OB_ENTRY_SPREAD_MIN", 0.1))
    hi_s = float(get_env_float("OPTUNA_OB_ENTRY_SPREAD_MAX", 8.0))
    lo_r = float(get_env_float("OPTUNA_OB_ENTRY_RATIO_MIN", 0.05))
    hi_r = float(get_env_float("OPTUNA_OB_ENTRY_RATIO_MAX", 1.5))
    lo_a = float(get_env_float("OPTUNA_OB_ENTRY_ASK_MULT_MIN", 1.0))
    hi_a = float(get_env_float("OPTUNA_OB_ENTRY_ASK_MULT_MAX", 80.0))
    if hi_s < lo_s:
        lo_s, hi_s = hi_s, lo_s
    if hi_r < lo_r:
        lo_r, hi_r = hi_r, lo_r
    if hi_a < lo_a:
        lo_a, hi_a = hi_a, lo_a
    return {
        "max_spread_pct": r1(trial.suggest_float("max_spread_pct", lo_s, hi_s, step=0.1)),
        "min_bid_ask_ratio": r2(trial.suggest_float("min_bid_ask_ratio", lo_r, hi_r, step=0.05)),
        "ask_max_mult": r1(trial.suggest_float("ask_max_mult", lo_a, hi_a, step=1.0)),
    }


def suggest_orderbook_entry_tpe(trial: optuna.Trial) -> Dict[str, Any]:
    """진입 호가 필터 축 — evaluate 의 _orderbook_filter_enabled / max_spread_* 와 동일 키.

    모멘텀·스캘·꼬리: trial 마다 ON/OFF categorical.
    돌파는 ``suggest_breakout_params_tpe`` 가 CLI 스위치로 고정(여기 안 씀).
    """
    if not optuna_tpe_include_orderbook():
        return {}
    enabled = bool(trial.suggest_categorical("_orderbook_filter_enabled", [False, True]))
    out = {
        "_orderbook_filter_enabled": enabled,
        # apply(orderbook_params_to_env_patch) 별칭
        "ob_filter_enabled": enabled,
    }
    out.update(suggest_orderbook_thresholds_tpe(trial))
    return out


def suggest_whipsaw_tpe(trial: optuna.Trial, *, force: bool = False) -> Dict[str, Any]:
    """휩쏘 TRIGGER 축 — merge_whipsaw_cfg_from_params / 꼬리 TPE 와 동일 키.

    force=True: 꼬리처럼 축이 필수인 전략 (INCLUDE_WHIPSAW OFF 여도 키 공간 유지).
    """
    if not force and not optuna_tpe_include_whipsaw():
        return {}
    return {
        "whipsaw_filter_enabled": trial.suggest_categorical("whipsaw_filter_enabled", [False, True]),
        "whipsaw_subbar_sec": trial.suggest_categorical(
            "whipsaw_subbar_sec", [15, 30, 45, 60],
        ),
        "whipsaw_lookback_sec": trial.suggest_categorical(
            "whipsaw_lookback_sec", [60, 90, 120, 180],
        ),
        "whipsaw_dip_pct": r4(
            trial.suggest_float("whipsaw_dip_pct", 0.001, 0.01, step=0.001),
        ),
        "whipsaw_recovery_tol_pct": r4(
            trial.suggest_float("whipsaw_recovery_tol_pct", 0.0005, 0.003, step=0.0005),
        ),
    }