#!/usr/bin/env python3 """ 더블 볼린저 백테스트 공통 로더 — backtest_web(api/backtest/dbband) 와 dbband_param_search 공통. """ from __future__ import annotations from datetime import datetime from typing import Any, Dict, List, Optional, Tuple import kis_trader.engine.dbband_engine as bbe DBBAND_STRATEGY_ID = "DBBAND" VALID_TIMEFRAMES = (3, 5, 15, 60) def date_keys(start: str, end: str) -> Tuple[str, str, str, str]: start_key = start.replace("-", "") + "0000" end_key = end.replace("-", "") + "2359" return start_key, end_key, start_key[:8], end_key[:8] def resolve_dbband_universe( start_ymd: str, end_ymd: str, *, use_saved_history: bool, strategy_id: str = DBBAND_STRATEGY_ID, ) -> Tuple[Optional[Dict[str, List[str]]], str, int, int]: if use_saved_history and strategy_id: try: from kis_trader.database.db_manager import get_db as _get_ext_db history = _get_ext_db().get_universe_by_candle_time( strategy_id=strategy_id, start_ymd=start_ymd, end_ymd=end_ymd, ) if history: return history, "history", len(history), 1 except Exception: pass return None, "all", 0, 1 def load_dbband_candles_by_code( db, start_key: str, end_key: str, timeframe: int, trend_ma_period: int = 200, ) -> Tuple[Dict[str, List[Dict]], int]: tf = int(timeframe) if tf not in VALID_TIMEFRAMES: raise ValueError(f"timeframe 은 {VALID_TIMEFRAMES} 중 하나여야 합니다") codes_raw = db.conn.execute( "SELECT DISTINCT code FROM ws_candles WHERE timeframe=%s " "AND candle_time >= %s AND candle_time <= %s ORDER BY code", [tf, start_key, end_key], ).fetchall() codes = [r["code"] for r in codes_raw] min_bars = max(int(trend_ma_period) + 10, 50) candles_by_code: Dict[str, List[Dict]] = {} total = 0 for code in codes: rows = db.conn.execute( "SELECT candle_time, open, high, low, close, volume " "FROM ws_candles WHERE timeframe=%s AND code=%s " "AND candle_time >= %s AND candle_time <= %s AND is_confirmed=1 " "ORDER BY candle_time ASC", [tf, code, start_key, end_key], ).fetchall() if len(rows) < min_bars: continue candles_by_code[code] = [dict(r) for r in rows] total += len(rows) return candles_by_code, total def _t2dt(candle_time: str) -> datetime: return datetime.strptime(str(candle_time)[:12], "%Y%m%d%H%M") def normalize_stored_min_candles(candles: List[Dict]) -> List[Dict]: """holding_min_candles(candle_date) → 엔진용 candle_time(YYYYMMDDHHMM).""" out: List[Dict] = [] for c in candles: d = dict(c) if not d.get("candle_time"): raw = d.get("candle_date") or d.get("candle_time_str") or "" s = str(raw).strip() if len(s) >= 19 and (" " in s or "-" in s[:5]): d["candle_time"] = s.replace("-", "").replace(" ", "").replace(":", "")[:12] else: digits = "".join(ch for ch in s if ch.isdigit()) d["candle_time"] = digits[:12] if digits else s out.append(d) return out def attach_dbband_trade_pnl( trades: List[Dict], *, slot_money: float, fee_rate: float, sell_tax: float, slip_pct: float = 0.0, ) -> None: """더블BB 손익. slip_pct: 백테 체결 슬리피지(편도 %) — 진입·청산 모두 불리. 숏은 진입=매도(낮게 -)·청산=매수(높게 +), 롱은 반대. 표시가는 그대로, pnl 만 반영. 0=OFF(동작 불변). """ slip = max(0.0, float(slip_pct or 0.0)) / 100.0 for t in trades: qty = t.get("qty") if qty is None: qty = max(1, int(slot_money / max(1, t["entry"]))) t["qty"] = qty ep = float(t["entry"]) xp = float(t["exit"]) side = str(t.get("side") or "long").lower() if slip > 0: if side == "short": ep = ep * (1.0 - slip) # 숏 진입(매도) 불리: 더 낮게 체결 xp = xp * (1.0 + slip) # 숏 청산(매수) 불리: 더 높게 체결 else: ep = ep * (1.0 + slip) # 롱 진입(매수) 불리: 더 높게 체결 xp = xp * (1.0 - slip) # 롱 청산(매도) 불리: 더 낮게 체결 fee = (ep + xp) * qty * fee_rate tax = xp * qty * sell_tax if side == "short": t["pnl"] = round((ep - xp) * qty - fee - tax) else: t["pnl"] = round((xp - ep) * qty - fee - tax) t["hold_min"] = round( (_t2dt(t["exit_time"]) - _t2dt(t["entry_time"])).total_seconds() / 60, 1, ) def fee_and_slot_from_env_row(row: Optional[Dict[str, Any]]) -> Tuple[float, float, float]: if not row: return 0.015 / 100, 0.18 / 100, 3_000_000.0 r = dict(row) fee_rate = float(r.get("FEE_RATE_PCT") or 0.015) / 100 sell_tax = float(r.get("SELL_TAX_RATE_PCT") or 0.18) / 100 slot = float(r.get("DBBAND_SLOT_MONEY") or r.get("SLOT_MONEY_DEFAULT") or 3_000_000) return fee_rate, sell_tax, slot def _pct_ui(v: Any, default: float = 0.0) -> float: """엔진 비율(0.02) 또는 퍼센트(2.0) → UI 퍼센트.""" try: x = float(v) return round(x * 100, 3) if 0 < abs(x) < 1 else round(x, 3) except (TypeError, ValueError): return default def _period_days(start: str, end: str, fallback: int = 1) -> int: try: s = (start or "").replace("-", "")[:8] e = (end or "").replace("-", "")[:8] if len(s) == 8 and len(e) == 8: d0 = datetime.strptime(s, "%Y%m%d") d1 = datetime.strptime(e, "%Y%m%d") return max(1, (d1 - d0).days + 1) except (TypeError, ValueError): pass return max(1, int(fallback)) def cfg_to_ui_params( cfg: Dict[str, Any], *, code: str, name: str, start: str, end: str, tf: int, candle_count: int, param_source: str, portfolio: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: pf = dict(portfolio or {}) slot = float(cfg.get("slot_money") or pf.get("slot_money") or 3_000_000) mxs = int(cfg.get("max_stocks") or pf.get("max_stocks") or 1) tb = float(cfg.get("total_budget_krw") or pf.get("total_budget_krw") or 0) if tb <= 0: tb = float(mxs * slot) utf = cfg.get("use_trend_filter", True) trend_on = utf in (True, 1, "1", "true", "yes", "on") return { "code": code, "name": name or code, "start": start, "end": end, "timeframe": int(tf), "candle_count": int(candle_count), "param_source": param_source, "universe_source": "single_stock", "bb_period": int(cfg.get("bb_period") or 20), "bb_inner_std": float(cfg.get("bb_inner_std") or 2.0), "bb_outer_std": float(cfg.get("bb_outer_std") or 3.0), "trend_ma_period": int(cfg.get("trend_ma_period") or 200), "use_trend_filter": trend_on, "sl_pct": _pct_ui(cfg.get("sl_pct"), 2.0), "tp_pct": _pct_ui(cfg.get("tp_pct"), 3.0), "tp_mode": str(cfg.get("tp_mode") or "opposite_band"), "rr_ratio": float(cfg.get("rr_ratio") or 2.0), "exit_mode": str(cfg.get("exit_mode") or "classic"), "shoulder_min_high": _pct_ui(cfg.get("shoulder_min_high"), 0.3), "shoulder_cut_pct": _pct_ui(cfg.get("shoulder_cut_pct"), 0.2), "entry_valid_bars": int(cfg.get("entry_valid_bars") or 3), "max_hold_bars": int(cfg.get("max_hold_bars") or 16), "time_start_hm": int(cfg.get("time_start_hm") or 930), "time_end_hm": int(cfg.get("time_end_hm") or 1500), "cooldown_min": float(cfg.get("cooldown_min") or 15.0), "max_daily": int(cfg.get("max_daily") or 3), "slot_money": int(slot), "max_stocks": mxs, "total_budget_krw": int(tb), "side_mode": str(cfg.get("side_mode") or "long_only"), } def build_dbband_backtest_report( trades: List[Dict], candles: List[Dict], cfg: Dict[str, Any], *, code: str, name: str, start_date: str, end_date: str, tf: int, candle_count: int, param_source: str, portfolio: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """모멘텀·꼬리잡기 웹과 동일 지표 + Buy&Hold 벤치마크.""" from kis_trader.backtest.backtest_portfolio_common import summarize_trades pf = dict(portfolio or {}) slot = float(cfg.get("slot_money") or pf.get("slot_money") or 3_000_000) tb = float(cfg.get("total_budget_krw") or pf.get("total_budget_krw") or 0) if tb <= 0: mxs = int(cfg.get("max_stocks") or pf.get("max_stocks") or 1) tb = float(mxs * slot) period_days = _period_days(start_date, end_date, fallback=1) base = summarize_trades(trades, total_budget_krw=tb, period_days=period_days) wins = [t for t in trades if int(t.get("pnl") or 0) > 0] losses = [t for t in trades if int(t.get("pnl") or 0) < 0] win_pnl = sum(int(t.get("pnl") or 0) for t in wins) loss_pnl = sum(int(t.get("pnl") or 0) for t in losses) profit_factor = round(abs(win_pnl / loss_pnl), 2) if loss_pnl != 0 else 9999.0 peak, mdd, cum = 0.0, 0.0, 0.0 equity: List[Dict[str, Any]] = [] daily_map: Dict[str, int] = {} for t in sorted(trades, key=lambda x: str(x.get("exit_time") or "")): pnl = int(t.get("pnl") or 0) cum += pnl if cum > peak: peak = cum dd = peak - cum if dd > mdd: mdd = dd et = str(t.get("exit_time") or "") day = et[:8] if len(day) == 8: equity.append({ "date": f"{day[:4]}-{day[4:6]}-{day[6:]}", "cum_pnl": round(cum), "pnl": pnl, }) daily_map[day] = daily_map.get(day, 0) + pnl daily_list = [ {"date": f"{d[:4]}-{d[4:6]}-{d[6:]}", "pnl": round(v)} for d, v in sorted(daily_map.items()) ] reasons: Dict[str, int] = {} for t in trades: r = str(t.get("reason") or "기타") reasons[r] = reasons.get(r, 0) + 1 closes = [float(c.get("close") or 0) for c in candles if float(c.get("close") or 0) > 0] if len(closes) >= 2 and closes[0] > 0: bnh_pct = round((closes[-1] - closes[0]) / closes[0] * 100, 2) bnh_pnl = round(slot * bnh_pct / 100) else: bnh_pct = 0.0 bnh_pnl = 0 bot_pct = float(base.get("bot_pct") or 0) alpha_pct = round(bot_pct - bnh_pct, 2) budget_warning = pf.get("budget_warning") if int(cfg.get("max_stocks") or pf.get("max_stocks") or 1) > 1: budget_warning = ( budget_warning or "" ) + (" | " if budget_warning else "") + ( "DBBAND 종목별 백테: 동시보유>1 설정은 이 화면에서 1종목 시뮬입니다." ) summary = { "total_trades": base.get("total_trades", 0), "win_trades": len(wins), "loss_trades": len(losses), "win_rate": base.get("win_rate", 0), "total_pnl": base.get("total_pnl", 0), "avg_hold_min": base.get("avg_hold_min", 0), "profit_factor": profit_factor, "max_drawdown": round(mdd), "bot_pct": bot_pct, "daily_avg_pct": base.get("daily_avg_pct", 0), "backtest_days": period_days, "bnh_pct": bnh_pct, "bnh_pnl": bnh_pnl, "alpha_pct": alpha_pct, "budget_warning": budget_warning, "sell_reasons": reasons, } params_ui = cfg_to_ui_params( cfg, code=code, name=name, start=start_date, end=end_date, tf=tf, candle_count=candle_count, param_source=param_source, portfolio=pf, ) return { "params": params_ui, "summary": summary, "equity": equity, "daily": daily_list, "reasons": reasons, } def resolve_dbband_portfolio_params( env_row: Optional[Dict[str, Any]], base_defaults: Optional[Dict[str, Any]] = None, *, slot_money: Optional[float] = None, max_stocks: Optional[int] = None, total_budget_krw: Optional[float] = None, ) -> Dict[str, Any]: base = dict(base_defaults or bbe.get_dbband_defaults_from_db()) if slot_money is not None: base["slot_money"] = float(slot_money) if max_stocks is not None: base["max_stocks"] = int(max_stocks) if total_budget_krw is not None: base["total_budget_krw"] = float(total_budget_krw) if env_row: r = dict(env_row) if slot_money is None and r.get("DBBAND_SLOT_MONEY"): base["slot_money"] = float(r["DBBAND_SLOT_MONEY"]) if max_stocks is None and r.get("DBBAND_MAX_STOCKS"): base["max_stocks"] = int(r["DBBAND_MAX_STOCKS"]) if total_budget_krw is None and r.get("DBBAND_TOTAL_BUDGET_KRW"): base["total_budget_krw"] = float(r["DBBAND_TOTAL_BUDGET_KRW"]) from kis_trader.backtest.backtest_portfolio_common import resolve_portfolio_params return resolve_portfolio_params( env_row, base, strategy="DBBAND", slot_money=slot_money, max_stocks=max_stocks, total_budget_krw=total_budget_krw, )