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