1374 lines
51 KiB
Python
1374 lines
51 KiB
Python
"""
|
||
kis_trader/backtest/optuna_orderbook_recommend.py
|
||
=================================================
|
||
Optuna 차트 캔들 최적화(Stage 1)가 완료된 후, 후처리(Stage 2)로 진입/익절/손절 켜기·끄기 8방(방 안 TPE)을
|
||
돌린다. 시뮬은 켠 축만 스택(진입→익절호가→손절호가). 실매 엔진은 수정하지 않는다.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import math
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
import optuna
|
||
|
||
from database import TradeDB
|
||
|
||
logger = logging.getLogger("OptunaOBRecommend")
|
||
optuna.logging.set_verbosity(optuna.logging.WARNING)
|
||
|
||
|
||
@dataclass
|
||
class Snap:
|
||
t: datetime
|
||
total_bid: int
|
||
total_ask: int
|
||
best_bid: int
|
||
best_ask: int
|
||
bid_qty_l3: int = 0
|
||
ask_qty_l3: int = 0
|
||
source: str = "" # kiwoom_0d / ls_uh1 … 후처리 피드 추적용
|
||
|
||
|
||
@dataclass
|
||
class TradeInfo:
|
||
code: str
|
||
name: str
|
||
buy_dt: datetime
|
||
buy_price: float
|
||
sell_price: float
|
||
qty: int
|
||
actual_pnl: float
|
||
actual_profit_rate: float
|
||
entry_snaps: List[Snap]
|
||
holding_snaps: List[Snap]
|
||
orig_spread_pct: float
|
||
orig_bid_ask_ratio: float
|
||
orig_ask_qty_l3: int
|
||
passed_current: bool
|
||
|
||
|
||
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 _ob_n_trials(default: int = 1000) -> int:
|
||
from kis_trader.utils.env import get_env_int
|
||
return max(10, int(get_env_int("OPTUNA_OB_RECOMMEND_TRIALS", int(default))))
|
||
|
||
|
||
def _ob_lookback_horizon() -> Tuple[timedelta, timedelta]:
|
||
from kis_trader.utils.env import get_env_int
|
||
lb = max(1, int(get_env_int("OPTUNA_OB_LOOKBACK_MIN", 30)))
|
||
hz = max(1, int(get_env_int("OPTUNA_OB_HORIZON_MIN", 6)))
|
||
return timedelta(minutes=lb), timedelta(minutes=hz)
|
||
|
||
|
||
def _parse_dt(v: Any) -> datetime:
|
||
if isinstance(v, datetime):
|
||
return v
|
||
s = str(v or "").strip()
|
||
if not s:
|
||
raise ValueError("empty dt")
|
||
if s[:10].isdigit() and ("-" in s[:12] or " " in s or "T" in s):
|
||
return datetime.strptime(s[:19].replace("T", " "), "%Y-%m-%d %H:%M:%S")
|
||
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.strptime(s[:19], "%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
def _snap_to_dt(snap_time: str) -> Optional[datetime]:
|
||
s = str(snap_time or "").strip()
|
||
if not s or len(s) < 14 or s == "None":
|
||
return None
|
||
try:
|
||
return datetime(int(s[:4]), int(s[4:6]), int(s[6:8]), int(s[8:10]), int(s[10:12]), int(s[12:14]))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _parse_ratchet_tiers(val: str) -> List[Tuple[int, float]]:
|
||
t = []
|
||
for p in str(val or "").strip().split(","):
|
||
if ":" in p:
|
||
parts = p.split(":")
|
||
try:
|
||
t.append((int(parts[0].strip()), float(parts[1].strip())))
|
||
except ValueError:
|
||
pass
|
||
t.sort(key=lambda x: x[0])
|
||
if not t:
|
||
t = [(10, 2.6), (13, 2.2)]
|
||
return t
|
||
|
||
|
||
def resolve_orderbook_recommend_table(
|
||
*,
|
||
ob_table: Optional[str] = None,
|
||
history_source: Optional[str] = None,
|
||
ob_source: Optional[str] = None,
|
||
) -> Tuple[str, Tuple[str, ...]]:
|
||
"""후처리 호가 테이블 — 전략명 하드코딩 금지.
|
||
|
||
우선순위 (백테 ``trigger_snapshot_loader`` / ``OB_SOURCE`` 와 동일 축):
|
||
1) 명시 ``ob_table``
|
||
2) ``ob_source`` 또는 env ``OB_SOURCE`` (kis|kiwoom|kiwoom_0d|ls)
|
||
3) ``history_source`` 또는 ``BACKTEST_UNIVERSE_HISTORY_SOURCE`` (ls → ls_ws_orderbook)
|
||
|
||
Returns:
|
||
(table_name, source_filter) — source_filter 비어 있으면 source 조건 없음.
|
||
"""
|
||
import os
|
||
|
||
if ob_table and str(ob_table).strip():
|
||
t = str(ob_table).strip()
|
||
if t == "kis_ws_orderbook":
|
||
return t, tuple()
|
||
if t == "ls_ws_orderbook":
|
||
return t, ("ls_uh1", "ls_h1", "ls_ha", "ls_nh1")
|
||
return t, ("kiwoom_0d",)
|
||
|
||
raw_ob = (ob_source if ob_source is not None else os.environ.get("OB_SOURCE", "")).strip().lower()
|
||
if raw_ob in ("kis", "kis_ws"):
|
||
return "kis_ws_orderbook", tuple()
|
||
if raw_ob in ("ls", "ls_condition", "ls_ws", "ls_afr"):
|
||
return "ls_ws_orderbook", ("ls_uh1", "ls_h1", "ls_ha", "ls_nh1")
|
||
if raw_ob in ("kiwoom", "kiwoom_0d", "0d"):
|
||
return "ws_orderbook", ("kiwoom_0d",)
|
||
|
||
try:
|
||
from kis_trader.backtest.universe_history_source import (
|
||
resolve_backtest_universe_history_source,
|
||
)
|
||
|
||
hs = resolve_backtest_universe_history_source(history_source)
|
||
except Exception:
|
||
hs = str(history_source or "kiwoom").strip().lower()
|
||
if hs in ("ls", "ls_condition", "ls_afr", "ls_ws"):
|
||
hs = "ls"
|
||
else:
|
||
hs = "kiwoom"
|
||
|
||
if hs == "ls":
|
||
return "ls_ws_orderbook", ("ls_uh1", "ls_h1", "ls_ha", "ls_nh1")
|
||
# 기본(키움 이력) — 실수집 본체
|
||
return "ws_orderbook", ("kiwoom_0d",)
|
||
|
||
|
||
def get_orderbook_table_for_strategy(strategy: str) -> str:
|
||
"""호환용 — 전략명으로 LS 강제하지 않음. history/ob_source 해석."""
|
||
table, _src = resolve_orderbook_recommend_table()
|
||
return table
|
||
|
||
|
||
def _strategy_config_table_and_prefix(strat_upper: str) -> Tuple[str, str]:
|
||
s = (strat_upper or "").strip().upper()
|
||
if "BREAKOUT" in s:
|
||
return "config_breakout", "BREAKOUT_"
|
||
if "SCALP" in s:
|
||
return "config_scalp", "SCALP_"
|
||
if s in ("SHORT", "TAIL") or "TAIL" in s:
|
||
return "config_short", "TAIL_"
|
||
if "US_MOMENTUM" in s or s.startswith("US"):
|
||
return "config_us_momentum", "US_MOMENTUM_"
|
||
return "config_momentum", "MOMENTUM_"
|
||
|
||
|
||
def _fetch_ob_snaps(
|
||
db: Any,
|
||
table: str,
|
||
cols: List[str],
|
||
source_filter: Tuple[str, ...],
|
||
code: str,
|
||
buy_dt: datetime,
|
||
lookback: timedelta,
|
||
horizon: timedelta,
|
||
) -> List[Snap]:
|
||
sel = ["snap_time", "total_bid_qty", "total_ask_qty", "best_bid", "best_ask"]
|
||
if "bid_qty_l3" in cols:
|
||
sel.append("bid_qty_l3")
|
||
if "ask_qty_l3" in cols:
|
||
sel.append("ask_qty_l3")
|
||
if "source" in cols:
|
||
sel.append("source")
|
||
snap_sql = (
|
||
f"SELECT {', '.join(sel)} FROM {table} "
|
||
"WHERE code=%s AND snap_time >= %s AND snap_time < %s"
|
||
)
|
||
snap_params: List[Any] = [
|
||
code,
|
||
(buy_dt - lookback).strftime("%Y%m%d%H%M%S"),
|
||
(buy_dt + horizon).strftime("%Y%m%d%H%M%S"),
|
||
]
|
||
if source_filter and "source" in cols:
|
||
ph = ",".join(["%s"] * len(source_filter))
|
||
snap_sql += f" AND source IN ({ph})"
|
||
snap_params.extend(source_filter)
|
||
snap_sql += " ORDER BY snap_time ASC"
|
||
s_rows = db.conn.execute(snap_sql, tuple(snap_params)).fetchall()
|
||
snaps: List[Snap] = []
|
||
for sr in s_rows:
|
||
s_dt = _snap_to_dt(sr["snap_time"])
|
||
if not s_dt:
|
||
continue
|
||
snaps.append(
|
||
Snap(
|
||
t=s_dt,
|
||
total_bid=int(sr["total_bid_qty"] or 0),
|
||
total_ask=int(sr["total_ask_qty"] or 0),
|
||
best_bid=int(sr["best_bid"] or 0),
|
||
best_ask=int(sr["best_ask"] or 0),
|
||
bid_qty_l3=int(sr["bid_qty_l3"] or 0) if "bid_qty_l3" in sr else 0,
|
||
ask_qty_l3=int(sr["ask_qty_l3"] or 0) if "ask_qty_l3" in sr else 0,
|
||
source=str(sr.get("source") or "").strip().lower() if "source" in sr else "",
|
||
)
|
||
)
|
||
return snaps
|
||
|
||
|
||
def _tradeinfo_from_fill(
|
||
*,
|
||
code: str,
|
||
name: str,
|
||
buy_dt: datetime,
|
||
buy_price: float,
|
||
sell_price: float,
|
||
qty: int,
|
||
actual_pnl: float,
|
||
actual_profit_rate: float,
|
||
snaps: List[Snap],
|
||
) -> Optional[TradeInfo]:
|
||
if not snaps or buy_price <= 0 or qty <= 0:
|
||
return None
|
||
before = [s for s in snaps if s.t <= buy_dt]
|
||
entry_snaps = before if before else snaps
|
||
holding_snaps = [s for s in snaps if s.t > buy_dt]
|
||
best_entry_snap = entry_snaps[-1]
|
||
bid, ask = best_entry_snap.best_bid, best_entry_snap.best_ask
|
||
spread_pct = ((ask - bid) / ((ask + bid) / 2.0)) * 100.0 if (ask > 0 and bid > 0) else 0.0
|
||
tot_b, tot_a = best_entry_snap.total_bid, best_entry_snap.total_ask
|
||
ratio = (tot_b / tot_a) if tot_a > 0 else 999.0
|
||
return TradeInfo(
|
||
code=code,
|
||
name=name,
|
||
buy_dt=buy_dt,
|
||
buy_price=buy_price,
|
||
sell_price=sell_price,
|
||
qty=qty,
|
||
actual_pnl=actual_pnl,
|
||
actual_profit_rate=actual_profit_rate,
|
||
entry_snaps=entry_snaps,
|
||
holding_snaps=holding_snaps,
|
||
orig_spread_pct=spread_pct,
|
||
orig_bid_ask_ratio=ratio,
|
||
orig_ask_qty_l3=int(best_entry_snap.ask_qty_l3 or 0),
|
||
passed_current=False,
|
||
)
|
||
|
||
|
||
def raw_fills_to_ob_trades(
|
||
raw_fills: List[Dict[str, Any]],
|
||
*,
|
||
db: Any,
|
||
table: str,
|
||
cols: List[str],
|
||
source_filter: Tuple[str, ...],
|
||
) -> List[TradeInfo]:
|
||
"""백테 체결 dict(buy_time/buy_price/qty/pnl) → 호가 TradeInfo."""
|
||
lookback, horizon = _ob_lookback_horizon()
|
||
out: List[TradeInfo] = []
|
||
n_miss = 0
|
||
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_dt(raw_dt)
|
||
except Exception:
|
||
continue
|
||
buy_price = float(b.get("buy_price") or b.get("entry") or 0)
|
||
qty = int(b.get("qty") or 0)
|
||
actual_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))
|
||
actual_profit_rate = float(b.get("profit_rate") or b.get("actual_profit_rate") or 0)
|
||
sell_price = float(b.get("sell_price") or b.get("exit") or 0)
|
||
snaps = _fetch_ob_snaps(db, table, cols, source_filter, code, buy_dt, lookback, horizon)
|
||
ti = _tradeinfo_from_fill(
|
||
code=code,
|
||
name=str(b.get("name") or code),
|
||
buy_dt=buy_dt,
|
||
buy_price=buy_price,
|
||
sell_price=sell_price,
|
||
qty=qty,
|
||
actual_pnl=actual_pnl,
|
||
actual_profit_rate=actual_profit_rate,
|
||
snaps=snaps,
|
||
)
|
||
# 후처리 피드 추적: 스냅 hit/miss + 벤더·bid/ask
|
||
try:
|
||
from kis_trader.backtest.optuna_feed_trace import maybe_log_bt_ob_postprocess_sample
|
||
|
||
if ti is not None:
|
||
es = ti.entry_snaps[-1] if ti.entry_snaps else None
|
||
maybe_log_bt_ob_postprocess_sample(
|
||
code=code,
|
||
buy_time=buy_dt.strftime("%Y%m%d%H%M%S"),
|
||
buy_price=buy_price,
|
||
snap_t=es.t.strftime("%Y%m%d%H%M%S") if es else "",
|
||
best_bid=int(es.best_bid) if es else 0,
|
||
best_ask=int(es.best_ask) if es else 0,
|
||
spread_pct=float(ti.orig_spread_pct),
|
||
bid_ask_ratio=float(ti.orig_bid_ask_ratio),
|
||
snap_source=str(es.source or "") if es else "",
|
||
hit=True,
|
||
)
|
||
else:
|
||
n_miss += 1
|
||
maybe_log_bt_ob_postprocess_sample(
|
||
code=code,
|
||
buy_time=buy_dt.strftime("%Y%m%d%H%M%S"),
|
||
buy_price=buy_price,
|
||
hit=False,
|
||
)
|
||
except Exception:
|
||
if ti is None:
|
||
n_miss += 1
|
||
if ti:
|
||
out.append(ti)
|
||
if n_miss > 0:
|
||
logger.info(
|
||
"🔎 [호가후처리] 체결→스냅 미스 %s/%s건 (필터·기간·종목 공백)",
|
||
n_miss, len(raw_fills or []),
|
||
)
|
||
return out
|
||
|
||
|
||
def _ob_axis_n_trials(n_trials: int) -> int:
|
||
"""레거시: 축 독립 500회. 8방 경로(_ob_combo_n_trials)에서는 미사용."""
|
||
from kis_trader.utils.env import get_env_int
|
||
axis = int(get_env_int("OPTUNA_OB_AXIS_TRIALS", 0) or 0)
|
||
if axis > 0:
|
||
return max(10, axis)
|
||
return max(10, int(n_trials or _ob_n_trials(500)))
|
||
|
||
|
||
def _ob_combo_n_trials(n_on: int) -> int:
|
||
"""8방 중 켜진 축 개수별 trial. 축당 500×8 금지 — 합이 구 1,500 근방."""
|
||
from kis_trader.utils.env import get_env_int
|
||
|
||
n = int(n_on or 0)
|
||
if n <= 0:
|
||
return 0
|
||
if n == 1:
|
||
raw = int(get_env_int("OPTUNA_OB_COMBO_TRIALS_SINGLE", 150) or 0)
|
||
return max(10, raw if raw > 0 else 150)
|
||
if n == 2:
|
||
raw = int(get_env_int("OPTUNA_OB_COMBO_TRIALS_DOUBLE", 200) or 0)
|
||
return max(10, raw if raw > 0 else 200)
|
||
raw = int(get_env_int("OPTUNA_OB_COMBO_TRIALS_TRIPLE", 250) or 0)
|
||
return max(10, raw if raw > 0 else 250)
|
||
|
||
|
||
def _ob_orig_stats(trades: List[TradeInfo]) -> Dict[str, Any]:
|
||
orig_cnt = len(trades)
|
||
if orig_cnt <= 0:
|
||
return {"count": 0, "win_rate": 0.0, "pnl": 0, "avg_rate": 0.0}
|
||
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
|
||
return {
|
||
"count": orig_cnt,
|
||
"win_rate": round(orig_win, 1),
|
||
"pnl": _krw_int(orig_pnl),
|
||
"avg_rate": round(orig_rate, 2),
|
||
}
|
||
|
||
|
||
def _suite_stats(rows: List[Tuple[float, float, str]], *, skip_reject: bool = True) -> Dict[str, Any]:
|
||
kept = [(p, r) for p, r, t in rows if (not skip_reject) or t != "ENTRY_REJECTED"]
|
||
t_cnt = len(kept)
|
||
if t_cnt <= 0:
|
||
return {"count": 0, "win_rate": 0.0, "pnl": 0, "avg_rate": 0.0, "pnl_diff": 0}
|
||
tot_pnl = sum(p for p, _ in kept)
|
||
tot_rate = sum(r for _, r in kept)
|
||
w_cnt = sum(1 for p, _ in kept if p > 0)
|
||
return {
|
||
"count": t_cnt,
|
||
"win_rate": round(w_cnt / t_cnt * 100.0, 1),
|
||
"pnl": _krw_int(tot_pnl),
|
||
"avg_rate": round(tot_rate / t_cnt, 2),
|
||
}
|
||
|
||
|
||
def _axis_score(cnt: int, orig_cnt: int, win_r: float, pnl: float) -> float:
|
||
if cnt < max(3, int(orig_cnt * 0.3)):
|
||
return -999999999.0
|
||
score = (pnl / 100000.0) + win_r * 2.0
|
||
if win_r >= 60.0:
|
||
score += (win_r - 60.0) * 1.5
|
||
return score
|
||
|
||
|
||
def _snap_or_and_price(s: Snap) -> Tuple[float, float]:
|
||
cur_p = float(s.best_bid if s.best_bid > 0 else s.best_ask)
|
||
ratio = (s.total_bid / s.total_ask) if s.total_ask > 0 else 1.0
|
||
return cur_p, ratio
|
||
|
||
|
||
def _sim_entry(tr: TradeInfo, p: Dict[str, Any]) -> Tuple[float, float, str]:
|
||
if tr.orig_spread_pct > float(p["max_spread_pct"]) or tr.orig_bid_ask_ratio < float(p["min_bid_ask_ratio"]):
|
||
return (0.0, 0.0, "ENTRY_REJECTED")
|
||
# 실매 매도벽: L3 매도잔량 > 주문수량 × 배수 이면 탈락. L3 없으면 이 축은 스킵.
|
||
ask_mult = float(p.get("ask_max_mult") or 0.0)
|
||
ask_l3 = int(tr.orig_ask_qty_l3 or 0)
|
||
if ask_mult > 0 and tr.qty > 0 and ask_l3 > 0:
|
||
if ask_l3 > int(tr.qty * ask_mult):
|
||
return (0.0, 0.0, "ENTRY_REJECTED")
|
||
return (tr.actual_pnl, tr.actual_profit_rate, "ORIGINAL")
|
||
|
||
|
||
def _sim_exit_ob(tr: TradeInfo, p: Dict[str, Any]) -> Tuple[float, float, str]:
|
||
"""실매 _check_exit_ob_l3 와 동일 조건 — 엔진 함수는 수정하지 않고 호출만."""
|
||
from kis_trader.engine.momentum_hts_logic import _check_exit_ob_l3
|
||
|
||
if not p.get("exit_ob_enabled"):
|
||
return (tr.actual_pnl, tr.actual_profit_rate, "ORIGINAL")
|
||
params = {
|
||
"exit_ob_enabled": True,
|
||
"exit_ob_ratio_min": float(p.get("exit_ob_ratio_min") or p.get("ob_ratio_min") or 0.4),
|
||
"exit_ob_ma_window": int(p.get("exit_ob_ma_window") or p.get("ma_window") or 5),
|
||
"exit_ob_min_profit_pct": float(p.get("exit_ob_min_profit_pct") or p.get("min_profit_pct") or 0.005),
|
||
"exit_ob_min_hold_bars": int(p.get("exit_ob_min_hold_bars") or p.get("min_hold_bars") or 3),
|
||
}
|
||
history: List[Optional[float]] = []
|
||
for s in tr.holding_snaps:
|
||
cur_p, ratio = _snap_or_and_price(s)
|
||
if cur_p <= 0:
|
||
history.append(None)
|
||
continue
|
||
history.append(ratio)
|
||
hold_bars = max(0, int((s.t - tr.buy_dt).total_seconds() / 60.0))
|
||
if _check_exit_ob_l3(params, history, tr.buy_price, cur_p, hold_bars):
|
||
realized = (cur_p - tr.buy_price) * tr.qty
|
||
rate = ((cur_p - tr.buy_price) / tr.buy_price) * 100.0 if tr.buy_price else 0.0
|
||
return (realized, rate, "OB_EXIT")
|
||
return (tr.actual_pnl, tr.actual_profit_rate, "HOLD_TO_ORIG")
|
||
|
||
|
||
def _sim_stop_ob(tr: TradeInfo, p: Dict[str, Any]) -> Tuple[float, float, str]:
|
||
"""실매 _check_stop_ob 와 동일 조건 — 엔진 본체 미수정."""
|
||
from kis_trader.engine.momentum_hts_logic import _check_stop_ob
|
||
|
||
if not p.get("stop_ob_enabled"):
|
||
return (tr.actual_pnl, tr.actual_profit_rate, "ORIGINAL")
|
||
params = {
|
||
"stop_ob_enabled": True,
|
||
"stop_ob_ratio_min": float(p.get("stop_ob_ratio_min") or 0.4),
|
||
"stop_ob_ma_window": int(p.get("stop_ob_ma_window") or 5),
|
||
"stop_ob_min_loss_pct": float(p.get("stop_ob_min_loss_pct") or 0.003),
|
||
"stop_ob_min_hold_bars": int(p.get("stop_ob_min_hold_bars") or 2),
|
||
}
|
||
history: List[Optional[float]] = []
|
||
for s in tr.holding_snaps:
|
||
cur_p, ratio = _snap_or_and_price(s)
|
||
if cur_p <= 0:
|
||
history.append(None)
|
||
continue
|
||
history.append(ratio)
|
||
hold_bars = max(0, int((s.t - tr.buy_dt).total_seconds() / 60.0))
|
||
if _check_stop_ob(params, history, tr.buy_price, cur_p, hold_bars):
|
||
realized = (cur_p - tr.buy_price) * tr.qty
|
||
rate = ((cur_p - tr.buy_price) / tr.buy_price) * 100.0 if tr.buy_price else 0.0
|
||
return (realized, rate, "OB_STOP")
|
||
return (tr.actual_pnl, tr.actual_profit_rate, "HOLD_TO_ORIG")
|
||
|
||
|
||
def _entry_sim_p(p: Dict[str, Any]) -> Dict[str, Any]:
|
||
return {
|
||
"max_spread_pct": p.get("max_spread_pct", p.get("orderbook_max_spread_pct")),
|
||
"min_bid_ask_ratio": p.get("min_bid_ask_ratio", p.get("orderbook_min_bid_ask_ratio")),
|
||
"ask_max_mult": p.get("ask_max_mult", p.get("orderbook_entry_ask_max_mult")),
|
||
}
|
||
|
||
|
||
def _flag_on(p: Dict[str, Any], raw_key: str, enabled_key: str) -> bool:
|
||
if raw_key in p:
|
||
return bool(p.get(raw_key))
|
||
return bool(p.get(enabled_key))
|
||
|
||
|
||
def _sim_stacked(tr: TradeInfo, p: Dict[str, Any]) -> Tuple[float, float, str]:
|
||
"""켜진 축만. 진입 탈락 → 보유 중 익절호가 → 손절호가 (실매 OB 순서)."""
|
||
use_e = _flag_on(p, "entry_on", "orderbook_filter_enabled")
|
||
use_x = _flag_on(p, "exit_on", "exit_ob_enabled")
|
||
use_s = _flag_on(p, "stop_on", "stop_ob_enabled")
|
||
if use_e:
|
||
ep = _entry_sim_p(p)
|
||
if ep.get("max_spread_pct") is None or ep.get("min_bid_ask_ratio") is None:
|
||
return (0.0, 0.0, "ENTRY_REJECTED")
|
||
r = _sim_entry(tr, ep)
|
||
if r[2] == "ENTRY_REJECTED":
|
||
return r
|
||
if not (use_x or use_s):
|
||
return (tr.actual_pnl, tr.actual_profit_rate, "ORIGINAL")
|
||
|
||
from kis_trader.engine.momentum_hts_logic import _check_exit_ob_l3, _check_stop_ob
|
||
|
||
exit_p = None
|
||
stop_p = None
|
||
if use_x:
|
||
exit_p = {
|
||
"exit_ob_enabled": True,
|
||
"exit_ob_ratio_min": float(p.get("exit_ob_ratio_min") or p.get("ob_ratio_min") or 0.4),
|
||
"exit_ob_ma_window": int(p.get("exit_ob_ma_window") or p.get("ma_window") or 5),
|
||
"exit_ob_min_profit_pct": float(p.get("exit_ob_min_profit_pct") or p.get("min_profit_pct") or 0.005),
|
||
"exit_ob_min_hold_bars": int(p.get("exit_ob_min_hold_bars") or p.get("min_hold_bars") or 3),
|
||
}
|
||
if use_s:
|
||
stop_p = {
|
||
"stop_ob_enabled": True,
|
||
"stop_ob_ratio_min": float(p.get("stop_ob_ratio_min") or 0.4),
|
||
"stop_ob_ma_window": int(p.get("stop_ob_ma_window") or 5),
|
||
"stop_ob_min_loss_pct": float(p.get("stop_ob_min_loss_pct") or 0.003),
|
||
"stop_ob_min_hold_bars": int(p.get("stop_ob_min_hold_bars") or 2),
|
||
}
|
||
history: List[Optional[float]] = []
|
||
for s in tr.holding_snaps:
|
||
cur_p, ratio = _snap_or_and_price(s)
|
||
if cur_p <= 0:
|
||
history.append(None)
|
||
continue
|
||
history.append(ratio)
|
||
hold_bars = max(0, int((s.t - tr.buy_dt).total_seconds() / 60.0))
|
||
if exit_p and _check_exit_ob_l3(exit_p, history, tr.buy_price, cur_p, hold_bars):
|
||
realized = (cur_p - tr.buy_price) * tr.qty
|
||
rate = ((cur_p - tr.buy_price) / tr.buy_price) * 100.0 if tr.buy_price else 0.0
|
||
return (realized, rate, "OB_EXIT")
|
||
if stop_p and _check_stop_ob(stop_p, history, tr.buy_price, cur_p, hold_bars):
|
||
realized = (cur_p - tr.buy_price) * tr.qty
|
||
rate = ((cur_p - tr.buy_price) / tr.buy_price) * 100.0 if tr.buy_price else 0.0
|
||
return (realized, rate, "OB_STOP")
|
||
return (tr.actual_pnl, tr.actual_profit_rate, "HOLD_TO_ORIG")
|
||
|
||
|
||
def _finalize_combo_params(
|
||
p: Dict[str, Any],
|
||
*,
|
||
use_e: bool,
|
||
use_x: bool,
|
||
use_s: bool,
|
||
) -> Dict[str, Any]:
|
||
out: Dict[str, Any] = {}
|
||
if use_e:
|
||
out["orderbook_filter_enabled"] = True
|
||
out["orderbook_max_spread_pct"] = p.get("max_spread_pct", p.get("orderbook_max_spread_pct"))
|
||
out["orderbook_min_bid_ask_ratio"] = p.get("min_bid_ask_ratio", p.get("orderbook_min_bid_ask_ratio"))
|
||
out["orderbook_entry_ask_max_mult"] = p.get("ask_max_mult", p.get("orderbook_entry_ask_max_mult"))
|
||
else:
|
||
out["orderbook_filter_enabled"] = False
|
||
if use_x:
|
||
out["exit_ob_enabled"] = True
|
||
out["exit_ob_min_hold_bars"] = p.get("exit_ob_min_hold_bars")
|
||
out["exit_ob_ratio_min"] = p.get("exit_ob_ratio_min")
|
||
out["exit_ob_min_profit_pct"] = p.get("exit_ob_min_profit_pct")
|
||
out["exit_ob_ma_window"] = p.get("exit_ob_ma_window")
|
||
else:
|
||
out["exit_ob_enabled"] = False
|
||
if use_s:
|
||
out["stop_ob_enabled"] = True
|
||
out["stop_ob_min_hold_bars"] = p.get("stop_ob_min_hold_bars")
|
||
out["stop_ob_ratio_min"] = p.get("stop_ob_ratio_min")
|
||
out["stop_ob_min_loss_pct"] = p.get("stop_ob_min_loss_pct")
|
||
out["stop_ob_ma_window"] = p.get("stop_ob_ma_window")
|
||
else:
|
||
out["stop_ob_enabled"] = False
|
||
return out
|
||
|
||
|
||
def _suggest_combo(
|
||
trial: Any,
|
||
*,
|
||
use_e: bool,
|
||
use_x: bool,
|
||
use_s: bool,
|
||
) -> Dict[str, Any]:
|
||
from kis_trader.utils.env import get_env_float, get_env_int
|
||
|
||
p: Dict[str, Any] = {"entry_on": bool(use_e), "exit_on": bool(use_x), "stop_on": bool(use_s)}
|
||
if use_e:
|
||
lo_s = float(get_env_float("OPTUNA_OB_ENTRY_SPREAD_MIN", 0.1))
|
||
hi_s = float(get_env_float("OPTUNA_OB_ENTRY_SPREAD_MAX", 8.0))
|
||
lo_r = float(get_env_float("OPTUNA_OB_ENTRY_RATIO_MIN", 0.05))
|
||
hi_r = float(get_env_float("OPTUNA_OB_ENTRY_RATIO_MAX", 1.5))
|
||
lo_a = float(get_env_float("OPTUNA_OB_ENTRY_ASK_MULT_MIN", 1.0))
|
||
hi_a = float(get_env_float("OPTUNA_OB_ENTRY_ASK_MULT_MAX", 80.0))
|
||
if hi_s < lo_s:
|
||
lo_s, hi_s = hi_s, lo_s
|
||
if hi_r < lo_r:
|
||
lo_r, hi_r = hi_r, lo_r
|
||
if hi_a < lo_a:
|
||
lo_a, hi_a = hi_a, lo_a
|
||
p["max_spread_pct"] = trial.suggest_float("max_spread_pct", lo_s, hi_s, step=0.1)
|
||
p["min_bid_ask_ratio"] = trial.suggest_float("min_bid_ask_ratio", lo_r, hi_r, step=0.05)
|
||
p["ask_max_mult"] = trial.suggest_float("ask_max_mult", lo_a, hi_a, step=1.0)
|
||
if use_x:
|
||
p["exit_ob_enabled"] = True
|
||
p["exit_ob_min_hold_bars"] = trial.suggest_int(
|
||
"exit_ob_min_hold_bars",
|
||
int(get_env_int("OPTUNA_OB_EXIT_HOLD_MIN", 1)),
|
||
int(get_env_int("OPTUNA_OB_EXIT_HOLD_MAX", 5)),
|
||
)
|
||
p["exit_ob_ratio_min"] = trial.suggest_float(
|
||
"exit_ob_ratio_min",
|
||
float(get_env_float("OPTUNA_OB_EXIT_RATIO_MIN", 0.2)),
|
||
float(get_env_float("OPTUNA_OB_EXIT_RATIO_MAX", 0.8)),
|
||
step=0.05,
|
||
)
|
||
p["exit_ob_min_profit_pct"] = trial.suggest_float(
|
||
"exit_ob_min_profit_pct",
|
||
float(get_env_float("OPTUNA_OB_EXIT_PROFIT_MIN", 0.003)),
|
||
float(get_env_float("OPTUNA_OB_EXIT_PROFIT_MAX", 0.02)),
|
||
step=0.001,
|
||
)
|
||
p["exit_ob_ma_window"] = trial.suggest_int(
|
||
"exit_ob_ma_window",
|
||
int(get_env_int("OPTUNA_OB_EXIT_MA_MIN", 3)),
|
||
int(get_env_int("OPTUNA_OB_EXIT_MA_MAX", 10)),
|
||
)
|
||
if use_s:
|
||
p["stop_ob_enabled"] = True
|
||
p["stop_ob_min_hold_bars"] = trial.suggest_int(
|
||
"stop_ob_min_hold_bars",
|
||
int(get_env_int("OPTUNA_OB_STOP_HOLD_MIN", 1)),
|
||
int(get_env_int("OPTUNA_OB_STOP_HOLD_MAX", 5)),
|
||
)
|
||
p["stop_ob_ratio_min"] = trial.suggest_float(
|
||
"stop_ob_ratio_min",
|
||
float(get_env_float("OPTUNA_OB_STOP_RATIO_MIN", 0.2)),
|
||
float(get_env_float("OPTUNA_OB_STOP_RATIO_MAX", 0.8)),
|
||
step=0.05,
|
||
)
|
||
p["stop_ob_min_loss_pct"] = trial.suggest_float(
|
||
"stop_ob_min_loss_pct",
|
||
float(get_env_float("OPTUNA_OB_STOP_LOSS_MIN", 0.001)),
|
||
float(get_env_float("OPTUNA_OB_STOP_LOSS_MAX", 0.02)),
|
||
step=0.001,
|
||
)
|
||
p["stop_ob_ma_window"] = trial.suggest_int(
|
||
"stop_ob_ma_window",
|
||
int(get_env_int("OPTUNA_OB_STOP_MA_MIN", 3)),
|
||
int(get_env_int("OPTUNA_OB_STOP_MA_MAX", 10)),
|
||
)
|
||
return p
|
||
|
||
|
||
def _optimize_combo(
|
||
trades: List[TradeInfo],
|
||
*,
|
||
orig_cnt: int,
|
||
n_trials: int,
|
||
lg: logging.Logger,
|
||
use_e: bool,
|
||
use_x: bool,
|
||
use_s: bool,
|
||
axis_name: str,
|
||
combo_id: str,
|
||
) -> Dict[str, Any]:
|
||
orig = _ob_orig_stats(trades)
|
||
mask = {"entry": bool(use_e), "exit": bool(use_x), "stop": bool(use_s)}
|
||
if int(n_trials or 0) <= 0:
|
||
return {
|
||
"ok": True,
|
||
"reason": "base_no_tpe",
|
||
"params": _finalize_combo_params({}, use_e=False, use_x=False, use_s=False),
|
||
"recommended_stats": orig,
|
||
"orig_stats": orig,
|
||
"n_trials": 0,
|
||
"combo_id": combo_id,
|
||
"mask": mask,
|
||
}
|
||
|
||
def suggest(trial: optuna.Trial) -> Dict[str, Any]:
|
||
return _suggest_combo(trial, use_e=use_e, use_x=use_x, use_s=use_s)
|
||
|
||
rec = _run_axis_study(
|
||
trades=trades,
|
||
orig_cnt=orig_cnt,
|
||
n_trials=n_trials,
|
||
suggest_fn=suggest,
|
||
sim_fn=_sim_stacked,
|
||
skip_reject=bool(use_e),
|
||
lg=lg,
|
||
axis_name=axis_name,
|
||
)
|
||
rec["combo_id"] = combo_id
|
||
rec["mask"] = mask
|
||
if rec.get("ok"):
|
||
rec["params"] = _finalize_combo_params(
|
||
rec.get("params") or {}, use_e=use_e, use_x=use_x, use_s=use_s,
|
||
)
|
||
rows = [_sim_stacked(t, rec["params"]) for t in trades]
|
||
rec_st = _suite_stats(rows, skip_reject=bool(use_e))
|
||
rec_st["pnl_diff"] = _krw_int(int(rec_st["pnl"]) - int(orig["pnl"]))
|
||
rec["recommended_stats"] = rec_st
|
||
rec["orig_stats"] = orig
|
||
return rec
|
||
|
||
|
||
def _run_axis_study(
|
||
*,
|
||
trades: List[TradeInfo],
|
||
orig_cnt: int,
|
||
n_trials: int,
|
||
suggest_fn: Any,
|
||
sim_fn: Any,
|
||
skip_reject: bool,
|
||
lg: logging.Logger,
|
||
axis_name: str,
|
||
) -> Dict[str, Any]:
|
||
orig = _ob_orig_stats(trades)
|
||
valid_records: List[Dict[str, Any]] = []
|
||
|
||
def obj_func(trial: optuna.Trial) -> float:
|
||
params = suggest_fn(trial)
|
||
rows = [sim_fn(t, params) for t in trades]
|
||
st = _suite_stats(rows, skip_reject=skip_reject)
|
||
score = _axis_score(int(st["count"]), orig_cnt, float(st["win_rate"]), float(st["pnl"]))
|
||
if score > -1e8:
|
||
valid_records.append({"score": score, "params": params, "stats": st})
|
||
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:
|
||
opp.on_ob_axis_trial(n, int(n_trials), axis_name)
|
||
|
||
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:
|
||
lg.info("⚡ [%s] 유효 trial 없음", axis_name)
|
||
return {"ok": False, "reason": "no_valid_trials", "params": {}, "recommended_stats": orig}
|
||
|
||
# 합의: 숫자 median, bool 최빈
|
||
keys = list(top5[0]["params"].keys())
|
||
cons: Dict[str, Any] = {}
|
||
for k in keys:
|
||
vs = [r["params"][k] for r in top5 if k in r["params"]]
|
||
if not vs:
|
||
continue
|
||
if isinstance(vs[0], bool):
|
||
cons[k] = sum(1 for v in vs if v) >= (len(vs) / 2.0)
|
||
elif isinstance(vs[0], int) and not isinstance(vs[0], bool):
|
||
cons[k] = int(round(sum(float(v) for v in vs) / len(vs)))
|
||
else:
|
||
cons[k] = round(sum(float(v) for v in vs) / len(vs), 4)
|
||
rows = [sim_fn(t, cons) for t in trades]
|
||
rec_st = _suite_stats(rows, skip_reject=skip_reject)
|
||
rec_st["pnl_diff"] = _krw_int(int(rec_st["pnl"]) - int(orig["pnl"]))
|
||
lg.info(
|
||
"⚡ [%s] 합의 %s | 건=%s WR=%.1f PnL=%s",
|
||
axis_name, cons, rec_st["count"], rec_st["win_rate"], rec_st["pnl"],
|
||
)
|
||
return {"ok": True, "params": cons, "recommended_stats": rec_st, "orig_stats": orig, "n_trials": n_trials}
|
||
|
||
|
||
def _optimize_entry_axis(trades: List[TradeInfo], *, orig_cnt: int, n_trials: int, lg: logging.Logger) -> Dict[str, Any]:
|
||
return _optimize_combo(
|
||
trades, orig_cnt=orig_cnt, n_trials=n_trials, lg=lg,
|
||
use_e=True, use_x=False, use_s=False, axis_name="100 매수호가", combo_id="e",
|
||
)
|
||
|
||
|
||
def _optimize_exit_axis(trades: List[TradeInfo], *, orig_cnt: int, n_trials: int, lg: logging.Logger) -> Dict[str, Any]:
|
||
return _optimize_combo(
|
||
trades, orig_cnt=orig_cnt, n_trials=n_trials, lg=lg,
|
||
use_e=False, use_x=True, use_s=False, axis_name="010 익절호가", combo_id="x",
|
||
)
|
||
|
||
|
||
def _optimize_stop_axis(trades: List[TradeInfo], *, orig_cnt: int, n_trials: int, lg: logging.Logger) -> Dict[str, Any]:
|
||
return _optimize_combo(
|
||
trades, orig_cnt=orig_cnt, n_trials=n_trials, lg=lg,
|
||
use_e=False, use_x=False, use_s=True, axis_name="001 손절호가", combo_id="s",
|
||
)
|
||
|
||
|
||
def recommend_orderbook_parameters(
|
||
strategy: str = "MOMENTUM",
|
||
n_trials: int = 0,
|
||
ob_table: Optional[str] = None,
|
||
history_source: Optional[str] = None,
|
||
ob_source: Optional[str] = None,
|
||
log: Optional[logging.Logger] = None,
|
||
raw_fills: Optional[List[Dict[str, Any]]] = None,
|
||
date_from: Optional[str] = None,
|
||
date_to: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""호가 후처리. raw_fills 있으면 그 체결만(백테 앵커). 없으면 trade_history(실매 참고행)."""
|
||
lg = log or logger
|
||
if int(n_trials or 0) <= 0:
|
||
n_trials = _ob_n_trials(500)
|
||
strat_upper = strategy.upper()
|
||
table, source_filter = resolve_orderbook_recommend_table(
|
||
ob_table=ob_table,
|
||
history_source=history_source,
|
||
ob_source=ob_source,
|
||
)
|
||
lg.info(
|
||
"📌 [호가 후처리] strategy=%s table=%s source_filter=%s",
|
||
strat_upper, table, source_filter or "(all)",
|
||
)
|
||
|
||
db = TradeDB()
|
||
# 1. 테이블 존재 여부 및 컬럼 검사
|
||
try:
|
||
cols = [r["Field"] for r in db.conn.execute(f"SHOW COLUMNS FROM {table}").fetchall()]
|
||
need = {"code", "snap_time", "total_bid_qty", "total_ask_qty", "best_bid", "best_ask"}
|
||
if need - set(cols):
|
||
lg.warning("⚠️ [%s] 호가 테이블 필수 컬럼 부족. 추천 생략.", table)
|
||
return {"ok": False, "reason": "insufficient_columns", "table": table}
|
||
except Exception as exc:
|
||
lg.warning("⚠️ [%s] 테이블 조회 실패: %s. 추천 생략.", table, exc)
|
||
return {"ok": False, "reason": "table_not_found", "table": table}
|
||
|
||
# 코어 TPE 호가OFF와 무관 — 후처리에서 DB 호가 벤더·기간을 추적 로그
|
||
try:
|
||
from kis_trader.backtest.optuna_feed_trace import (
|
||
log_bt_feed_chain_banner,
|
||
log_bt_postprocess_ob_db_scope,
|
||
reset_bt_feed_sample_counter,
|
||
)
|
||
|
||
reset_bt_feed_sample_counter(postprocess=True)
|
||
log_bt_feed_chain_banner(context="호가후처리")
|
||
log_bt_postprocess_ob_db_scope(
|
||
db,
|
||
table=table,
|
||
cols=cols,
|
||
source_filter=source_filter,
|
||
date_from=str(date_from or ""),
|
||
date_to=str(date_to or ""),
|
||
context="호가후처리",
|
||
)
|
||
except Exception as e:
|
||
lg.debug("호가후처리 피드추적 스킵: %s", e)
|
||
|
||
date_sql = f"SELECT DISTINCT SUBSTR(snap_time, 1, 8) as dt FROM {table}"
|
||
date_params: Tuple[Any, ...] = ()
|
||
if source_filter and "source" in cols:
|
||
ph = ",".join(["%s"] * len(source_filter))
|
||
date_sql += f" WHERE source IN ({ph})"
|
||
date_params = tuple(source_filter)
|
||
date_sql += " ORDER BY dt"
|
||
date_rows = db.conn.execute(date_sql, date_params).fetchall()
|
||
avail_dates = [str(r["dt"]) for r in date_rows if r["dt"] and str(r["dt"]) != "None"]
|
||
# source 필터에 안 걸린 구행만 있을 때 — 필터 없이 1회 재시도
|
||
if not avail_dates and source_filter and "source" in cols:
|
||
date_rows = db.conn.execute(
|
||
f"SELECT DISTINCT SUBSTR(snap_time, 1, 8) as dt FROM {table} ORDER BY dt"
|
||
).fetchall()
|
||
avail_dates = [str(r["dt"]) for r in date_rows if r["dt"] and str(r["dt"]) != "None"]
|
||
if avail_dates:
|
||
source_filter = tuple()
|
||
lg.info("📌 [호가 후처리] source 필터 미스 → 전체 source 사용 table=%s", table)
|
||
if not avail_dates:
|
||
return {
|
||
"ok": False,
|
||
"reason": "no_orderbook_snapshots",
|
||
"table": table,
|
||
"source_filter": list(source_filter),
|
||
}
|
||
|
||
trades: List[TradeInfo] = []
|
||
lookback, horizon = _ob_lookback_horizon()
|
||
if raw_fills:
|
||
trades = raw_fills_to_ob_trades(
|
||
list(raw_fills),
|
||
db=db,
|
||
table=table,
|
||
cols=cols,
|
||
source_filter=source_filter,
|
||
)
|
||
else:
|
||
date_from_s = str(date_from or "").strip()[:10]
|
||
date_to_s = str(date_to or "").strip()[:10]
|
||
for dt_str in avail_dates:
|
||
day_hyphen = f"{dt_str[:4]}-{dt_str[4:6]}-{dt_str[6:]}"
|
||
if date_from_s and day_hyphen < date_from_s:
|
||
continue
|
||
if date_to_s and day_hyphen > date_to_s:
|
||
continue
|
||
buys = db.conn.execute(
|
||
"""
|
||
SELECT id, code, name, buy_date, buy_price, sell_price, qty, profit_rate, realized_pnl
|
||
FROM trade_history
|
||
WHERE strategy=%s AND DATE(buy_date)=%s
|
||
ORDER BY buy_date
|
||
""",
|
||
(strat_upper, day_hyphen),
|
||
).fetchall()
|
||
for b in buys:
|
||
try:
|
||
buy_dt = _parse_dt(b["buy_date"])
|
||
except Exception:
|
||
continue
|
||
snaps = _fetch_ob_snaps(
|
||
db, table, cols, source_filter, str(b["code"]), buy_dt, lookback, horizon,
|
||
)
|
||
ti = _tradeinfo_from_fill(
|
||
code=str(b["code"]),
|
||
name=str(b.get("name") or b["code"]),
|
||
buy_dt=buy_dt,
|
||
buy_price=float(b["buy_price"] or 0),
|
||
sell_price=float(b["sell_price"] or 0),
|
||
qty=int(b["qty"] or 0),
|
||
actual_pnl=float(b["realized_pnl"] or 0),
|
||
actual_profit_rate=float(b["profit_rate"] or 0),
|
||
snaps=snaps,
|
||
)
|
||
if ti:
|
||
trades.append(ti)
|
||
|
||
if len(trades) < 3:
|
||
n_raw = len(raw_fills) if raw_fills else 0
|
||
lg.warning(
|
||
"⚠️ [%s] 호가 연제 가능한 실제 매수 건수(%s건, 체결원본=%s)가 부족하여 최적화 생략.",
|
||
strat_upper, len(trades), n_raw,
|
||
)
|
||
return {
|
||
"ok": False,
|
||
"reason": "not_enough_trades",
|
||
"trade_count": len(trades),
|
||
"fill_count": n_raw,
|
||
}
|
||
|
||
# 연동된 진입 스냅 벤더 요약 (후처리 원인파악)
|
||
try:
|
||
from collections import Counter as _Ctr
|
||
|
||
_src_ctr = _Ctr()
|
||
for _ti in trades:
|
||
_es = _ti.entry_snaps[-1] if _ti.entry_snaps else None
|
||
_src_ctr[str((_es.source if _es else "") or "?").strip().lower() or "?"] += 1
|
||
lg.info(
|
||
"🔎 [호가후처리] 연동체결=%d건 | 진입스냅벤더 %s",
|
||
len(trades),
|
||
" ".join(f"{k}={v}" for k, v in _src_ctr.most_common()),
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
orig_stats = _ob_orig_stats(trades)
|
||
orig_cnt = int(orig_stats["count"])
|
||
can_exit_stop = strat_upper in ("MOMENTUM", "BREAKOUT")
|
||
combo_specs: List[Tuple[str, str, bool, bool, bool]] = [
|
||
("e", "100 매수호가", True, False, False),
|
||
]
|
||
if can_exit_stop:
|
||
combo_specs.extend(
|
||
[
|
||
("x", "010 익절호가", False, True, False),
|
||
("s", "001 손절호가", False, False, True),
|
||
("ex", "110 매수+익절호가", True, True, False),
|
||
("es", "101 매수+손절호가", True, False, True),
|
||
("xs", "011 익절+손절호가", False, True, True),
|
||
("exs", "111 호가전부", True, True, True),
|
||
]
|
||
)
|
||
from kis_trader.backtest import optuna_post_progress as opp
|
||
n_ax = len(combo_specs)
|
||
opp.set_ob_axes(n_ax, 0)
|
||
combos: Dict[str, Any] = {
|
||
"base": {
|
||
"ok": True,
|
||
"reason": "base_no_tpe",
|
||
"params": {},
|
||
"recommended_stats": orig_stats,
|
||
"orig_stats": orig_stats,
|
||
"n_trials": 0,
|
||
"combo_id": "base",
|
||
"mask": {"entry": False, "exit": False, "stop": False},
|
||
}
|
||
}
|
||
for i, (cid, label, use_e, use_x, use_s) in enumerate(combo_specs):
|
||
n_on = int(use_e) + int(use_x) + int(use_s)
|
||
n_tr = _ob_combo_n_trials(n_on)
|
||
opp.begin_ob_axis(label, i, n_tr)
|
||
combos[cid] = _optimize_combo(
|
||
trades,
|
||
orig_cnt=orig_cnt,
|
||
n_trials=n_tr,
|
||
lg=lg,
|
||
use_e=use_e,
|
||
use_x=use_x,
|
||
use_s=use_s,
|
||
axis_name=label,
|
||
combo_id=cid,
|
||
)
|
||
|
||
def _pick(*ids: str) -> Dict[str, Any]:
|
||
for i in ids:
|
||
c = combos.get(i)
|
||
if isinstance(c, dict) and c.get("ok"):
|
||
return c
|
||
return {"ok": False, "reason": "no_combo", "params": {}, "recommended_stats": orig_stats}
|
||
|
||
# 적용 버튼: 진입=100 · 익절까지=110 스택 · 손절까지=111 스택
|
||
entry_axis = _pick("e")
|
||
exit_axis = _pick("ex", "x") if can_exit_stop else {"ok": False, "reason": "n/a_strategy", "params": {}, "recommended_stats": {}}
|
||
stop_axis = _pick("exs", "es", "xs", "s") if can_exit_stop else {"ok": False, "reason": "n/a_strategy", "params": {}, "recommended_stats": {}}
|
||
|
||
merged_params: Dict[str, Any] = {}
|
||
for ax in (entry_axis, exit_axis, stop_axis):
|
||
if ax.get("ok"):
|
||
merged_params.update(ax.get("params") or {})
|
||
|
||
rec_stats = dict(stop_axis.get("recommended_stats") or exit_axis.get("recommended_stats") or entry_axis.get("recommended_stats") or orig_stats)
|
||
lg.info(
|
||
"⚡ [호가 8방] 전략=%s 모수=%d e=%s ex=%s exs=%s",
|
||
strat_upper,
|
||
orig_cnt,
|
||
"ok" if entry_axis.get("ok") else entry_axis.get("reason"),
|
||
"ok" if exit_axis.get("ok") else exit_axis.get("reason"),
|
||
"ok" if stop_axis.get("ok") else stop_axis.get("reason"),
|
||
)
|
||
out = {
|
||
"ok": any(bool((combos.get(cid) or {}).get("ok")) for cid, *_rest in combo_specs),
|
||
"strategy": strat_upper,
|
||
"ob_table": table,
|
||
"n_trials": sum(int((combos.get(c[0]) or {}).get("n_trials") or 0) for c in combo_specs),
|
||
"trade_count": orig_cnt,
|
||
"orig_stats": orig_stats,
|
||
"recommended_stats": rec_stats,
|
||
"entry": entry_axis,
|
||
"exit": exit_axis,
|
||
"stop": stop_axis,
|
||
"combos": combos,
|
||
"params": merged_params,
|
||
}
|
||
_attach_whipsaw_per_combo(out, trades=trades, strat_upper=strat_upper, lg=lg)
|
||
return out
|
||
|
||
|
||
def _whipsaw_per_combo_enabled() -> bool:
|
||
from kis_trader.utils.env import get_env_bool
|
||
return bool(get_env_bool("OPTUNA_WHIPSAW_PER_COMBO", True))
|
||
|
||
|
||
def _whipsaw_per_combo_n_trials() -> int:
|
||
from kis_trader.utils.env import get_env_int
|
||
return max(10, int(get_env_int("OPTUNA_WHIPSAW_PER_COMBO_TRIALS", 100)))
|
||
|
||
|
||
def _whipsaw_per_combo_min_trades() -> int:
|
||
from kis_trader.utils.env import get_env_int
|
||
return max(1, int(get_env_int("OPTUNA_WHIPSAW_PER_COMBO_MIN_TRADES", 3)))
|
||
|
||
|
||
def _strategy_runs_whipsaw(strat_upper: str) -> bool:
|
||
"""꼬리·돌파는 UI/특성상 휩쏘 후처리 스킵 (기존 whipSkip 과 동일)."""
|
||
u = str(strat_upper or "").strip().upper()
|
||
if u in ("TAIL", "SHORT", "BREAKOUT"):
|
||
return False
|
||
return u in ("MOMENTUM", "SCALP", "SCALPING", "US_MOMENTUM")
|
||
|
||
|
||
def _trade_to_raw_fill(tr: TradeInfo, pnl: float, rate: float) -> Dict[str, Any]:
|
||
return {
|
||
"code": tr.code,
|
||
"name": tr.name,
|
||
"buy_date": tr.buy_dt.strftime("%Y-%m-%d %H:%M:%S") if tr.buy_dt else "",
|
||
"buy_time": tr.buy_dt.strftime("%Y%m%d%H%M%S") if tr.buy_dt else "",
|
||
"buy_price": float(tr.buy_price or 0),
|
||
"actual_pnl": float(pnl),
|
||
"pnl": float(pnl),
|
||
"realized_pnl": float(pnl),
|
||
"profit_rate": float(rate),
|
||
"actual_profit_rate": float(rate),
|
||
}
|
||
|
||
|
||
def _combo_stack_params(combo: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""방 params + mask → _sim_stacked 용."""
|
||
p = dict(combo.get("params") or {})
|
||
mask = combo.get("mask") if isinstance(combo.get("mask"), dict) else {}
|
||
use_e = bool(mask.get("entry")) if mask else bool(p.get("orderbook_filter_enabled"))
|
||
use_x = bool(mask.get("exit")) if mask else bool(p.get("exit_ob_enabled"))
|
||
use_s = bool(mask.get("stop")) if mask else bool(p.get("stop_ob_enabled"))
|
||
p["entry_on"] = use_e
|
||
p["exit_on"] = use_x
|
||
p["stop_on"] = use_s
|
||
p["orderbook_filter_enabled"] = use_e
|
||
p["exit_ob_enabled"] = use_x
|
||
p["stop_ob_enabled"] = use_s
|
||
return p
|
||
|
||
|
||
def _slim_ws_local(rec: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
||
if not isinstance(rec, dict):
|
||
return {"ok": False, "reason": "none"}
|
||
return {
|
||
"ok": bool(rec.get("ok")),
|
||
"reason": rec.get("reason") or "",
|
||
"n_trials": rec.get("n_trials"),
|
||
"trade_count": int(rec.get("trade_count") or 0),
|
||
"params": dict(rec.get("params") or {}),
|
||
"orig_stats": dict(rec.get("orig_stats") or {}),
|
||
"recommended_stats": dict(rec.get("recommended_stats") or {}),
|
||
}
|
||
|
||
|
||
def _attach_whipsaw_per_combo(
|
||
ob_out: Dict[str, Any],
|
||
*,
|
||
trades: List[TradeInfo],
|
||
strat_upper: str,
|
||
lg: logging.Logger,
|
||
) -> None:
|
||
"""
|
||
호가 방마다 통과 체결만으로 휩쏘 TPE.
|
||
- base/진입OFF 방: 전체 체결
|
||
- 진입ON 방: ENTRY_REJECTED 제외, PnL은 스택 시뮬 결과
|
||
"""
|
||
combos = ob_out.get("combos") if isinstance(ob_out.get("combos"), dict) else {}
|
||
if not combos:
|
||
return
|
||
if not _whipsaw_per_combo_enabled():
|
||
for c in combos.values():
|
||
if isinstance(c, dict) and "whipsaw" not in c:
|
||
c["whipsaw"] = {"ok": False, "reason": "per_combo_off"}
|
||
return
|
||
if not _strategy_runs_whipsaw(strat_upper):
|
||
for c in combos.values():
|
||
if isinstance(c, dict):
|
||
c["whipsaw"] = {"ok": False, "reason": "n/a_strategy"}
|
||
return
|
||
|
||
from kis_trader.backtest.optuna_whipsaw_recommend import recommend_whipsaw_parameters
|
||
|
||
n_tr = _whipsaw_per_combo_n_trials()
|
||
min_tr = _whipsaw_per_combo_min_trades()
|
||
mk = "US" if "US" in strat_upper else "KR"
|
||
order = ("base", "e", "x", "s", "ex", "es", "xs", "exs")
|
||
ids = [cid for cid in order if cid in combos] + [c for c in combos if c not in order]
|
||
|
||
lg.info(
|
||
"📡 [휩쏘×호가방] 전략=%s · 방=%d · 방당 trial=%d · min_trades=%d",
|
||
strat_upper, len(ids), n_tr, min_tr,
|
||
)
|
||
try:
|
||
from kis_trader.backtest import optuna_post_progress as opp
|
||
opp.set_ob_axes(len(ids), 0) # 재사용: 축 진행 표시
|
||
except Exception:
|
||
opp = None
|
||
|
||
for i, cid in enumerate(ids):
|
||
c = combos.get(cid)
|
||
if not isinstance(c, dict):
|
||
continue
|
||
label = f"휩쏘@{cid}"
|
||
if opp:
|
||
try:
|
||
opp.begin_ob_axis(label, i, n_tr)
|
||
except Exception:
|
||
pass
|
||
if not c.get("ok") and cid != "base":
|
||
c["whipsaw"] = {"ok": False, "reason": "combo_not_ok"}
|
||
continue
|
||
stack_p = _combo_stack_params(c)
|
||
raw_fills: List[Dict[str, Any]] = []
|
||
for tr in trades:
|
||
pnl, rate, reason = _sim_stacked(tr, stack_p)
|
||
if reason == "ENTRY_REJECTED":
|
||
continue
|
||
raw_fills.append(_trade_to_raw_fill(tr, pnl, rate))
|
||
if len(raw_fills) < min_tr:
|
||
c["whipsaw"] = {
|
||
"ok": False,
|
||
"reason": "not_enough_trades",
|
||
"trade_count": len(raw_fills),
|
||
}
|
||
lg.info("📡 [휩쏘×호가방] %s 스킵 — 통과 %d<%d", cid, len(raw_fills), min_tr)
|
||
continue
|
||
try:
|
||
rec = recommend_whipsaw_parameters(
|
||
strategy=strat_upper,
|
||
n_trials=n_tr,
|
||
log=lg,
|
||
raw_fills=raw_fills,
|
||
market=mk,
|
||
progress_label=label,
|
||
)
|
||
except Exception as exc:
|
||
lg.warning("⚠️ [휩쏘×호가방] %s 실패: %s", cid, exc)
|
||
c["whipsaw"] = {"ok": False, "reason": str(exc)}
|
||
continue
|
||
c["whipsaw"] = _slim_ws_local(rec)
|
||
if rec.get("ok"):
|
||
rs = rec.get("recommended_stats") or {}
|
||
lg.info(
|
||
"📡 [휩쏘×호가방] %s ok · fills=%d · WR=%s PnL=%s",
|
||
cid, len(raw_fills), rs.get("win_rate"), rs.get("pnl"),
|
||
)
|
||
|
||
# 앵커 top-level 호환: base 방 휩쏘
|
||
base_ws = (combos.get("base") or {}).get("whipsaw")
|
||
if isinstance(base_ws, dict):
|
||
ob_out["whipsaw_base"] = base_ws
|
||
|
||
|
||
def attach_orderbook_recommend(
|
||
out_data: Dict[str, Any],
|
||
*,
|
||
log: Optional[logging.Logger] = None,
|
||
) -> Dict[str, Any]:
|
||
"""out_data에 호가 진입/청산 합의 수치 추천(orderbook_recommend)을 첨부."""
|
||
lg = log or logger
|
||
strat = str(out_data.get("strategy") or "MOMENTUM").strip().upper()
|
||
hist = (
|
||
out_data.get("universe_history_source")
|
||
or out_data.get("_universe_history_source")
|
||
or out_data.get("history_source")
|
||
)
|
||
ob_src = out_data.get("ob_source") or out_data.get("orderbook_source")
|
||
rec = recommend_orderbook_parameters(
|
||
strategy=strat,
|
||
n_trials=0,
|
||
history_source=hist,
|
||
ob_source=ob_src,
|
||
log=lg,
|
||
date_from=str(out_data.get("start") or "") or None,
|
||
date_to=str(out_data.get("end") or "") or None,
|
||
)
|
||
out_data["orderbook_recommend"] = rec
|
||
mc = out_data.get("mode_combo")
|
||
if isinstance(mc, dict):
|
||
mc["orderbook_recommend"] = rec
|
||
if not rec.get("ok"):
|
||
lg.info(
|
||
"⚡ [호가 수급 합의 추천] 생략 — %s (table=%s)",
|
||
rec.get("reason") or "n/a",
|
||
rec.get("table") or "?",
|
||
)
|
||
return out_data
|
||
|
||
|
||
def _env_pfx(strat: str) -> str:
|
||
u = str(strat or "").strip().upper()
|
||
if u in ("SHORT", "TAIL"):
|
||
return "TAIL"
|
||
if u in ("SCALPING", "SCALP"):
|
||
return "SCALP"
|
||
return u
|
||
|
||
|
||
def _axis_params(rec: Optional[Dict[str, Any]], axis: str = "") -> Dict[str, Any]:
|
||
if not isinstance(rec, dict):
|
||
return {}
|
||
if axis:
|
||
nested = rec.get(axis)
|
||
if isinstance(nested, dict) and (nested.get("params") or nested.get("ok")):
|
||
return dict(nested.get("params") or {})
|
||
return dict(rec.get("params") or {})
|
||
|
||
|
||
def build_entry_ob_env_patch(rec: Dict[str, Any], strategy: str = "") -> Dict[str, str]:
|
||
"""진입 호가필터만 (*_ORDERBOOK_*)."""
|
||
strat = str(strategy or rec.get("strategy") or "").strip().upper()
|
||
pfx = _env_pfx(strat)
|
||
p = _axis_params(rec, "entry")
|
||
if not pfx or not p:
|
||
return {}
|
||
if p.get("orderbook_max_spread_pct") is None or p.get("orderbook_min_bid_ask_ratio") is None:
|
||
return {}
|
||
out = {
|
||
f"{pfx}_ORDERBOOK_FILTER_ENABLED": "true" if p.get("orderbook_filter_enabled", True) else "false",
|
||
f"{pfx}_ORDERBOOK_MAX_SPREAD_PCT": str(p["orderbook_max_spread_pct"]),
|
||
f"{pfx}_ORDERBOOK_MIN_BID_ASK_RATIO": str(p["orderbook_min_bid_ask_ratio"]),
|
||
}
|
||
if p.get("orderbook_entry_ask_max_mult") is not None:
|
||
out[f"{pfx}_ORDERBOOK_ENTRY_ASK_MAX_MULT"] = str(p["orderbook_entry_ask_max_mult"])
|
||
return out
|
||
|
||
|
||
def build_exit_ob_env_patch(rec: Dict[str, Any], strategy: str = "") -> Dict[str, str]:
|
||
"""익절 호가매도만 (*_EXIT_OB_*). MOMENTUM/BREAKOUT만."""
|
||
strat = str(strategy or rec.get("strategy") or "").strip().upper()
|
||
pfx = _env_pfx(strat)
|
||
if pfx not in ("MOMENTUM", "BREAKOUT"):
|
||
return {}
|
||
p = _axis_params(rec, "exit")
|
||
if not p or "exit_ob_enabled" not in p:
|
||
return {}
|
||
patch = {
|
||
f"{pfx}_EXIT_OB_ENABLED": "true" if p.get("exit_ob_enabled") else "false",
|
||
}
|
||
if p.get("exit_ob_enabled"):
|
||
if p.get("exit_ob_ratio_min") is not None:
|
||
patch[f"{pfx}_EXIT_OB_RATIO_MIN"] = str(p["exit_ob_ratio_min"])
|
||
if p.get("exit_ob_ma_window") is not None:
|
||
patch[f"{pfx}_EXIT_OB_MA_WINDOW"] = str(p["exit_ob_ma_window"])
|
||
if p.get("exit_ob_min_profit_pct") is not None:
|
||
patch[f"{pfx}_EXIT_OB_MIN_PROFIT_PCT"] = str(p["exit_ob_min_profit_pct"])
|
||
if p.get("exit_ob_min_hold_bars") is not None:
|
||
patch[f"{pfx}_EXIT_OB_MIN_HOLD_BARS"] = str(p["exit_ob_min_hold_bars"])
|
||
return patch
|
||
|
||
|
||
def build_stop_ob_env_patch(rec: Dict[str, Any], strategy: str = "") -> Dict[str, str]:
|
||
"""손절 호가 (*_STOP_OB_*). MOMENTUM/BREAKOUT만. 실매 엔진 미변경."""
|
||
strat = str(strategy or rec.get("strategy") or "").strip().upper()
|
||
pfx = _env_pfx(strat)
|
||
if pfx not in ("MOMENTUM", "BREAKOUT"):
|
||
return {}
|
||
p = _axis_params(rec, "stop")
|
||
if not p or "stop_ob_enabled" not in p:
|
||
return {}
|
||
patch = {
|
||
f"{pfx}_STOP_OB_ENABLED": "true" if p.get("stop_ob_enabled") else "false",
|
||
}
|
||
if p.get("stop_ob_enabled"):
|
||
if p.get("stop_ob_ratio_min") is not None:
|
||
patch[f"{pfx}_STOP_OB_RATIO_MIN"] = str(p["stop_ob_ratio_min"])
|
||
if p.get("stop_ob_ma_window") is not None:
|
||
patch[f"{pfx}_STOP_OB_MA_WINDOW"] = str(p["stop_ob_ma_window"])
|
||
if p.get("stop_ob_min_loss_pct") is not None:
|
||
patch[f"{pfx}_STOP_OB_MIN_LOSS_PCT"] = str(p["stop_ob_min_loss_pct"])
|
||
if p.get("stop_ob_min_hold_bars") is not None:
|
||
patch[f"{pfx}_STOP_OB_MIN_HOLD_BARS"] = str(p["stop_ob_min_hold_bars"])
|
||
return patch
|
||
|
||
|
||
def build_orderbook_env_patch(
|
||
rec: Dict[str, Any],
|
||
*,
|
||
include_stop: bool = False,
|
||
) -> Dict[str, str]:
|
||
"""CLI 호환: 진입+익절. STOP은 include_stop=True 일 때만."""
|
||
if not rec or not rec.get("ok"):
|
||
return {}
|
||
strat = str(rec.get("strategy") or "").strip().upper()
|
||
patch: Dict[str, str] = {}
|
||
patch.update(build_entry_ob_env_patch(rec, strat))
|
||
patch.update(build_exit_ob_env_patch(rec, strat))
|
||
if include_stop:
|
||
patch.update(build_stop_ob_env_patch(rec, strat))
|
||
return patch
|