ls증권 히스토리 구독 넣음

kis_trader/backtest/optuna_daily_trail_recommend.py

@@ -0,0 +1,315 @@
+#!/usr/bin/env python3
+"""
+Optuna 결과(mode_combo top_pnls / mode·best PnL) → 당일손익 다단트레일 추천.
+
+- 탐색 축에는 넣지 않음 (그리드/TPE 제외).
+- JSON·Optuna 탭 미리보기용 추천만 생성.
+- DB 반영은 apply-best / 웹「DB 적용」시에만 ({SID}_DAILY_PROFIT_* , 마스터 키 미사용).
+"""
+from __future__ import annotations
+
+import logging
+import statistics
+from typing import Any, Dict, List, Optional, Sequence
+
+logger = logging.getLogger("optuna_daily_trail_recommend")
+
+# Optuna strategy 슬러그 → DB 접두
+_STRATEGY_PREFIX: Dict[str, str] = {
+    "tail": "SHORT",
+    "short": "SHORT",
+    "scalp": "SCALP",
+    "scalping": "SCALP",
+    "momentum": "MOMENTUM",
+    "us_momentum": "US_MOMENTUM",
+    "breakout": "BREAKOUT",
+}
+
+
+def strategy_daily_profit_prefix(strategy: str) -> str:
+    s = str(strategy or "").strip().lower()
+    return _STRATEGY_PREFIX.get(s, s.upper() if s else "SHORT")
+
+
+def _env_float(key: str, default: float) -> float:
+    try:
+        from kis_trader.utils.env import get_env_float
+        return float(get_env_float(key, float(default)))
+    except Exception:
+        return float(default)
+
+
+def _env_int(key: str, default: int) -> int:
+    try:
+        from kis_trader.utils.env import get_env_int
+        return int(get_env_int(key, int(default)))
+    except Exception:
+        return int(default)
+
+
+def _env_str(key: str, default: str) -> str:
+    try:
+        from kis_trader.utils.env import get_env_from_db
+        raw = get_env_from_db(key, default)
+        if raw is None or str(raw).strip() in ("", "None"):
+            return str(default)
+        return str(raw).strip()
+    except Exception:
+        return str(default)
+
+
+def _env_bool(key: str, default: bool = True) -> bool:
+    try:
+        from kis_trader.utils.env import get_env_bool
+        return bool(get_env_bool(key, bool(default)))
+    except Exception:
+        return bool(default)
+
+
+def _round_arm_krw(value: float, step: int) -> int:
+    step = max(1000, int(step or 5000))
+    if value <= 0:
+        return 0
+    return int(max(step, round(float(value) / step) * step))
+
+
+def _parse_drops(raw: str) -> List[float]:
+    parts = [p.strip() for p in str(raw or "").split(",") if p.strip()]
+    out: List[float] = []
+    for p in parts[:3]:
+        try:
+            out.append(max(5.0, min(80.0, float(p))))
+        except (TypeError, ValueError):
+            continue
+    while len(out) < 3:
+        out.append([40.0, 30.0, 20.0][len(out)])
+    return out[:3]
+
+
+def _positive_pnls(vals: Sequence[Any]) -> List[float]:
+    out: List[float] = []
+    for v in vals or []:
+        try:
+            x = float(v)
+        except (TypeError, ValueError):
+            continue
+        if x > 0 and abs(x) < 1e15:
+            out.append(x)
+    return out
+
+
+def recommend_daily_trail_tiers(
+    *,
+    top_pnls: Optional[Sequence[Any]] = None,
+    mode_pnl: Optional[float] = None,
+    best_pnl: Optional[float] = None,
+    strategy: str = "",
+) -> Dict[str, Any]:
+    """
+    추천 공식:
+      anchor = max(mode_pnl, median(top_pnls), best_pnl * BEST_FRAC)
+      arm = round(anchor * ARM_FRAC)  (step·min_arm 적용)
+      tiers = arm:d1, (arm*2):d2, (arm*4):d3
+    PnL 전부 ≤0 이면 ok=False (적용 스킵 대상).
+    """
+    tops = _positive_pnls(list(top_pnls or []))
+    med = float(statistics.median(tops)) if tops else 0.0
+    mode_v = float(mode_pnl or 0.0)
+    best_v = float(best_pnl or 0.0)
+    best_frac = _env_float("OPTUNA_DAILY_TRAIL_BEST_FRAC", 0.70)
+    arm_frac = _env_float("OPTUNA_DAILY_TRAIL_ARM_FRAC", 0.60)
+    step = _env_int("OPTUNA_DAILY_TRAIL_ARM_STEP", 5000)
+    min_arm = _env_int("OPTUNA_DAILY_TRAIL_MIN_ARM", 10000)
+    drops = _parse_drops(_env_str("OPTUNA_DAILY_TRAIL_TIER_DROPS", "40,30,20"))
+
+    candidates = [x for x in (mode_v, med, best_v * best_frac) if x > 0]
+    anchor = max(candidates) if candidates else 0.0
+    arm_raw = anchor * arm_frac if anchor > 0 else 0.0
+    arm = _round_arm_krw(arm_raw, step)
+    if arm > 0:
+        arm = max(arm, min_arm)
+        arm = _round_arm_krw(float(arm), step)
+
+    prefix = strategy_daily_profit_prefix(strategy)
+    if arm <= 0:
+        return {
+            "ok": False,
+            "strategy": str(strategy or "").strip().lower(),
+            "prefix": prefix,
+            "reason": "양수 PnL 앵커 없음 — 다단트레일 추천 생략",
+            "anchor_krw": 0,
+            "arm_krw": 0,
+            "tiers": "",
+            "mode": "trailing",
+            "enabled": False,
+            "inputs": {
+                "mode_pnl": mode_v,
+                "best_pnl": best_v,
+                "top_median": med,
+                "top_pnls_head": tops[:10],
+            },
+            "formula": {
+                "arm_frac": arm_frac,
+                "best_frac": best_frac,
+                "step": step,
+                "min_arm": min_arm,
+                "drops": drops,
+            },
+            "note": "apply 시에도 TRAIL 미기록 (수익 앵커 없음)",
+        }
+
+    t1, t2, t3 = int(arm), int(arm * 2), int(arm * 4)
+    d1, d2, d3 = drops
+    tiers = f"{t1}:{d1:.0f},{t2}:{d2:.0f},{t3}:{d3:.0f}"
+    return {
+        "ok": True,
+        "strategy": str(strategy or "").strip().lower(),
+        "prefix": prefix,
+        "reason": "",
+        "anchor_krw": int(round(anchor)),
+        "arm_krw": int(arm),
+        "tiers": tiers,
+        "mode": "trailing",
+        "enabled": True,
+        "inputs": {
+            "mode_pnl": mode_v,
+            "best_pnl": best_v,
+            "top_median": med,
+            "top_pnls_head": tops[:10],
+        },
+        "formula": {
+            "arm_frac": arm_frac,
+            "best_frac": best_frac,
+            "step": step,
+            "min_arm": min_arm,
+            "drops": drops,
+        },
+        "note": (
+            f"apply 시 {prefix}_DAILY_PROFIT_TRAIL_TIERS={tiers} "
+            f"(ENABLED=true, MODE=trailing). 운영 UI에서 끄거나 수정 가능."
+        ),
+    }
+
+
+def recommend_from_optuna_out_data(out_data: Dict[str, Any]) -> Dict[str, Any]:
+    """Optuna 결과 dict(mode_combo·results)에서 추천 생성."""
+    data = out_data or {}
+    strat = str(data.get("strategy") or "").strip().lower()
+    mc = data.get("mode_combo") or {}
+    top_pnls = list(mc.get("top_pnls") or [])
+    bt = mc.get("backtest") or {}
+    mode_pnl = bt.get("total_pnl")
+    vs = mc.get("vs_best") or {}
+    best_pnl = vs.get("best_pnl")
+    if best_pnl is None:
+        res0 = (data.get("results") or [None])[0]
+        if res0:
+            best_pnl = res0.get("total_pnl")
+    if not top_pnls:
+        # gated/학습 Top 에서도 보조
+        for row in (data.get("results_gated") or data.get("results") or [])[:20]:
+            try:
+                top_pnls.append(float(row.get("total_pnl") or 0))
+            except (TypeError, ValueError):
+                pass
+    return recommend_daily_trail_tiers(
+        top_pnls=top_pnls,
+        mode_pnl=float(mode_pnl) if mode_pnl is not None else None,
+        best_pnl=float(best_pnl) if best_pnl is not None else None,
+        strategy=strat,
+    )
+
+
+def attach_daily_trail_recommend(
+    out_data: Dict[str, Any],
+    *,
+    log: Optional[logging.Logger] = None,
+) -> Dict[str, Any]:
+    """out_data 에 daily_trail_recommend 기록 (+ mode_combo 안에도 복사)."""
+    lg = log or logger
+    rec = recommend_from_optuna_out_data(out_data)
+    out_data["daily_trail_recommend"] = rec
+    mc = out_data.get("mode_combo")
+    if isinstance(mc, dict):
+        mc["daily_trail_recommend"] = rec
+    if rec.get("ok"):
+        lg.info(
+            "📅 [다단트레일 추천] %s arm=%s tiers=%s (anchor=%s)",
+            rec.get("prefix"),
+            rec.get("arm_krw"),
+            rec.get("tiers"),
+            rec.get("anchor_krw"),
+        )
+    else:
+        lg.info("📅 [다단트레일 추천] 생략 — %s", rec.get("reason") or "n/a")
+    return out_data
+
+
+def build_daily_trail_env_patch(rec: Dict[str, Any]) -> Dict[str, str]:
+    """전략별 DAILY_PROFIT 패치 (마스터 키 없음)."""
+    if not rec or not rec.get("ok"):
+        return {}
+    prefix = str(rec.get("prefix") or "").strip().upper()
+    tiers = str(rec.get("tiers") or "").strip()
+    if not prefix or not tiers:
+        return {}
+    mode = str(rec.get("mode") or "trailing").strip().lower() or "trailing"
+    return {
+        f"{prefix}_DAILY_PROFIT_TARGET_ENABLED": "true",
+        f"{prefix}_DAILY_PROFIT_MODE": mode,
+        f"{prefix}_DAILY_PROFIT_TRAIL_TIERS": tiers,
+        f"{prefix}_DAILY_PROFIT_TRAIL_ARM_KRW": str(int(rec.get("arm_krw") or 0)),
+    }
+
+
+def apply_daily_trail_recommend_patch(
+    rec: Dict[str, Any],
+    *,
+    log: Optional[logging.Logger] = None,
+) -> Dict[str, Any]:
+    """
+    추천 → DB apply_env_patch.
+    OPTUNA_DAILY_TRAIL_APPLY_ON_BEST=false 이면 스킵.
+    """
+    lg = log or logger
+    if not _env_bool("OPTUNA_DAILY_TRAIL_APPLY_ON_BEST", True):
+        return {"applied": False, "reason": "OPTUNA_DAILY_TRAIL_APPLY_ON_BEST=false"}
+    patch = build_daily_trail_env_patch(rec)
+    if not patch:
+        return {"applied": False, "reason": rec.get("reason") or "추천 없음", "recommend": rec}
+    try:
+        from kis_trader.backtest.param_search_apply_snapshot import apply_env_patch
+        env_id = apply_env_patch(patch)
+    except Exception as exc:
+        lg.warning("⚠️ 다단트레일 추천 DB 반영 실패: %s", exc)
+        return {"applied": False, "error": str(exc), "patch": patch, "recommend": rec}
+    lg.info(
+        "🚀 [Optuna apply] 다단트레일 추천 반영 %s → %s",
+        rec.get("prefix"), patch,
+    )
+    return {"applied": True, "env_id": env_id, "patch": patch, "recommend": rec}
+
+
+def apply_daily_trail_recommend_from_optuna_json(
+    result_json: Optional[str],
+    *,
+    strategy: str = "",
+    log: Optional[logging.Logger] = None,
+) -> Dict[str, Any]:
+    """결과 JSON 경로에서 추천 읽어(또는 재계산) DB 반영."""
+    import json
+    from pathlib import Path
+
+    lg = log or logger
+    if not result_json or not Path(result_json).is_file():
+        return {"applied": False, "reason": "result_json 없음"}
+    try:
+        data = json.loads(Path(result_json).read_text(encoding="utf-8"))
+    except Exception as exc:
+        return {"applied": False, "error": str(exc)}
+    if strategy and not data.get("strategy"):
+        data["strategy"] = strategy
+    rec = data.get("daily_trail_recommend")
+    if not isinstance(rec, dict) or not rec.get("ok"):
+        rec = recommend_from_optuna_out_data(data)
+    return apply_daily_trail_recommend_patch(rec, log=lg)