#!/usr/bin/env python3 """ 박스권 돌파(RANGE_BREAK) 시각순 포트폴리오 백테스트. """ from __future__ import annotations from typing import Any, Dict, List, Optional, Tuple from kis_trader.backtest.backtest_portfolio_common import ( flatten_remaining_portfolio_trades, min_invest_ratio_of_slot, portfolio_exposure_krw, target_qty_and_cost, ) from kis_trader.engine.range_break_engine import ( check_sell_signal_range_break_live, range_break_min_bars_required, range_break_scan_buy_at_bar, ) from kis_trader.engine.scalping_engine import _t2dt, _to_bool from kis_trader.engine.tick_exit_common import ( backtest_sell_slip_pct, backtest_tick_poll_ms, collect_minute_ticks, resolve_backtest_sell, strategy_tick_fallback_ohlc, strategy_use_tick_exit, ) from kis_trader.share.stock_share import share_denom_for_code from kis_trader.strategies.breakout import ( _bt_slot_key, breakout_invest_amount_krw, normalize_breakout_max_loss_krw, ) def _buy_priority_key( code: str, slot_key: str, universe_by_slot: Optional[Dict[str, List[str]]], ) -> Tuple[int, str]: if universe_by_slot is None: return (0, code) lst = universe_by_slot.get(slot_key) or [] try: return (lst.index(code), code) except ValueError: return (999999, code) def _max_stocks_from_params(params: Dict[str, Any]) -> int: for key in ("max_stocks", "range_break_max_stocks"): v = params.get(key) if v not in (None, "", 0): return max(1, int(v)) try: from kis_trader.utils.env import get_env_int n = get_env_int("RANGE_BREAK_MAX_STOCKS", 0) or get_env_int("MAX_STOCKS", 3) return max(1, int(n)) except Exception: return 3 def _total_budget_from_params(params: Dict[str, Any]) -> float: for key in ("total_budget_krw", "range_break_total_budget_krw"): v = params.get(key) if v not in (None, ""): try: return float(v) except (TypeError, ValueError): pass try: from kis_trader.utils.env import get_env_int cap = get_env_int("RANGE_BREAK_TOTAL_BUDGET_KRW", 0) if cap > 0: return float(cap) except Exception: pass return 0.0 def _resolve_invest_cap(params: Dict[str, Any]) -> float: slot_money = float(params.get("slot_money", 200_000)) if params.get("stop_loss_pct") not in (None, ""): sl_pct_ui = abs(float(params["stop_loss_pct"])) * 100.0 else: raw = abs(float(params.get("sl_pct", 3.0))) sl_pct_ui = raw if raw >= 0.5 else raw * 100.0 max_loss_krw = normalize_breakout_max_loss_krw(params.get("max_loss_krw", 200_000)) return breakout_invest_amount_krw(max_loss_krw, sl_pct_ui, slot_money) def run_range_break_backtest_portfolio( codes_candles: Dict[str, List[Dict]], params: Dict[str, Any], universe_by_slot: Optional[Dict[str, List[str]]] = None, ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]] = None, ) -> List[Dict]: """ 시각순 포트폴리오 박스권 돌파 백테스트. - 매도: 틱 우선 ``resolve_backtest_sell`` → ``check_sell_signal_range_break_live`` """ min_bars = range_break_min_bars_required(params) force_eod_exit = _to_bool(params.get("force_eod_exit"), False) cooldown_min = float(params.get("cooldown_min", 30)) max_daily = int(params.get("max_daily", 1)) max_stocks = _max_stocks_from_params(params) slot_money = float(params.get("slot_money", 200_000)) total_budget = _total_budget_from_params(params) if total_budget <= 0: total_budget = float(max_stocks * slot_money) min_invest_ratio = min_invest_ratio_of_slot(params, strategy="RANGE_BREAK") invest_cap = _resolve_invest_cap(params) buy_params = dict(params) skipped_micro_buys = 0 use_tick_exit = bool(ticks_by_code) and strategy_use_tick_exit( params, "RANGE_BREAK_BACKTEST_USE_TICK_EXIT", default=True, ) tick_fallback_ohlc = strategy_tick_fallback_ohlc( params, "RANGE_BREAK_BACKTEST_TICK_FALLBACK_OHLC", default=False, ) tick_poll_ms = backtest_tick_poll_ms(params, strategy_env="RANGE_BREAK_BACKTEST_POLL_MS") tick_sell_slip = backtest_sell_slip_pct( params, strategy_env="RANGE_BREAK_BACKTEST_SELL_SLIP_PCT", ) tick_exit_count = 0 ohlc_exit_count = 0 ctx_by_code: Dict[str, Dict[str, Any]] = {} all_times_set = set() for code, raw_rows in codes_candles.items(): if len(raw_rows) < min_bars: continue candles = [dict(r) for r in raw_rows] day_open_arr: List[float] = [0.0] * len(candles) _cur_day = None _cur_open = 0.0 _first_open = float(candles[0].get("open") or 0) if candles else 0.0 for _idx, _c in enumerate(candles): _d = str(_c.get("candle_time") or "")[:8] if _d != _cur_day: _cur_day = _d _cur_open = float(_c.get("open") or 0) day_open_arr[_idx] = _cur_open if _cur_open > 0 else _first_open ctx_by_code[code] = { "code": code, "candles": candles, "time_index": {c["candle_time"]: idx for idx, c in enumerate(candles)}, "day_open_arr": day_open_arr, "last_exit_dt": {}, "daily_cnt": {}, "pending_entry": None, } for c in candles: all_times_set.add(c["candle_time"]) all_times = sorted(all_times_set) portfolio: Dict[str, Dict[str, Any]] = {} all_trades: List[Dict] = [] from kis_trader.engine.scalping_engine import check_sell_signal_backtest_bar from kis_trader.backtest.backtest_env_timeline import apply_env_timeline_at for t in all_times: if apply_env_timeline_at(params, t, "RANGE_BREAK"): max_stocks = _max_stocks_from_params(params) slot_money = float(params.get("slot_money", 200_000)) total_budget = _total_budget_from_params(params) if total_budget <= 0: total_budget = float(max_stocks * slot_money) invest_cap = _resolve_invest_cap(params) cooldown_min = float(params.get("cooldown_min", 30)) max_daily = int(params.get("max_daily", 1)) buy_params.update({ k: params[k] for k in params if k not in buy_params or buy_params.get(k) != params[k] }) slot_key = _bt_slot_key(t, int(params.get("scan_interval_min", 1))) pending_codes = [ code for code, ctx in ctx_by_code.items() if ctx.get("pending_entry") and ctx["pending_entry"].get("entry_time") == t ] pending_codes.sort(key=lambda c: _buy_priority_key(c, slot_key, universe_by_slot)) for code in pending_codes: ctx = ctx_by_code[code] pe = ctx.pop("pending_entry", None) if not pe or code in portfolio: continue if len(portfolio) >= max_stocks: break entry_price = float(pe["entry_price"]) box_stop = float(pe.get("box_stop_line", 0) or 0) if entry_price <= 0: continue exposure = portfolio_exposure_krw(portfolio) remaining = max(0.0, total_budget - exposure) target_qty, target_cost = target_qty_and_cost(entry_price, invest_cap) min_required = target_cost * min_invest_ratio if target_qty < 1 or remaining < min_required: skipped_micro_buys += 1 continue invest = min(invest_cap, remaining, target_cost) qty = int(invest / entry_price) if qty < 1: skipped_micro_buys += 1 continue cost = qty * entry_price if cost < min_required: skipped_micro_buys += 1 continue if exposure + cost > total_budget + 1e-6: skipped_micro_buys += 1 continue portfolio[code] = { "entry_price": entry_price, "entry_time": t, "qty": qty, "max_price": entry_price, "box_stop_line": box_stop, } ctx["daily_cnt"][t[:8]] = ctx["daily_cnt"].get(t[:8], 0) + 1 break for code in list(portfolio.keys()): ctx = ctx_by_code.get(code) if ctx is None: continue idx = ctx["time_index"].get(t) if idx is None: continue candles = ctx["candles"] c = candles[idx] day = t[:8] cl = float(c["close"]) is_eod_raw = (idx == len(candles) - 1) or (candles[idx + 1]["candle_time"][:8] != day) is_eod = is_eod_raw and force_eod_exit pos = portfolio[code] if t == pos["entry_time"]: continue bar = dict(c) if "open" not in bar or bar.get("open") in (None, ""): bar["open"] = float(c.get("open") or cl) if use_tick_exit: minute_ticks = collect_minute_ticks(ticks_by_code, code, t) res5 = resolve_backtest_sell( pos, bar, params, is_eod=is_eod, sell_fn=check_sell_signal_range_break_live, low_mode="current", ticks=minute_ticks, use_tick_exit=use_tick_exit, tick_fallback_ohlc=tick_fallback_ohlc, poll_ms=tick_poll_ms, slip_pct=tick_sell_slip, ) if not res5: continue reason, exit_price, sell_time, _hold_min, exit_src = res5 if exit_src == "ws_ticks": tick_exit_count += 1 else: ohlc_exit_count += 1 else: res = check_sell_signal_backtest_bar( pos, bar, params, is_eod=is_eod, sell_fn=check_sell_signal_range_break_live, low_mode="current", ) if not res: continue reason, exit_price = res sell_time = t ohlc_exit_count += 1 all_trades.append({ "code": code, "buy_time": pos["entry_time"], "sell_time": sell_time or t, "buy_price": pos["entry_price"], "sell_price": round(exit_price, 2), "qty": pos.get("qty", 1), "pnl": 0, "sell_reason": reason, "hold_min": 0, }) ctx["last_exit_dt"][day] = _t2dt(sell_time or t) del portfolio[code] if len(portfolio) >= max_stocks: continue exposure = portfolio_exposure_krw(portfolio) if exposure >= total_budget - 1e-6: continue candidates: List[Tuple[Tuple[int, str], str, Dict[str, Any]]] = [] for code, ctx in ctx_by_code.items(): if code in portfolio or ctx.get("pending_entry"): continue idx = ctx["time_index"].get(t) if idx is None: continue candles = ctx["candles"] c = candles[idx] day = t[:8] cl = float(c["close"]) if cl <= 0: continue if universe_by_slot is not None: if code not in universe_by_slot.get(slot_key, []): continue if day in ctx["last_exit_dt"]: elapsed = (_t2dt(t) - ctx["last_exit_dt"][day]).total_seconds() / 60 if elapsed < cooldown_min: continue if ctx["daily_cnt"].get(day, 0) >= max_daily: continue state = { "last_exit_dt": ctx["last_exit_dt"].get(day), "daily_cnt": ctx["daily_cnt"].get(day, 0), } code_buy = dict(buy_params) code_buy["share_denom"] = share_denom_for_code(buy_params, code) _reason, _msg, signal, entry_price, entry_time = range_break_scan_buy_at_bar( candles, idx, code_buy, state=state, ) if not signal or entry_price <= 0 or not entry_time: continue if entry_time[:8] != day: continue pri = _buy_priority_key(code, slot_key, universe_by_slot) pe = { "entry_time": entry_time, "entry_price": entry_price, "box_stop_line": float(signal.get("box_stop_line", signal.get("box_high", 0)) or 0), } if entry_time == t and code not in portfolio: exposure = portfolio_exposure_krw(portfolio) remaining = max(0.0, total_budget - exposure) target_qty, target_cost = target_qty_and_cost(entry_price, invest_cap) min_required = target_cost * min_invest_ratio if ( len(portfolio) < max_stocks and target_qty >= 1 and remaining >= min_required and exposure + target_cost <= total_budget + 1e-6 ): invest = min(invest_cap, remaining, target_cost) qty = int(invest / entry_price) if qty < 1: continue cost = qty * entry_price if cost >= min_required: portfolio[code] = { "entry_price": entry_price, "entry_time": t, "qty": qty, "max_price": entry_price, "box_stop_line": pe["box_stop_line"], } ctx["daily_cnt"][day] = ctx["daily_cnt"].get(day, 0) + 1 continue candidates.append((pri, code, pe)) if not candidates: continue candidates.sort(key=lambda x: x[0]) _pri, pick_code, pe = candidates[0] ctx_by_code[pick_code]["pending_entry"] = pe skip_stats: Dict[str, Any] = {} if skipped_micro_buys: skip_stats["skipped_micro_buys"] = skipped_micro_buys if tick_exit_count or ohlc_exit_count: skip_stats["tick_exit_count"] = tick_exit_count skip_stats["ohlc_exit_count"] = ohlc_exit_count flat_n = flatten_remaining_portfolio_trades( portfolio, ctx_by_code, all_trades, params=params, strategy="RANGE_BREAK", ) if flat_n: skip_stats["bt_flatten_count"] = flat_n if skip_stats: params["_portfolio_skip_stats"] = skip_stats all_trades.sort(key=lambda x: x["sell_time"]) return all_trades