420 lines
14 KiB
Python
420 lines
14 KiB
Python
"""
|
|
kis_trader/backtest/optuna_whipsaw_recommend.py
|
|
=================================================
|
|
Optuna 차트 캔들 최적화 완료 후, 후처리로 고속 휩쏘 파라미터 탐색을 수행하여
|
|
전략별 최적의 휩쏘 필터 수치(Consensus)를 도출하고
|
|
Optuna out_data 및 Apply 패치에 자동으로 결합하는 모듈입니다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import optuna
|
|
|
|
from database import TradeDB
|
|
from kis_trader.engine.whipsaw_filter import whipsaw_reject_for_signal
|
|
|
|
logger = logging.getLogger("OptunaWhipsawRecommend")
|
|
optuna.logging.set_verbosity(optuna.logging.WARNING)
|
|
|
|
|
|
@dataclass
|
|
class TradeInfo:
|
|
code: str
|
|
name: str
|
|
buy_dt: datetime
|
|
buy_price: float
|
|
actual_pnl: float
|
|
actual_profit_rate: float
|
|
ticks: List[Dict[str, Any]]
|
|
|
|
|
|
def _krw_int(v: Any) -> int:
|
|
try:
|
|
x = float(v)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
if x != x or abs(x) >= 1e15:
|
|
return 0
|
|
return int(x)
|
|
|
|
|
|
def _whipsaw_n_trials(default: int = 500) -> int:
|
|
from kis_trader.utils.env import get_env_int
|
|
return max(10, int(get_env_int("OPTUNA_WHIPSAW_RECOMMEND_TRIALS", int(default))))
|
|
|
|
|
|
def _whipsaw_tick_lookback_sec(default: int = 180) -> int:
|
|
from kis_trader.utils.env import get_env_int
|
|
return max(30, int(get_env_int("OPTUNA_WHIPSAW_TICK_LOOKBACK_SEC", int(default))))
|
|
|
|
|
|
def _parse_buy_dt(raw_dt: Any) -> datetime:
|
|
if isinstance(raw_dt, datetime):
|
|
return raw_dt
|
|
s = str(raw_dt or "").strip()
|
|
try:
|
|
return datetime.strptime(s[:19].replace("T", " "), "%Y-%m-%d %H:%M:%S")
|
|
except ValueError:
|
|
pass
|
|
digits = "".join(ch for ch in s if ch.isdigit())
|
|
if len(digits) >= 14:
|
|
return datetime.strptime(digits[:14], "%Y%m%d%H%M%S")
|
|
if len(digits) >= 12:
|
|
return datetime.strptime(digits[:12], "%Y%m%d%H%M")
|
|
return datetime.fromisoformat(s)
|
|
|
|
|
|
def _ticks_for_buy(db: Any, code: str, buy_dt: datetime, *, market: str = "KR") -> List[Dict[str, Any]]:
|
|
look = _whipsaw_tick_lookback_sec(180)
|
|
start_tick_time = (buy_dt - timedelta(seconds=look)).strftime("%Y%m%d%H%M%S")
|
|
end_tick_time = buy_dt.strftime("%Y%m%d%H%M%S")
|
|
mk = str(market or "KR").strip().upper() or "KR"
|
|
return db.get_ws_ticks(code, market=mk, start_tick_time=start_tick_time, end_tick_time=end_tick_time) or []
|
|
|
|
|
|
def raw_fills_to_whipsaw_trades(
|
|
raw_fills: List[Dict[str, Any]],
|
|
*,
|
|
db: Any,
|
|
market: str = "KR",
|
|
) -> List[TradeInfo]:
|
|
trades: List[TradeInfo] = []
|
|
for b in raw_fills or []:
|
|
if not isinstance(b, dict):
|
|
continue
|
|
code = str(b.get("code") or "").strip()
|
|
raw_dt = b.get("buy_date") or b.get("buy_time") or b.get("entry_time")
|
|
try:
|
|
buy_dt = _parse_buy_dt(raw_dt)
|
|
except Exception:
|
|
continue
|
|
ticks = _ticks_for_buy(db, code, buy_dt, market=market)
|
|
if not ticks:
|
|
continue
|
|
pnl = float(
|
|
b.get("actual_pnl") if b.get("actual_pnl") is not None else (b.get("pnl") or b.get("realized_pnl") or 0)
|
|
)
|
|
trades.append(
|
|
TradeInfo(
|
|
code=code,
|
|
name=str(b.get("name") or code),
|
|
buy_dt=buy_dt,
|
|
buy_price=float(b.get("buy_price") or b.get("entry") or 0),
|
|
actual_pnl=pnl,
|
|
actual_profit_rate=float(b.get("profit_rate") or b.get("actual_profit_rate") or 0),
|
|
ticks=ticks,
|
|
)
|
|
)
|
|
return trades
|
|
|
|
|
|
def recommend_whipsaw_parameters(
|
|
*,
|
|
strategy: str = "MOMENTUM",
|
|
n_trials: int = 0,
|
|
days: int = 0,
|
|
log: Optional[logging.Logger] = None,
|
|
raw_fills: Optional[List[Dict[str, Any]]] = None,
|
|
date_from: Optional[str] = None,
|
|
date_to: Optional[str] = None,
|
|
market: str = "KR",
|
|
progress_label: str = "",
|
|
) -> Dict[str, Any]:
|
|
lg = log or logger
|
|
if int(n_trials or 0) <= 0:
|
|
n_trials = _whipsaw_n_trials(500)
|
|
if int(days or 0) <= 0:
|
|
from kis_trader.utils.env import get_env_int
|
|
days = max(1, int(get_env_int("OPTUNA_WHIPSAW_LOOKBACK_DAYS", 7)))
|
|
strat_upper = str(strategy).strip().upper()
|
|
mk = "US" if "US" in strat_upper else str(market or "KR").strip().upper() or "KR"
|
|
axis_name = str(progress_label or f"휩쏘-{strat_upper}").strip() or "휩쏘"
|
|
|
|
db = TradeDB()
|
|
try:
|
|
if raw_fills is not None:
|
|
trades = raw_fills_to_whipsaw_trades(list(raw_fills), db=db, market=mk)
|
|
else:
|
|
now = datetime.now()
|
|
if date_from and date_to:
|
|
start_date = str(date_from)[:10]
|
|
end_date = str(date_to)[:10]
|
|
buys = db.conn.execute(
|
|
"""
|
|
SELECT id, code, name, buy_date, buy_price, realized_pnl, profit_rate
|
|
FROM trade_history
|
|
WHERE strategy=%s AND DATE(buy_date) >= %s AND DATE(buy_date) <= %s
|
|
ORDER BY buy_date
|
|
""",
|
|
(strat_upper, start_date, end_date),
|
|
).fetchall()
|
|
else:
|
|
start_date = (now - timedelta(days=days)).strftime("%Y-%m-%d")
|
|
buys = db.conn.execute(
|
|
"""
|
|
SELECT id, code, name, buy_date, buy_price, realized_pnl, profit_rate
|
|
FROM trade_history
|
|
WHERE strategy=%s AND DATE(buy_date) >= %s
|
|
ORDER BY buy_date
|
|
""",
|
|
(strat_upper, start_date),
|
|
).fetchall()
|
|
|
|
trades = []
|
|
for b in buys:
|
|
try:
|
|
buy_dt = _parse_buy_dt(b["buy_date"])
|
|
except Exception:
|
|
continue
|
|
ticks = _ticks_for_buy(db, b["code"], buy_dt, market=mk)
|
|
if not ticks:
|
|
continue
|
|
trades.append(
|
|
TradeInfo(
|
|
code=b["code"],
|
|
name=str(b.get("name") or b["code"]),
|
|
buy_dt=buy_dt,
|
|
buy_price=float(b["buy_price"] or 0),
|
|
actual_pnl=float(b["realized_pnl"] or 0),
|
|
actual_profit_rate=float(b["profit_rate"] or 0),
|
|
ticks=ticks,
|
|
)
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
if len(trades) < 3:
|
|
lg.warning("⚠️ [%s] 휩쏘 연산 가능한 실제 틱 보유 매수 건수(%s건)가 부족하여 최적화 생략.", strat_upper, len(trades))
|
|
return {"ok": False, "reason": "not_enough_trades", "trade_count": len(trades)}
|
|
|
|
orig_cnt = len(trades)
|
|
orig_win = sum(1 for t in trades if t.actual_pnl > 0) / orig_cnt * 100.0
|
|
orig_pnl = sum(t.actual_pnl for t in trades)
|
|
orig_rate = sum(t.actual_profit_rate for t in trades) / orig_cnt
|
|
|
|
def _sim_trade(tr: TradeInfo, p: Dict[str, Any]) -> Tuple[float, float, str]:
|
|
params_for_eval = {
|
|
"whipsaw_filter_enabled": True,
|
|
"whipsaw_subbar_sec": p["subbar_sec"],
|
|
"whipsaw_lookback_sec": p["lookback_sec"],
|
|
"whipsaw_dip_pct": p["dip_pct"],
|
|
"whipsaw_recovery_tol_pct": p.get("recov_pct", 0.0),
|
|
}
|
|
|
|
sig_bar = {"low": tr.buy_price, "dt": tr.buy_dt}
|
|
reject_reason, _ = whipsaw_reject_for_signal(
|
|
params=params_for_eval,
|
|
strategy=strat_upper,
|
|
signal_bar=sig_bar,
|
|
current_price=tr.buy_price,
|
|
ticks=tr.ticks
|
|
)
|
|
|
|
if reject_reason:
|
|
return (0.0, 0.0, "ENTRY_REJECTED")
|
|
|
|
return (tr.actual_pnl, tr.actual_profit_rate, "ORIGINAL")
|
|
|
|
def _calc_suite(p: Dict[str, Any]) -> Tuple[int, float, float, float]:
|
|
t_cnt = 0
|
|
w_cnt = 0
|
|
tot_pnl = 0.0
|
|
tot_rate = 0.0
|
|
for t in trades:
|
|
pnl, rate, rtype = _sim_trade(t, p)
|
|
if rtype != "ENTRY_REJECTED":
|
|
t_cnt += 1
|
|
tot_pnl += pnl
|
|
tot_rate += rate
|
|
if pnl > 0:
|
|
w_cnt += 1
|
|
w_rate = (w_cnt / t_cnt * 100.0) if t_cnt > 0 else 0.0
|
|
avg_r = (tot_rate / t_cnt) if t_cnt > 0 else 0.0
|
|
return t_cnt, w_rate, tot_pnl, avg_r
|
|
|
|
valid_records: List[Dict[str, Any]] = []
|
|
|
|
def obj_func(trial: optuna.Trial) -> float:
|
|
params = {
|
|
"subbar_sec": trial.suggest_categorical("subbar_sec", [10, 15, 20, 30, 45, 60]),
|
|
"lookback_sec": trial.suggest_categorical("lookback_sec", [30, 45, 60, 90, 120, 180]),
|
|
"dip_pct": trial.suggest_float("dip_pct", 0.001, 0.010, step=0.001),
|
|
}
|
|
|
|
cnt, win_r, pnl, rate = _calc_suite(params)
|
|
if cnt < max(3, int(orig_cnt * 0.3)):
|
|
return -999999999.0
|
|
|
|
w_p = (pnl / 100000.0)
|
|
w_w = win_r * 2.0
|
|
score = w_p + w_w
|
|
if win_r >= 60.0:
|
|
score += (win_r - 60.0) * 1.5
|
|
|
|
valid_records.append({"score": score, "pnl": pnl, "win_rate": win_r, "count": cnt, "rate": rate, "params": params})
|
|
return score
|
|
|
|
study = optuna.create_study(direction="maximize")
|
|
from kis_trader.backtest import optuna_post_progress as opp
|
|
|
|
def _cb(_study: Any, _trial: Any) -> None:
|
|
n = len(_study.trials)
|
|
if n == 1 or n == int(n_trials) or n % 5 == 0:
|
|
try:
|
|
opp.on_ob_axis_trial(n, int(n_trials), axis_name)
|
|
except Exception:
|
|
pass
|
|
if n == 1 or n == int(n_trials) or n % 20 == 0:
|
|
lg.info("📡 [%s] trial %d/%d", axis_name, n, int(n_trials))
|
|
|
|
study.optimize(obj_func, n_trials=n_trials, callbacks=[_cb])
|
|
|
|
valid_records.sort(key=lambda x: x["score"], reverse=True)
|
|
top5 = valid_records[: min(5, len(valid_records))]
|
|
if not top5:
|
|
return {"ok": False, "reason": "no_valid_trials"}
|
|
|
|
# Consensus 도출
|
|
best = top5[0]
|
|
avg_subbar = int(sum(r["params"]["subbar_sec"] for r in top5) / len(top5))
|
|
avg_lookback = int(sum(r["params"]["lookback_sec"] for r in top5) / len(top5))
|
|
avg_dip = round(sum(r["params"]["dip_pct"] for r in top5) / len(top5), 4)
|
|
|
|
cons_params = {
|
|
"subbar_sec": avg_subbar,
|
|
"lookback_sec": avg_lookback,
|
|
"dip_pct": avg_dip,
|
|
}
|
|
c_cnt, c_win, c_pnl, c_rate = _calc_suite(cons_params)
|
|
|
|
lg.info(
|
|
"⚡ [휩쏘 필터 합의 추천] 전략=%s (모수=%d건, %d회 탐색) | subbar=%d lookback=%d dip=%.3f | 승률: %.1f%% 손익: %.0f원",
|
|
strat_upper,
|
|
len(trades),
|
|
n_trials,
|
|
avg_subbar,
|
|
avg_lookback,
|
|
avg_dip,
|
|
c_win,
|
|
c_pnl,
|
|
)
|
|
|
|
return {
|
|
"ok": True,
|
|
"strategy": strat_upper,
|
|
"n_trials": n_trials,
|
|
"trade_count": len(trades),
|
|
"orig_stats": {
|
|
"count": orig_cnt,
|
|
"win_rate": round(orig_win, 1),
|
|
"pnl": _krw_int(orig_pnl),
|
|
"avg_rate": round(orig_rate, 2),
|
|
},
|
|
"recommended_stats": {
|
|
"count": c_cnt,
|
|
"win_rate": round(c_win, 1),
|
|
"pnl": _krw_int(c_pnl),
|
|
"avg_rate": round(c_rate, 2),
|
|
"pnl_diff": _krw_int(c_pnl - orig_pnl),
|
|
},
|
|
"params": {
|
|
"whipsaw_filter_enabled": True,
|
|
"whipsaw_subbar_sec": avg_subbar,
|
|
"whipsaw_lookback_sec": avg_lookback,
|
|
"whipsaw_dip_pct": avg_dip,
|
|
},
|
|
}
|
|
|
|
|
|
def attach_whipsaw_recommend(
|
|
out_data: Dict[str, Any],
|
|
*,
|
|
log: Optional[logging.Logger] = None,
|
|
) -> Dict[str, Any]:
|
|
"""out_data에 휩쏘 필터 추천 결과를 첨부."""
|
|
lg = log or logger
|
|
strat = str(out_data.get("strategy") or "MOMENTUM").strip().upper()
|
|
rec = recommend_whipsaw_parameters(
|
|
strategy=strat,
|
|
n_trials=0,
|
|
log=lg,
|
|
date_from=str(out_data.get("start") or "") or None,
|
|
date_to=str(out_data.get("end") or "") or None,
|
|
market="US" if "US" in strat else "KR",
|
|
)
|
|
out_data["whipsaw_recommend"] = rec
|
|
mc = out_data.get("mode_combo")
|
|
if isinstance(mc, dict):
|
|
mc["whipsaw_recommend"] = rec
|
|
if not rec.get("ok"):
|
|
lg.info("⚡ [휩쏘 필터 합의 추천] 생략 — %s", rec.get("reason") or "n/a")
|
|
return out_data
|
|
|
|
|
|
_WHIPSAW_DB_SKIP_STRATS = {"TAIL", "SHORT", "BREAKOUT"}
|
|
|
|
|
|
def _whipsaw_env_prefix(strategy: str) -> str:
|
|
"""전략명 → env 접두사 (SCALPING→SCALP, SHORT→TAIL)."""
|
|
u = str(strategy or "").strip().upper()
|
|
if u in ("SHORT", "TAIL"):
|
|
return "TAIL"
|
|
if u in ("SCALPING", "SCALP"):
|
|
return "SCALP"
|
|
if u in ("US_MOMENTUM", "USMOMENTUM"):
|
|
return "US_MOMENTUM"
|
|
return u
|
|
|
|
|
|
def build_whipsaw_clear_patch(strategy: str) -> Dict[str, str]:
|
|
"""
|
|
「호가만」적용 시 휩쏘 OFF + 수치 비움.
|
|
(이전에는 MOMENTUM만 ENABLED=false 해서 스캘 등에 휩쏘가 남았음)
|
|
"""
|
|
pfx = _whipsaw_env_prefix(strategy)
|
|
if not pfx:
|
|
return {}
|
|
return {
|
|
f"{pfx}_WHIPSAW_FILTER_ENABLED": "false",
|
|
f"{pfx}_WHIPSAW_MODE": "",
|
|
f"{pfx}_WHIPSAW_SUBBAR_SEC": "",
|
|
f"{pfx}_WHIPSAW_LOOKBACK_SEC": "",
|
|
f"{pfx}_WHIPSAW_DIP_PCT": "",
|
|
f"{pfx}_WHIPSAW_RECOVERY_TOL_PCT": "",
|
|
}
|
|
|
|
|
|
def build_whipsaw_env_patch(rec: Dict[str, Any]) -> Dict[str, str]:
|
|
"""휩쏘 추천 결과를 DB env 패치 dict로 변환."""
|
|
if not rec or not rec.get("ok"):
|
|
return {}
|
|
|
|
strat = str(rec.get("strategy") or "").strip().upper()
|
|
pfx = _whipsaw_env_prefix(strat)
|
|
p = rec.get("params", {})
|
|
if not pfx or not p:
|
|
return {}
|
|
|
|
if strat in _WHIPSAW_DB_SKIP_STRATS:
|
|
logger.info(
|
|
"🚫 [%s] 휩쏘 필터 DB 적용 차단 (전략 특성상 UI 표시만) — 수치: subbar=%s lookback=%s dip=%s",
|
|
strat,
|
|
p.get('whipsaw_subbar_sec'),
|
|
p.get('whipsaw_lookback_sec'),
|
|
p.get('whipsaw_dip_pct'),
|
|
)
|
|
return {}
|
|
|
|
patch = {
|
|
f"{pfx}_WHIPSAW_FILTER_ENABLED": "true",
|
|
f"{pfx}_WHIPSAW_SUBBAR_SEC": str(p["whipsaw_subbar_sec"]),
|
|
f"{pfx}_WHIPSAW_LOOKBACK_SEC": str(p["whipsaw_lookback_sec"]),
|
|
f"{pfx}_WHIPSAW_DIP_PCT": str(p["whipsaw_dip_pct"]),
|
|
}
|
|
return patch
|