kis_bot/kis_trader/backtest/range_break_portfolio_backtest.py

#!/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 (
    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.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))
    sl_pct = abs(float(params.get("stop_loss_pct", params.get("sl_pct", -0.03))))
    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 * 100.0, 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,
) -> List[Dict]:
    """시각순 포트폴리오 박스권 돌파 백테스트."""
    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

    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

    for t in all_times:
        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)
            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
            all_trades.append({
                "code": code,
                "buy_time": pos["entry_time"],
                "sell_time": 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(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

    if skipped_micro_buys:
        params["_portfolio_skip_stats"] = {"skipped_micro_buys": skipped_micro_buys}
    all_trades.sort(key=lambda x: x["sell_time"])
    return all_trades