9ba9ab73b6
- Web UI: - Optuna 탭 추가 및 mode_combo (최빈값 조합), 사후합격 Top 10 시각화 기능 - 파라미터 분포(p25~p75, median, mode) 히스토그램 및 과적합(Overfit) 위험도 진단 UI 신설 - 체크박스 렌더링 깨짐 현상을 네이티브(appearance: auto)로 강제 복구 (CSS) - 다단 트레일링 스탑, 꼬리 진입/돌파 손절 등 고급 조건 설정 폼 UI 고도화 - Backend (Optuna Jobs): - CLI 환경에서 구동된 Optuna json 결과물을 웹 대시보드로 읽어오는 import 기능 강화 - JSON 메타데이터에 sort_by, mode, 호가 적용 여부 등 핵심 파라미터 파싱 누락 수정 - optuna_mode_combo.py 등 최빈값 조합 및 후보군 2차 검증을 위한 신규 모듈 추가 - DB & Execution: - WebSocket 호가/틱 피드 수집 통계(api_feed_collect_stats) 메모리 캐시 최적화 - KIS client 접속 키(approval_key) 등 인프라스트럭처 안정성 및 공유 관리 구조 개선 - 테스트 및 디버깅용 briefing 마크다운 자동 생성 기능 추가
688 lines
28 KiB
Python
688 lines
28 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
optuna_briefing.py — Optuna 결과 AI/규칙 브리핑 (이전 장 · 앞으로 장)
|
||
=====================================================================
|
||
TPE 탐색 JSON을 읽어 한국어 코멘트를 남긴다.
|
||
- 기본: 규칙 기반 분석 (키·외부 API 없어도 동작)
|
||
- PARAM_SEARCH_OPTUNA_BRIEFING_AI=1 이면 AI 보강 시도 순서:
|
||
1) GEMINI_API_KEY (구글) → GEMINI_MODEL_ID (기본 gemini-2.5-flash)
|
||
2) ANTHROPIC_API_KEY → CLAUDE_MODEL_ID
|
||
|
||
구글 키 넣는 방법 (택1):
|
||
A) DB overflow (권장)
|
||
.venv/bin/python - <<'PY'
|
||
from database import TradeDB
|
||
from datetime import datetime
|
||
db = TradeDB()
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
db._persist_env_config_overflow({
|
||
"GEMINI_API_KEY": "여기에_구글_AI_Studio_키",
|
||
"GEMINI_MODEL_ID": "gemini-2.5-flash",
|
||
"PARAM_SEARCH_OPTUNA_BRIEFING_AI": "1",
|
||
}, now)
|
||
from kis_trader.utils.env import invalidate_merged_env_cache
|
||
invalidate_merged_env_cache()
|
||
print("OK")
|
||
PY
|
||
B) 환경변수: export GEMINI_API_KEY=...
|
||
키 발급: https://aistudio.google.com/apikey
|
||
|
||
사용:
|
||
.venv/bin/python -m kis_trader.backtest.optuna_briefing path/to/optuna_*.json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
import os
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from kis_trader.utils.env import get_env_bool, get_env_from_db, get_env_int
|
||
|
||
logger = logging.getLogger("optuna_briefing")
|
||
|
||
_STRAT_KO = {
|
||
"tail": "꼬리",
|
||
"momentum": "모멘텀",
|
||
"breakout": "돌파",
|
||
"scalp": "스캘핑",
|
||
"short": "꼬리",
|
||
}
|
||
|
||
|
||
def _f(x: Any, default: float = 0.0) -> float:
|
||
try:
|
||
return float(x)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _i(x: Any, default: int = 0) -> int:
|
||
try:
|
||
return int(x)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _top_row(data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
for key in ("results_gated", "results", "results_all"):
|
||
rows = data.get(key) or []
|
||
if isinstance(rows, list) and rows:
|
||
return rows[0]
|
||
return None
|
||
|
||
|
||
def _live_pnl_snapshot(strategy: str, start: str, end: str) -> str:
|
||
"""실매 trade_history 요약 (스키마 확인 후). 실패 시 빈 문자열."""
|
||
try:
|
||
from database import TradeDB
|
||
except ImportError:
|
||
return ""
|
||
db = TradeDB()
|
||
try:
|
||
cols = [r["Field"] for r in db.conn.execute("SHOW COLUMNS FROM trade_history").fetchall()]
|
||
colset = set(cols)
|
||
strat_col = "strategy" if "strategy" in colset else None
|
||
pnl_col = "realized_pnl" if "realized_pnl" in colset else (
|
||
"pnl" if "pnl" in colset else None
|
||
)
|
||
time_col = "sell_date" if "sell_date" in colset else (
|
||
"exit_time" if "exit_time" in colset else (
|
||
"buy_date" if "buy_date" in colset else None
|
||
)
|
||
)
|
||
if not strat_col or not pnl_col or not time_col:
|
||
return f"(실매 조회 스킵: trade_history 컬럼 부족)"
|
||
aliases = {
|
||
"tail": ("%tail%", "%short%", "%TAIL%", "%SHORT%"),
|
||
"momentum": ("%momentum%", "%MOMENTUM%"),
|
||
"breakout": ("%breakout%", "%BREAKOUT%"),
|
||
"scalp": ("%scalp%", "%SCALP%", "%reversal%"),
|
||
}
|
||
likes = aliases.get(strategy, (f"%{strategy}%",))
|
||
or_sql = " OR ".join([f"`{strat_col}` LIKE %s"] * len(likes))
|
||
params = list(likes) + [start, end]
|
||
rows = db.conn.execute(
|
||
f"SELECT `{pnl_col}` AS pnl, `{time_col}` AS t "
|
||
f"FROM trade_history "
|
||
f"WHERE ({or_sql}) "
|
||
f"AND `{time_col}` >= %s AND `{time_col}` < DATE_ADD(%s, INTERVAL 1 DAY) "
|
||
f"ORDER BY `{time_col}`",
|
||
tuple(params),
|
||
).fetchall()
|
||
if not rows:
|
||
return f"실매({start}~{end}) 해당 전략 청산 기록 없음(또는 전략명 불일치)."
|
||
pnls = []
|
||
for r in rows:
|
||
if isinstance(r, dict):
|
||
pnls.append(_f(r.get("pnl")))
|
||
else:
|
||
pnls.append(_f(r[0]))
|
||
wins = sum(1 for p in pnls if p > 0)
|
||
total = sum(pnls)
|
||
wr = (wins / len(pnls) * 100.0) if pnls else 0.0
|
||
return (
|
||
f"실매 동기간 청산 {len(pnls)}건 · 승률 {wr:.0f}% · 합산 PnL {total:,.0f}원 "
|
||
f"(백테 Optuna와 숫자 달라도 정상 — 고스트·유니버스·틱 경로 차이)."
|
||
)
|
||
except Exception as exc:
|
||
return f"(실매 요약 실패: {exc})"
|
||
finally:
|
||
try:
|
||
db.conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _params_exit_ob_line(params: Optional[Dict[str, Any]]) -> str:
|
||
"""trial params → 호가·익절·손절 한 줄 (브리핑용)."""
|
||
if not isinstance(params, dict):
|
||
return "—"
|
||
try:
|
||
from kis_trader.backtest.optuna_web_jobs import _ob_whip_ui_from_params
|
||
ui = _ob_whip_ui_from_params(params)
|
||
return str(ui.get("ob_summary") or "—")
|
||
except Exception:
|
||
tp = params.get("tp_pct") or params.get("take_profit_pct")
|
||
sl = params.get("sl_pct") or params.get("stop_loss_pct")
|
||
ob = params.get("ob_filter_enabled")
|
||
bits: List[str] = []
|
||
if ob is True:
|
||
bits.append("호가ON")
|
||
elif ob is False:
|
||
bits.append("호가OFF")
|
||
if tp is not None:
|
||
bits.append(f"익절{float(tp):.1f}%")
|
||
if sl is not None:
|
||
bits.append(f"손절{float(sl):.1f}%")
|
||
return " ".join(bits) if bits else "—"
|
||
|
||
|
||
def _trial_metrics_line(row: Optional[Dict[str, Any]], *, tag: str) -> str:
|
||
if not isinstance(row, dict):
|
||
return f"- **{tag}**: 없음"
|
||
tn = row.get("optuna_trial_number")
|
||
tn_s = f"trial #{tn}" if tn is not None else "trial 없음(조립)"
|
||
pnl = _f(row.get("total_pnl"))
|
||
wr = _f(row.get("win_rate"))
|
||
pf = _f(row.get("pf"))
|
||
nt = _i(row.get("total_trades"))
|
||
stab = row.get("stability_score")
|
||
stab_s = f" · 안정점수 {float(stab):.0f}" if stab is not None else ""
|
||
prox = row.get("consensus_match_pct")
|
||
prox_s = ""
|
||
if prox is not None:
|
||
mn = row.get("consensus_match_n")
|
||
mo = row.get("consensus_match_of")
|
||
prox_s = f" · 근접 {prox}%"
|
||
if mn is not None and mo is not None:
|
||
prox_s += f" ({mn}/{mo}축 밴드안)"
|
||
prm = row.get("params") or row.get("merged_params") or {}
|
||
ob_line = _params_exit_ob_line(prm if isinstance(prm, dict) else {})
|
||
return (
|
||
f"- **{tag}** ({tn_s}): PnL {pnl:,.0f}원 · WR {wr:.1f}% · PF {pf:.2f} · "
|
||
f"거래 {nt}건{stab_s}{prox_s} · {ob_line}"
|
||
)
|
||
|
||
|
||
def build_final_selection_briefing_lines(data: Dict[str, Any]) -> List[str]:
|
||
"""
|
||
mode Top10 / mode_combo / gated / stable 기준 최종 선택 후보 (규칙 기반).
|
||
웹 하단 표와 동일 JSON 소스 — apply source 명시.
|
||
"""
|
||
from kis_trader.backtest.optuna_common import (
|
||
resolve_results_mode_consensus,
|
||
resolve_results_stable,
|
||
)
|
||
from kis_trader.backtest.optuna_mode_combo import (
|
||
resolve_mode_pool_kind,
|
||
select_mode_pool_rows,
|
||
_build_mode_band_profile,
|
||
_row_param_value,
|
||
)
|
||
from kis_trader.backtest.optuna_postprocess_topn import resolve_post_top_n
|
||
|
||
lines: List[str] = []
|
||
lines.append("## 최종 선택 후보 (규칙 · DB 적용 전 확인)")
|
||
lines.append(
|
||
"- 아래는 **역할이 다른** 4종 후보입니다. mode Top10 1위 ≠ 익절 대표 · "
|
||
"**DB 1순위는 사후합격(gated)** 입니다."
|
||
)
|
||
pool_kind = resolve_mode_pool_kind()
|
||
lines.append(f"- mode pool: `{pool_kind}` · scoring: `band_proximity_p25_p75`")
|
||
lines.append("")
|
||
|
||
top_n = resolve_post_top_n(10)
|
||
gated = list(data.get("results_gated") or [])
|
||
learn = list(data.get("results") or data.get("results_all") or [])
|
||
stable, stable_meta = resolve_results_stable(data, top_n=top_n)
|
||
mode_rows, mode_meta = resolve_results_mode_consensus(data, top_n=top_n)
|
||
mc = data.get("mode_combo") if isinstance(data.get("mode_combo"), dict) else {}
|
||
mc_bt = mc.get("backtest") if isinstance(mc.get("backtest"), dict) else {}
|
||
mc_params = mc.get("params") if isinstance(mc.get("params"), dict) else {}
|
||
|
||
g0 = gated[0] if gated else None
|
||
s0 = stable[0] if stable else None
|
||
m0 = mode_rows[0] if mode_rows else None
|
||
|
||
lines.append("### 후보 4종")
|
||
lines.append(_trial_metrics_line(g0, tag="① 사후합격 1위 (apply: gated)"))
|
||
lines.append(_trial_metrics_line(s0, tag="② 안정 1위 (apply: stable)"))
|
||
if mc_params or mc_bt:
|
||
mode_row = {
|
||
"optuna_trial_number": None,
|
||
"total_pnl": mc_bt.get("total_pnl"),
|
||
"win_rate": mc_bt.get("win_rate"),
|
||
"pf": mc_bt.get("pf"),
|
||
"total_trades": mc_bt.get("total_trades"),
|
||
"stability_score": mc_bt.get("stability_score"),
|
||
"params": mc_params,
|
||
}
|
||
freq_tp = (mc.get("freq") or {}).get("tp_pct") if isinstance(mc.get("freq"), dict) else None
|
||
extra = ""
|
||
if isinstance(freq_tp, dict) and freq_tp.get("value") is not None:
|
||
extra = f" · tp 최빈 {freq_tp.get('value')}% ({freq_tp.get('count')}/{freq_tp.get('of')})"
|
||
lines.append(_trial_metrics_line(mode_row, tag="③ mode_combo (apply: mode)") + extra)
|
||
else:
|
||
lines.append("- **③ mode_combo (apply: mode)**: 없음 (미산출·구 JSON)")
|
||
lines.append(_trial_metrics_line(m0, tag="④ mode Top10 1위 (apply: consensus · 밴드 전형 trial)"))
|
||
|
||
# tp 밴드 — mode Top10 1위가 tp 밖인지
|
||
if m0 and mode_meta.get("mode_pool_size"):
|
||
try:
|
||
pool = select_mode_pool_rows(learn, data=data)
|
||
keys: List[str] = list(data.get("grid_keys") or [])
|
||
if not keys and pool:
|
||
keys = list((pool[0].get("params") or {}).keys())
|
||
prof = _build_mode_band_profile(pool, keys)
|
||
tp_band = prof.get("tp_pct") or prof.get("take_profit_pct")
|
||
if tp_band and tp_band.get("kind") == "numeric":
|
||
p25 = float(tp_band["p25"])
|
||
p75 = float(tp_band["p75"])
|
||
prm = m0.get("params") or {}
|
||
tp_v = _row_param_value(m0, "tp_pct") or _row_param_value(m0, "take_profit_pct")
|
||
if tp_v is not None:
|
||
tp_f = float(tp_v)
|
||
in_band = p25 <= tp_f <= p75
|
||
lines.append("")
|
||
lines.append(
|
||
f"- mode pool tp_pct 밴드 p25~p75: **{p25:g}~{p75:g}%** · "
|
||
f"Top10 1위 tp={tp_f:g}% → "
|
||
f"{'밴드 **안**' if in_band else '밴드 **밖**(다른 축 보정으로 근접% 높음)'}"
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
if mode_meta.get("mode_pool_size") is not None:
|
||
lines.append(
|
||
f"- mode Top10 meta: pool {mode_meta.get('mode_pool_size')}건 · "
|
||
f"축 {mode_meta.get('band_axes') or mode_meta.get('mode_params_keys')}개"
|
||
)
|
||
|
||
lines.append("")
|
||
lines.append("### 추천 (자동 · confirm 필수)")
|
||
if g0:
|
||
lines.append(
|
||
"1. **실매 DB 적용 1순위 → ① 사후합격 1위** (`optunaApply gated`) — "
|
||
"WR/PF·min_trades 사후 통과."
|
||
)
|
||
if s0 and s0 is not g0:
|
||
lines.append(
|
||
"2. **변동성·손실일 줄이기 → ② 안정 1위** (`stable`) — "
|
||
"PnL보다 일별 안정 우선 시."
|
||
)
|
||
else:
|
||
lines.append(
|
||
"1. **사후합격 0건 → DB 적용 비권장.** 현행 DB 유지 + 유니버스·틱·기간 재검증."
|
||
)
|
||
if learn:
|
||
lines.append(
|
||
"2. 참고만: 학습 1위는 objective 최대일 뿐 사후 게이트 미통과일 수 있음."
|
||
)
|
||
if mc_params:
|
||
lines.append(
|
||
f"{'3' if g0 else '2'}. **2차 narrow·대표 숫자 → ③ mode_combo** — "
|
||
"축별 최빈 조립 · trial 번호 없음 · 익절은 pool 최빈값 참고."
|
||
)
|
||
if m0:
|
||
n = "4" if (g0 and mc_params) else ("3" if (g0 or mc_params) else "2")
|
||
lines.append(
|
||
f"{n}. **④ mode Top10 1위** — 양수 pool 전형 trial(다축 밴드 근접). "
|
||
"**익절 하나로 쓰지 말 것** · 표에서 「보기」→ consensus apply."
|
||
)
|
||
|
||
diag = data.get("overfit_diagnostics") if isinstance(data.get("overfit_diagnostics"), dict) else {}
|
||
risk = diag.get("overfit_risk_pct")
|
||
if risk is not None and float(risk) >= 55 and g0:
|
||
lines.append(
|
||
f"- ⚠ 과적합 위험 {risk}% — gated 적용 전 **웹백테 동일 기간 1회**·소액 관찰 권장."
|
||
)
|
||
if stable_meta.get("fallback_rank_only"):
|
||
lines.append(
|
||
f"- ⚠ 안정 게이트 0건 → 안정 Top은 **점수순 폴백** ({stable_meta.get('fallback_note', '')})"
|
||
)
|
||
lines.append("")
|
||
return lines
|
||
|
||
|
||
def build_rule_briefing(data: Dict[str, Any]) -> str:
|
||
"""규칙 기반 — 이전 장 / 앞으로 장 코멘트."""
|
||
strategy = str(data.get("strategy") or "tail").strip().lower()
|
||
if strategy in ("", "none"):
|
||
strategy = "tail"
|
||
name = _STRAT_KO.get(strategy, strategy)
|
||
start = str(data.get("start") or "?")
|
||
end = str(data.get("end") or "?")
|
||
days = _i(data.get("backtest_days"), 0)
|
||
mode = str(data.get("mode") or "")
|
||
n_req = _i(data.get("optuna_n_trials_requested"), 0)
|
||
n_done = _i(data.get("optuna_trials_completed"), 0)
|
||
n_all = _i(data.get("n_results_all"), len(data.get("results_all") or []))
|
||
n_gated = _i(data.get("n_results_gated"), len(data.get("results_gated") or []))
|
||
search_wr = _f(data.get("min_win_rate"))
|
||
search_pf = _f(data.get("min_pf"))
|
||
search_mt = _i(data.get("min_trades"), 1)
|
||
rg = data.get("report_gates") or {}
|
||
rep_wr = _f(rg.get("min_win_rate"), 40.0)
|
||
rep_pf = _f(rg.get("min_pf"), 1.0)
|
||
rep_mt = _i(rg.get("min_trades"), search_mt)
|
||
|
||
best_learn = (data.get("results") or data.get("results_all") or [None])[0]
|
||
best_gate = (data.get("results_gated") or [None])[0]
|
||
best = best_gate or best_learn
|
||
|
||
lines: List[str] = []
|
||
lines.append(f"# Optuna 브리핑 — {name} ({mode})")
|
||
lines.append(f"기간: {start} ~ {end} (거래일≈{days or '?'}일) · trials {n_done}/{n_req}")
|
||
lines.append(
|
||
f"탐색 게이트: WR≥{search_wr:g}% PF≥{search_pf:g} · min_trades≥{search_mt} "
|
||
f"| 후보 게이트: WR≥{rep_wr:g}% PF≥{rep_pf:g} · min_trades≥{rep_mt}"
|
||
)
|
||
lines.append(
|
||
f"- min_trades={search_mt}: **합격 인정 최소 거래 건수**(표본 부족·우연 승률 방지). "
|
||
f"이보다 적은 조합은 학습/후보 집계에서 제외됩니다."
|
||
)
|
||
lines.append(f"학습용 통과 {n_all}건 · 사후합격(results_gated) {n_gated}건")
|
||
lines.append("")
|
||
|
||
# 과적합·적용가능도 (휴리스틱) + 임계값 분포 요약
|
||
diag = data.get("overfit_diagnostics")
|
||
if not isinstance(diag, dict) or diag.get("overfit_risk_pct") is None:
|
||
try:
|
||
from kis_trader.backtest.optuna_common import build_optuna_overfit_diagnostics
|
||
diag = build_optuna_overfit_diagnostics(data)
|
||
data["overfit_diagnostics"] = diag
|
||
except Exception:
|
||
diag = {}
|
||
if isinstance(diag, dict) and diag.get("overfit_risk_pct") is not None:
|
||
lines.append("## 과적합·적용 가능도 (휴리스틱)")
|
||
lines.append(
|
||
f"- **과적합 위험 {diag.get('overfit_risk_pct')}%** · "
|
||
f"**적용 가능도 {diag.get('apply_readiness_pct')}%** · "
|
||
f"판정: **{diag.get('verdict')}** — {diag.get('verdict_ko')}"
|
||
)
|
||
for fac in (diag.get("factors") or [])[:6]:
|
||
if not isinstance(fac, dict):
|
||
continue
|
||
lines.append(
|
||
f" - {fac.get('label')}: +{fac.get('points')}p — {fac.get('detail')}"
|
||
)
|
||
dist = diag.get("threshold_distribution") or []
|
||
if dist:
|
||
lines.append(
|
||
f"- 임계값 분포 표본: {diag.get('threshold_pool')} "
|
||
f"상위 {diag.get('threshold_pool_n')}건"
|
||
)
|
||
lines.append(
|
||
"| 파라미터 | median | p25~p75 | mode(비율) |"
|
||
)
|
||
lines.append("|---|---:|---|---|")
|
||
for row in dist[:12]:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
med = row.get("median")
|
||
p25, p75 = row.get("p25"), row.get("p75")
|
||
band = (
|
||
f"{p25} ~ {p75}" if p25 is not None and p75 is not None else "—"
|
||
)
|
||
med_s = f"{med}" if med is not None else "—"
|
||
share = row.get("mode_share")
|
||
share_s = f"{float(share)*100:.0f}%" if share is not None else "?"
|
||
lines.append(
|
||
f"| `{row.get('param')}` | {med_s} | {band} | "
|
||
f"{row.get('mode')} ({share_s}) |"
|
||
)
|
||
if diag.get("note"):
|
||
lines.append(f"- _참고: {diag.get('note')}_")
|
||
lines.append("")
|
||
|
||
lines.extend(build_final_selection_briefing_lines(data))
|
||
|
||
lines.append("## 이전 장에서는")
|
||
live = _live_pnl_snapshot(strategy, start, end)
|
||
if live:
|
||
lines.append(f"- {live}")
|
||
if not best:
|
||
lines.append(
|
||
"- 이 기간 백테에서는 **유의미한 통과 조합이 거의 없었습니다.** "
|
||
"유니버스·틱·손절·시간창을 먼저 점검하세요 (돌파 0건이면 특히)."
|
||
)
|
||
else:
|
||
pnl = _f(best.get("total_pnl"))
|
||
wr = _f(best.get("win_rate"))
|
||
pf = _f(best.get("pf"))
|
||
nt = _i(best.get("total_trades"))
|
||
tn = best.get("optuna_trial_number")
|
||
tag = "사후합격 1위" if best_gate else "학습 1위(사후게이트 미달 가능)"
|
||
lines.append(
|
||
f"- {tag}: trial #{tn} · 거래 {nt}건 · 승률 {wr:.1f}% · PF {pf:.2f} · PnL {pnl:,.0f}원"
|
||
)
|
||
if wr >= 95.0 and nt < 15:
|
||
lines.append(
|
||
"- 승률이 비정상적으로 높고 표본이 작습니다 → **해당 장에만 맞는 과적합** 신호."
|
||
)
|
||
if days and days <= 2:
|
||
lines.append(
|
||
f"- 표본 장이 {days}일뿐이라, ‘이전 장에 잘 먹힌 숫자’일 뿐 **추세가 바뀌면 바로 깨질 수 있습니다.**"
|
||
)
|
||
if n_gated == 0 and n_all > 0:
|
||
lines.append(
|
||
"- 플러스/학습 후보는 있으나 승률·PF 사후 기준은 못 넘김 → **당장 DB 적용은 비권장.**"
|
||
)
|
||
if strategy == "breakout" and n_all == 0:
|
||
lines.append(
|
||
"- 돌파는 실매 고스트퍼지·숄더 청산과 백테 풀경로가 어긋나기 쉽습니다. "
|
||
"‘손실이 지워진 실매’와 ‘손절까지 포함한 백테’를 같은 숫자로 보지 마세요."
|
||
)
|
||
|
||
lines.append("")
|
||
lines.append("## 앞으로 장에서는")
|
||
lines.append(
|
||
"- **1~2일 Optuna best를 바로 실매에 넣지 마세요.** 다일(≥5거래일) 재검증·웹백테 1회 후 검토."
|
||
)
|
||
lines.append(
|
||
"- 적용 시에는 `results_gated` 후보만. 탐색용 best(objective)와 다를 수 있습니다."
|
||
)
|
||
lines.append(
|
||
"- 장이 좁아지거나 변동성이 커지면: 손절·트레일·시간창이 먼저 깨집니다. "
|
||
"승률 자랑보다 **손절이 살아 있는지**를 보세요."
|
||
)
|
||
if best_gate is None:
|
||
lines.append(
|
||
"- 현재 사후합격 0건 → 파라미터 유지(현행 DB) + 유니버스/데이터 점검이 우선입니다."
|
||
)
|
||
else:
|
||
lines.append(
|
||
"- 사후합격이 있어도 ‘내일도 같은 장’이 아닙니다. "
|
||
"페이퍼/소액·모의로 1~2일 관찰 후 확대하세요."
|
||
)
|
||
lines.append("")
|
||
lines.append(f"_생성: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (rule)_")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _briefing_prompt(rule_text: str, data: Dict[str, Any]) -> str:
|
||
compact = {
|
||
"strategy": data.get("strategy"),
|
||
"mode": data.get("mode"),
|
||
"start": data.get("start"),
|
||
"end": data.get("end"),
|
||
"backtest_days": data.get("backtest_days"),
|
||
"min_win_rate": data.get("min_win_rate"),
|
||
"min_pf": data.get("min_pf"),
|
||
"report_gates": data.get("report_gates"),
|
||
"n_results_all": data.get("n_results_all"),
|
||
"n_results_gated": data.get("n_results_gated"),
|
||
"optuna_best_trial_number": data.get("optuna_best_trial_number"),
|
||
"optuna_best_value": data.get("optuna_best_value"),
|
||
"top_gated": (data.get("results_gated") or [None])[0],
|
||
"top_learning": (data.get("results") or [None])[0],
|
||
"top_stable": (data.get("results_stable") or [None])[0],
|
||
"top_mode_consensus": (data.get("results_mode") or [None])[0],
|
||
"mode_combo_params": (data.get("mode_combo") or {}).get("params"),
|
||
"mode_combo_vs_best": (data.get("mode_combo") or {}).get("vs_best"),
|
||
"mode_consensus_meta": data.get("mode_consensus_meta"),
|
||
}
|
||
return (
|
||
"당신은 한국 주식 퀀트 헤지펀드 리스크 매니저입니다. "
|
||
"아래 Optuna TPE 결과와 규칙 브리핑을 읽고, 초보자도 이해하게 "
|
||
"「이전 장에서는」/「앞으로 장에서는」 두 절로만 한국어 코멘트를 쓰세요. "
|
||
"과적합·표본부족·실매↔백테 괴리(고스트퍼지 등)를 분명히 경고하세요. "
|
||
"mode Top10 1위를 익절 대표로 단정하지 마세요 — DB 1순위는 사후합격(gated). "
|
||
"특정 종목 매수 추천·확정 수익 약속 금지. 200~400자.\n\n"
|
||
f"[규칙 브리핑]\n{rule_text}\n\n"
|
||
f"[요약 JSON]\n{json.dumps(compact, ensure_ascii=False, default=str)[:6000]}"
|
||
)
|
||
|
||
|
||
def _call_gemini_briefing(rule_text: str, data: Dict[str, Any]) -> Optional[str]:
|
||
"""GEMINI_API_KEY(구글) 있으면 코멘트. 실패 시 None."""
|
||
api_key = (
|
||
str(get_env_from_db("GEMINI_API_KEY", "") or "").strip()
|
||
or str(os.environ.get("GEMINI_API_KEY", "") or "").strip()
|
||
or str(os.environ.get("GOOGLE_API_KEY", "") or "").strip()
|
||
)
|
||
if not api_key:
|
||
return None
|
||
try:
|
||
import google.genai as genai
|
||
except ImportError:
|
||
logger.warning("google-genai 미설치 — pip install google-genai")
|
||
return None
|
||
prompt = _briefing_prompt(rule_text, data)
|
||
try:
|
||
model = (
|
||
str(get_env_from_db("GEMINI_MODEL_ID", "gemini-2.5-flash") or "").strip()
|
||
or "gemini-2.5-flash"
|
||
)
|
||
client = genai.Client(api_key=api_key)
|
||
response = client.models.generate_content(model=model, contents=prompt)
|
||
text = getattr(response, "text", None) or ""
|
||
if not text and getattr(response, "candidates", None):
|
||
try:
|
||
text = response.candidates[0].content.parts[0].text
|
||
except Exception:
|
||
text = ""
|
||
return (text or "").strip() or None
|
||
except Exception as exc:
|
||
logger.warning("Gemini 브리핑 실패: %s", exc)
|
||
return None
|
||
|
||
|
||
def _call_claude_briefing(rule_text: str, data: Dict[str, Any]) -> Optional[str]:
|
||
"""ANTHROPIC_API_KEY 있으면 코멘트. 실패 시 None."""
|
||
api_key = (
|
||
str(get_env_from_db("ANTHROPIC_API_KEY", "") or "").strip()
|
||
or str(os.environ.get("ANTHROPIC_API_KEY", "") or "").strip()
|
||
)
|
||
if not api_key:
|
||
return None
|
||
try:
|
||
import anthropic
|
||
except ImportError:
|
||
logger.warning("anthropic 미설치 — Claude 스킵")
|
||
return None
|
||
prompt = _briefing_prompt(rule_text, data)
|
||
try:
|
||
model = str(get_env_from_db("CLAUDE_MODEL_ID", "claude-sonnet-4-5") or "").strip() or "claude-sonnet-4-5"
|
||
max_tok = int(get_env_int("CLAUDE_MAX_TOKENS", 1024))
|
||
client = anthropic.Anthropic(api_key=api_key)
|
||
resp = client.messages.create(
|
||
model=model,
|
||
max_tokens=min(2048, max(256, max_tok)),
|
||
messages=[{"role": "user", "content": prompt}],
|
||
)
|
||
text = resp.content[0].text if resp.content else ""
|
||
return (text or "").strip() or None
|
||
except Exception as exc:
|
||
logger.warning("Claude 브리핑 실패: %s", exc)
|
||
return None
|
||
|
||
|
||
def _call_ai_briefing(rule_text: str, data: Dict[str, Any]) -> Tuple[Optional[str], str]:
|
||
"""AI 보강. 반환: (텍스트, provider명). Gemini → Claude."""
|
||
if not get_env_bool("PARAM_SEARCH_OPTUNA_BRIEFING_AI", True):
|
||
return None, "BRIEFING_AI=off"
|
||
gem = _call_gemini_briefing(rule_text, data)
|
||
if gem:
|
||
return gem, "gemini"
|
||
claude = _call_claude_briefing(rule_text, data)
|
||
if claude:
|
||
return claude, "claude"
|
||
has_g = bool(
|
||
str(get_env_from_db("GEMINI_API_KEY", "") or "").strip()
|
||
or str(os.environ.get("GEMINI_API_KEY", "") or "").strip()
|
||
)
|
||
has_c = bool(
|
||
str(get_env_from_db("ANTHROPIC_API_KEY", "") or "").strip()
|
||
or str(os.environ.get("ANTHROPIC_API_KEY", "") or "").strip()
|
||
)
|
||
if not has_g and not has_c:
|
||
return None, "no_api_key(GEMINI_API_KEY 또는 ANTHROPIC_API_KEY)"
|
||
return None, "api_call_failed"
|
||
|
||
|
||
def write_briefing_for_json(
|
||
json_path: str,
|
||
*,
|
||
log: Optional[logging.Logger] = None,
|
||
) -> Optional[str]:
|
||
"""JSON 옆 .briefing.md 저장 + 로그/프린트. 경로 반환."""
|
||
lg = log or logger
|
||
abs_path = os.path.abspath(json_path)
|
||
if not os.path.isfile(abs_path):
|
||
lg.warning("⚠️ 브리핑 스킵 — JSON 없음: %s", abs_path)
|
||
return None
|
||
try:
|
||
with open(abs_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
except (OSError, json.JSONDecodeError) as exc:
|
||
lg.warning("⚠️ 브리핑 JSON 읽기 실패: %s", abs_path)
|
||
return None
|
||
|
||
try:
|
||
from kis_trader.backtest.optuna_common import attach_optuna_overfit_diagnostics
|
||
attach_optuna_overfit_diagnostics(data)
|
||
except Exception:
|
||
pass
|
||
|
||
rule = build_rule_briefing(data)
|
||
ai, provider = _call_ai_briefing(rule, data)
|
||
if ai:
|
||
body = (
|
||
rule
|
||
+ f"\n\n## AI 코멘트 (이전 장 / 앞으로 장) [{provider}]\n\n"
|
||
+ ai
|
||
+ "\n"
|
||
)
|
||
else:
|
||
body = (
|
||
rule
|
||
+ f"\n\n## AI 코멘트\n\n_(미생성: {provider}. "
|
||
"구글 키는 GEMINI_API_KEY 를 DB/환경에 넣으세요 — "
|
||
"kis_trader/backtest/optuna_briefing.py 모듈 docstring 참고)_\n"
|
||
)
|
||
lg.info("ℹ️ AI 브리핑 없음: %s", provider)
|
||
|
||
out_md = abs_path.rsplit(".", 1)[0] + ".briefing.md"
|
||
try:
|
||
with open(out_md, "w", encoding="utf-8") as f:
|
||
f.write(body)
|
||
except OSError as exc:
|
||
lg.warning("⚠️ 브리핑 저장 실패: %s", exc)
|
||
return None
|
||
|
||
try:
|
||
data["briefing_md_path"] = out_md
|
||
data["briefing_text"] = body
|
||
data["briefing_ai_provider"] = provider if ai else None
|
||
with open(abs_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||
except OSError:
|
||
pass
|
||
|
||
lg.info("🗣️ Optuna 브리핑: %s", out_md)
|
||
print(f"OPTUNA_BRIEFING_MD={out_md}", flush=True)
|
||
preview = (ai or rule).strip().splitlines()
|
||
for line in preview[:12]:
|
||
print(line, flush=True)
|
||
if len(preview) > 12:
|
||
print("… (전체는 briefing.md)", flush=True)
|
||
return out_md
|
||
|
||
|
||
def main() -> None:
|
||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||
p = argparse.ArgumentParser(description="Optuna 결과 브리핑")
|
||
p.add_argument("json_path", help="optuna_*.json 경로")
|
||
args = p.parse_args()
|
||
path = write_briefing_for_json(args.json_path)
|
||
if not path:
|
||
raise SystemExit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|