#!/usr/bin/env python3 """ 박스권 돌파(RANGE_BREAK) 백테스트 공통 로더 — backtest_web / param_search 단일 진입점. """ from __future__ import annotations from typing import Any, Dict, List, Optional, Tuple from kis_trader.backtest.backtest_portfolio_common import ( attach_scalp_trade_pnl, backtest_slip_pct, build_budget_warning, fee_and_slot_from_env_row, merge_portfolio_into_params, min_invest_ratio_of_slot, resolve_portfolio_params, summarize_trades, ) from kis_trader.backtest.breakout_tick_loader import ( load_breakout_ticks_by_code, tick_coverage_stats, ) from kis_trader.engine.range_break_engine import ( RANGE_BREAK_STRATEGY_ID, range_break_min_bars_required, run_range_break_backtest, ) from kis_trader.engine.tick_exit_common import strategy_use_tick_exit RANGE_BREAK_STRATEGY_ID = RANGE_BREAK_STRATEGY_ID # noqa: F811 — re-export 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_range_break_universe( start_ymd: str, end_ymd: str, *, use_saved_history: bool, strategy_id: str = RANGE_BREAK_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 from kis_trader.backtest.universe_timeline import ( universe_exit_debounce_sec_for_strategy, ) history = _get_ext_db().get_universe_by_candle_time( strategy_id=strategy_id, start_ymd=start_ymd, end_ymd=end_ymd, exit_debounce_sec=universe_exit_debounce_sec_for_strategy(strategy_id), ) if history: return history, "history", len(history), 1 except Exception: pass return None, "all", 0, 1 def load_range_break_candles_by_code( db, start_key: str, end_key: str, params: Optional[Dict[str, Any]] = None, ) -> Tuple[Dict[str, List[Dict]], int]: """ws_candles 1분봉 전 종목 로드.""" p = dict(params or {}) min_bars = range_break_min_bars_required(p) codes_raw = db.conn.execute( "SELECT DISTINCT code FROM ws_candles WHERE timeframe=1 " "AND candle_time >= %s AND candle_time <= %s ORDER BY code", [start_key, end_key], ).fetchall() codes = [r["code"] for r in codes_raw] candles_by_code: Dict[str, List[Dict]] = {} total_candles = 0 for code in codes: rows = db.conn.execute( "SELECT candle_time, open, high, low, close, volume " "FROM ws_candles WHERE timeframe=1 AND code=%s " "AND candle_time >= %s AND candle_time <= %s AND is_confirmed=1 " "ORDER BY candle_time ASC", [code, start_key, end_key], ).fetchall() if len(rows) < min_bars: continue candles_by_code[code] = [dict(r) for r in rows] total_candles += len(rows) return candles_by_code, total_candles def run_range_break_backtest_web_aligned( candles_by_code: Dict[str, List[Dict]], params: Dict[str, Any], universe_by_slot: Optional[Dict[str, List[str]]], *, slot_money: float, fee_rate: float, sell_tax: float, max_stocks: Optional[int] = None, total_budget_krw: Optional[float] = None, meta_out: Optional[Dict[str, Any]] = None, ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]] = None, ) -> List[Dict]: engine_params = dict(params) engine_params["slot_money"] = float(slot_money) if max_stocks is not None: engine_params["max_stocks"] = int(max_stocks) if total_budget_krw is not None: tb = float(total_budget_krw) engine_params["total_budget_krw"] = tb if tb > 0 else float( int(engine_params.get("max_stocks") or 3) * slot_money ) if universe_by_slot is not None: engine_params.setdefault("scan_interval_min", 1) engine_params.setdefault("portfolio_mode", True) from kis_trader.backtest.backtest_env_timeline import attach_backtest_env_timeline_to_params attach_backtest_env_timeline_to_params(engine_params, meta_out, "RANGE_BREAK") loaded_ticks: Dict[str, Dict[str, List[Dict]]] = dict(ticks_by_code or {}) tick_meta: Dict[str, Any] = {} if strategy_use_tick_exit(engine_params, "RANGE_BREAK_BACKTEST_USE_TICK_EXIT", default=True): if not loaded_ticks and meta_out is not None: start_key = str(meta_out.get("start_key") or "") end_key = str(meta_out.get("end_key") or "") db = meta_out.get("db") if db and start_key and end_key: loaded_ticks, tick_rows = load_breakout_ticks_by_code( db, start_key, end_key, set(candles_by_code.keys()), ) tick_meta = tick_coverage_stats(candles_by_code, loaded_ticks) tick_meta["ws_tick_rows_loaded"] = tick_rows if tick_rows <= 0: from kis_trader.utils.logger import get_logger as _get_logger _get_logger("kis_trader.range_break_backtest").warning( "⚠️ ws_ticks 데이터 없음 — RANGE_BREAK 틱 청산 스킵 " "(틱 수집 후 재백테 권장, FALLBACK_OHLC=1 시 OHLC)", ) elif loaded_ticks: tick_meta = tick_coverage_stats(candles_by_code, loaded_ticks) tick_meta["ws_tick_rows_loaded"] = sum( len(lst) for cm in loaded_ticks.values() for lst in cm.values() ) trades = run_range_break_backtest( candles_by_code, engine_params, universe_by_slot=universe_by_slot, ticks_by_code=loaded_ticks or None, ) attach_scalp_trade_pnl( trades, fee_rate=fee_rate, sell_tax=sell_tax, slip_pct=backtest_slip_pct(engine_params), ) if meta_out is not None: skip_stats = engine_params.get("_portfolio_skip_stats") or {} meta_out["skip_stats"] = dict(skip_stats) meta_out["engine_params"] = engine_params if tick_meta: from kis_trader.backtest.breakout_tick_loader import enrich_tick_meta_with_traded_codes tick_meta = enrich_tick_meta_with_traded_codes( tick_meta, candles_by_code, loaded_ticks, trades, ) meta_out["tick_backtest"] = tick_meta if tick_meta.get("ws_tick_rows_loaded", 0) > 0: meta_out["backtest_buy_source"] = "ws_ticks" else: meta_out["backtest_buy_source"] = "ohlc_fallback" else: meta_out["backtest_buy_source"] = "align" return trades def resolve_range_break_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]: return resolve_portfolio_params( env_row, base_defaults, strategy="RANGE_BREAK", slot_money=slot_money, max_stocks=max_stocks, total_budget_krw=total_budget_krw, ) def merge_range_break_portfolio_into_params( params: Dict[str, Any], portfolio: Dict[str, Any], ) -> Dict[str, Any]: return merge_portfolio_into_params(params, portfolio) def build_range_break_budget_warning( portfolio: Dict[str, Any], skip_stats: Optional[Dict[str, Any]] = None, ) -> Optional[str]: ratio = min_invest_ratio_of_slot({}, strategy="RANGE_BREAK") return build_budget_warning(portfolio, skip_stats, min_invest_ratio=ratio) def summarize_range_break_trades( trades: List[Dict], *, total_budget_krw: float, period_days: int = 1, ) -> Dict[str, Any]: return summarize_trades( trades, total_budget_krw=total_budget_krw, period_days=period_days, ) def fee_and_slot_from_env( row: Optional[Dict[str, Any]], ) -> Tuple[float, float, float]: return fee_and_slot_from_env_row(row, strategy="RANGE_BREAK")