#!/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