316 lines
10 KiB
Python
316 lines
10 KiB
Python
#!/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)
|