kis_bot/kis_trader/backtest/momentum_portfolio_backtest.py

#!/usr/bin/env python3
"""
모멘텀 시각순 포트폴리오 백테스트 — tail/breakout 과 동일 구조.

청산: ws_ticks 틱 리플레이. 진입: live_align(T-1신호→T시가) + ws_ticks 첫 체결.
"""
from __future__ import annotations

from typing import Any, Dict, List, Optional, Set, Tuple

from datetime import datetime, timedelta

from kis_trader.backtest.backtest_portfolio_common import (
    attach_scalp_trade_pnl,
    backtest_slip_pct,
    min_invest_ratio_of_slot,
    portfolio_exposure_krw,
    target_qty_and_cost,
)
from kis_trader.engine.momentum_engine import (
    MOMENTUM_STRATEGY_ID,
    _slot_key,
    _t2dt,
    _to_bool,
    effective_tp_pct_from_params,
    eval_momentum_buy_at_index,
)
from kis_trader.engine.indicator_cache import (
    attach_indicator_caches_to_params,
    get_indicator_cache_from_params,
)
from kis_trader.engine.whipsaw_filter import inject_whipsaw_ticks_into_params
from kis_trader.backtest.trigger_snapshot_loader import inject_trigger_snapshots_into_params
from kis_trader.engine.momentum_tick_replay import (
    align_momentum_entry_from_ticks,
    collect_minute_ticks,
    momentum_backtest_live_scan_queue_enabled,
    momentum_backtest_scan_sec,
    momentum_backtest_use_tick_exit,
    momentum_live_align_enabled,
    resolve_momentum_sell_for_bar,
    try_momentum_sell_on_ticks,
)
from kis_trader.backtest.momentum_universe_timeline import (
    MomentumUniverseTimeline,
    momentum_backtest_universe_scan_at_enabled,
)


def _buy_priority_key(
    code: str,
    slot_key: str,
    universe_by_slot: Optional[Dict[str, List[str]]],
    universe_codes: Optional[List[str]] = None,
) -> Tuple[int, str]:
    if universe_codes is not None:
        try:
            return (universe_codes.index(code), code)
        except ValueError:
            return (999999, code)
    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", "momentum_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("MOMENTUM_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", "momentum_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("MOMENTUM_TOTAL_BUDGET_KRW", 0)
        if cap > 0:
            return float(cap)
    except Exception:
        pass
    return 0.0


def _resolve_invest_cap_krw(params: Dict[str, Any], slot_money: float) -> float:
    sl_pct = abs(float(params.get("sl_pct", 0.015)))
    max_loss_krw = float(params.get("max_loss_krw", 200_000.0))
    invest_amount = float(slot_money)
    if max_loss_krw > 0 and sl_pct > 0:
        invest_amount = min(max_loss_krw / sl_pct, float(slot_money))
    return invest_amount


def _try_open_momentum_position(
    portfolio: Dict[str, Dict[str, Any]],
    code: str,
    pe: Dict[str, Any],
    *,
    invest_cap: float,
    total_budget: float,
    min_invest_ratio: float,
    max_stocks: int,
    entry_stats: Dict[str, int],
) -> bool:
    if code in portfolio or len(portfolio) >= max_stocks:
        return False
    entry_price = float(pe.get("entry_price") or 0)
    if entry_price <= 0:
        return False
    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:
        return False
    invest = min(invest_cap, remaining, target_cost)
    qty = int(invest / entry_price)
    if qty < 1:
        return False
    cost = qty * entry_price
    if cost < min_required or exposure + cost > total_budget + 1e-6:
        return False
    entry_time = str(pe.get("entry_time") or "")
    portfolio[code] = {
        "entry_price": entry_price,
        "entry_time": entry_time,
        "qty": qty,
        "stop": pe["stop"],
        "target": pe["target"],
        "max_price": entry_price,
        "rsi": pe.get("rsi"),
    }
    src = str(pe.get("entry_source") or "ohlc_open")
    if src == "ws_ticks":
        entry_stats["tick_entry_count"] = entry_stats.get("tick_entry_count", 0) + 1
    else:
        entry_stats["ohlc_entry_count"] = entry_stats.get("ohlc_entry_count", 0) + 1
    return True


def _time_bounds_hm(params: Dict[str, Any]) -> Tuple[int, int]:
    ts = int(params.get("time_start_hm", 900))
    te = int(params.get("mom_time_end_hm", params.get("time_end_hm", 1430)))
    return ts, te


def _hm_to_minutes(hm: int) -> int:
    return (hm // 100) * 60 + (hm % 100)


def _build_scan_time_keys(
    minute_set: Set[str],
    scan_sec: int,
    time_start_hm: int,
    time_end_hm: int,
) -> List[str]:
    """장중 분봉이 있는 구간만 N초 간격 스캔 시각(YYYYMMDDHHMMSS) 생성."""
    if not minute_set or scan_sec < 1:
        return []
    start_min = _hm_to_minutes(time_start_hm)
    end_min = _hm_to_minutes(time_end_hm)
    days = sorted({m[:8] for m in minute_set})
    out: List[str] = []
    for day in days:
        day_minutes = sorted(m for m in minute_set if m.startswith(day))
        for minute_key in day_minutes:
            hm = int(minute_key[8:12])
            bar_min = _hm_to_minutes(hm)
            if bar_min < start_min or bar_min >= end_min:
                continue
            base = datetime.strptime(minute_key, "%Y%m%d%H%M")
            sec = 0
            while sec < 60:
                out.append(base.replace(second=sec).strftime("%Y%m%d%H%M%S"))
                sec += scan_sec
    return out


def _is_minute_tail_scan(scan_key: str, scan_sec: int) -> bool:
    sec = int(str(scan_key)[-2:])
    return sec + scan_sec >= 60


def _record_momentum_sell(
    *,
    portfolio: Dict[str, Dict[str, Any]],
    code: str,
    ctx: Dict[str, Any],
    pos: Dict[str, Any],
    reason: str,
    exit_price: float,
    sell_time_key: str,
    hold_min: float,
    exit_source: str,
    all_trades: List[Dict],
    tick_exit_count: int,
    ohlc_exit_count: int,
) -> Tuple[int, int]:
    trade: Dict[str, Any] = {
        "code": code,
        "buy_time": pos["entry_time"],
        "sell_time": sell_time_key,
        "buy_price": pos["entry_price"],
        "sell_price": round(exit_price, 2),
        "qty": pos.get("qty", 1),
        "pnl": 0,
        "sell_reason": reason,
        "hold_min": hold_min,
        "exit_source": exit_source,
        "strategy": MOMENTUM_STRATEGY_ID,
    }
    if pos.get("rsi") is not None:
        trade["rsi_entry"] = round(float(pos["rsi"]), 1)
    all_trades.append(trade)
    day = sell_time_key[:8]
    ctx["last_exit_dt"][day] = _t2dt(sell_time_key)
    del portfolio[code]
    if exit_source == "ws_ticks":
        tick_exit_count += 1
    else:
        ohlc_exit_count += 1
    return tick_exit_count, ohlc_exit_count


def _process_sells_for_scan(
    portfolio: Dict[str, Dict[str, Any]],
    ctx_by_code: Dict[str, Dict[str, Any]],
    scan_key: str,
    *,
    params: Dict[str, Any],
    force_eod_exit: bool,
    ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]],
    all_trades: List[Dict],
    tick_exit_count: int,
    ohlc_exit_count: int,
    scan_sec: int,
) -> Tuple[int, int]:
    """스캔 시각까지 틱·OHLC 청산 (실매 루프: 매도 먼저)."""
    bar_t = scan_key[:12]
    for code in list(portfolio.keys()):
        ctx = ctx_by_code.get(code)
        if ctx is None:
            continue
        idx = ctx["time_index"].get(bar_t)
        if idx is None:
            continue
        candles = ctx["candles"]
        c = candles[idx]
        pos = portfolio[code]
        if str(pos.get("entry_time") or "")[:12] == bar_t:
            continue
        day = bar_t[:8]
        is_eod_raw = (idx == len(candles) - 1) or (candles[idx + 1]["candle_time"][:8] != day)
        is_eod = is_eod_raw and force_eod_exit and _is_minute_tail_scan(scan_key, scan_sec)
        entry_time = str(pos.get("entry_time") or "")

        sold = False
        if momentum_backtest_use_tick_exit(params) and ticks_by_code:
            minute_ticks = collect_minute_ticks(ticks_by_code, code, bar_t)
            # 공유메모리 컬럼 뷰면 dict 재구성 없이 뷰 캡핑(동일 문자열 비교). 아니면 기존 리스트 캡핑.
            try:
                from kis_trader.backtest.shared_ticks import TickColumnView
                _is_view = isinstance(minute_ticks, TickColumnView)
            except Exception:
                _is_view = False
            if _is_view:
                capped = minute_ticks.cap_by_tick_time_le(scan_key[:14])
            else:
                capped = [
                    tk for tk in minute_ticks
                    if str(tk.get("tick_time") or "")[:14] <= scan_key[:14]
                ]
            if capped:
                tick_res = try_momentum_sell_on_ticks(
                    pos, capped, params, is_eod=is_eod, entry_time=entry_time,
                )
                if tick_res:
                    reason, fill_px, sell_time, hold_min = tick_res
                    tick_exit_count, ohlc_exit_count = _record_momentum_sell(
                        portfolio=portfolio, code=code, ctx=ctx, pos=pos,
                        reason=reason, exit_price=fill_px, sell_time_key=sell_time,
                        hold_min=hold_min, exit_source="ws_ticks",
                        all_trades=all_trades,
                        tick_exit_count=tick_exit_count, ohlc_exit_count=ohlc_exit_count,
                    )
                    sold = True

        if sold:
            continue

        if not _is_minute_tail_scan(scan_key, scan_sec):
            continue

        cur_c_info = {
            "open": float(c["open"]),
            "high": float(c["high"]),
            "low": float(c["low"]),
            "close": float(c["close"]),
            "candle_time": bar_t,
        }
        sell_res = resolve_momentum_sell_for_bar(
            pos, cur_c_info, params,
            is_eod=is_eod,
            ticks_by_code=ticks_by_code,
            code=code,
        )
        if not sell_res:
            continue
        reason, exit_price, sell_time_key, hold_min, exit_source = sell_res
        tick_exit_count, ohlc_exit_count = _record_momentum_sell(
            portfolio=portfolio, code=code, ctx=ctx, pos=pos,
            reason=reason, exit_price=exit_price, sell_time_key=sell_time_key,
            hold_min=hold_min, exit_source=exit_source,
            all_trades=all_trades,
            tick_exit_count=tick_exit_count, ohlc_exit_count=ohlc_exit_count,
        )
    return tick_exit_count, ohlc_exit_count


def _universe_codes_for_scan(
    *,
    scan_key: str,
    slot_key: str,
    universe_by_slot: Optional[Dict[str, List[str]]],
    universe_timeline: Optional[MomentumUniverseTimeline],
    use_scan_at: bool,
) -> Optional[List[str]]:
    if use_scan_at and universe_timeline is not None:
        return universe_timeline.codes_at(scan_key)
    if universe_by_slot is None:
        return None
    return universe_by_slot.get(slot_key, [])


def _collect_buy_candidates(
    *,
    bar_t: str,
    slot_key: str,
    ctx_by_code: Dict[str, Dict[str, Any]],
    portfolio: Dict[str, Dict[str, Any]],
    params: Dict[str, Any],
    universe_by_slot: Optional[Dict[str, List[str]]],
    universe_codes: Optional[List[str]] = None,
    live_align: bool,
    ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]],
    orderbook_by_code: Optional[Dict[str, Dict[str, List[Any]]]],
    program_by_code: Optional[Dict[str, Dict[str, List[Any]]]],
    sl_pct: float,
    tp_pct: float,
    min_tick_time: str = "",
    eval_memo: Optional[Dict[Tuple, Any]] = None,
) -> List[Tuple[Tuple[int, str], str, Dict[str, Any]]]:
    candidates: List[Tuple[Tuple[int, str], str, Dict[str, Any]]] = []
    # 순회 대상 종목: 유니버스가 있으면 그 종목만 순회 (전종목 261개 → 유니버스 ~28개).
    # 기존엔 전종목을 돌며 universe_codes 에 없는 종목을 버려 ~9배 낭비했음.
    # 후보는 아래에서 우선순위 키로 재정렬하므로 순회 순서는 결과에 무관 → 동작 불변.
    if universe_codes is not None:
        iter_codes = universe_codes
    elif universe_by_slot is not None:
        iter_codes = universe_by_slot.get(slot_key, [])
    else:
        iter_codes = list(ctx_by_code.keys())
    seen_codes: Set[str] = set()
    for code in iter_codes:
        if code in seen_codes:  # 유니버스 중복 종목 1회만 평가 (전종목 순회와 동일 결과)
            continue
        seen_codes.add(code)
        ctx = ctx_by_code.get(code)
        if ctx is None:
            continue
        if code in portfolio or ctx.get("pending_entry"):
            continue
        idx = ctx["time_index"].get(bar_t)
        if idx is None:
            continue
        candles = ctx["candles"]
        c = candles[idx]
        day = bar_t[:8]
        cl = float(c["close"])
        if cl <= 0:
            continue
        if live_align:
            if idx < 6:
                continue
            signal_idx = idx - 1
            signal_bar_time = candles[signal_idx]["candle_time"]
            entry_bar_time = bar_t
            entry_open = float(c["open"])
            if entry_open <= 0:
                continue
        else:
            if idx < 5:
                continue
            signal_idx = idx
            signal_bar_time = bar_t
            if idx + 1 >= len(candles):
                continue
            next_c = candles[idx + 1]
            if next_c["candle_time"][:8] != day:
                continue
            entry_bar_time = next_c["candle_time"]
            entry_open = float(next_c["open"])
            if entry_open <= 0:
                continue
        eval_params = dict(params)
        ic = get_indicator_cache_from_params(params, code)
        if ic is not None:
            eval_params["_indicator_cache"] = ic
        inject_whipsaw_ticks_into_params(
            eval_params,
            ticks_by_code=ticks_by_code,
            code=code,
            bar_candle_time=signal_bar_time,
            strategy="MOMENTUM",
            tf_min=1,
        )
        inject_trigger_snapshots_into_params(
            eval_params,
            orderbook_by_code=orderbook_by_code,
            program_by_code=program_by_code,
            code=code,
            bar_candle_time=entry_bar_time if live_align else signal_bar_time,
            prefer_time=min_tick_time or (entry_bar_time if live_align else signal_bar_time),
        )
        eval_params.setdefault(
            "skip_hts_scan_dupes",
            universe_codes is not None or universe_by_slot is not None,
        )
        state = {
            "daily_cnt": ctx["daily_cnt"].get(day, 0),
            "last_exit_dt": ctx["last_exit_dt"].get(day),
        }
        # eval 메모이즈: 10초 스캔큐가 같은 분·종목을 6번 평가하던 중복 제거.
        # eval_memo 가 None 이 아닐 때만(=틱·호가·프로그램·verdict 데이터 전무로
        # 스캔초에 결과가 무관할 때만) 동작 → 데이터 있으면 기존 경로 100% 불변.
        # 키: (종목, 신호봉idx, 당일매수수, 마지막청산시각) — 매수신호 결과를 좌우하는 상태 전부.
        if eval_memo is not None:
            _led = state["last_exit_dt"]
            _memo_key = (
                code, signal_idx, int(state["daily_cnt"] or 0),
                _led.isoformat() if _led is not None else "",
            )
            _cached = eval_memo.get(_memo_key)
            if _cached is not None:
                reject, _msg, sig = _cached
            else:
                reject, _msg, sig = eval_momentum_buy_at_index(
                    candles, signal_idx, eval_params, state,
                )
                eval_memo[_memo_key] = (reject, _msg, sig)
        else:
            reject, _msg, sig = eval_momentum_buy_at_index(
                candles, signal_idx, eval_params, state,
            )
        if reject or not sig:
            continue
        entry_price, entry_time_key, entry_src = align_momentum_entry_from_ticks(
            ticks_by_code, code, entry_bar_time, entry_open, params,
            min_tick_time=min_tick_time,
        )
        pe_data: Dict[str, Any] = {
            "entry_time": entry_time_key,
            "entry_price": entry_price,
            "entry_source": entry_src,
            "stop": entry_price * (1 - sl_pct),
            "target": entry_price * (1 + tp_pct),
            "rsi": sig.get("rsi"),
        }
        candidates.append((
            _buy_priority_key(code, slot_key, universe_by_slot, universe_codes),
            code, pe_data,
        ))
    return candidates


def run_momentum_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,
    orderbook_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
    program_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
) -> List[Dict]:
    """시각순 포트폴리오 백테스트 — MOMENTUM 전용."""
    rsi_period = int(params.get("rsi_period", 3))
    min_bars = max(rsi_period + 5, 6)
    force_eod_exit = _to_bool(params.get("force_eod_exit"), False)
    sl_pct = abs(float(params.get("sl_pct", 0.015)))
    tp_pct = effective_tp_pct_from_params(params)
    max_stocks = _max_stocks_from_params(params)
    slot_money = float(params.get("slot_money", 300_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=MOMENTUM_STRATEGY_ID)
    invest_cap = _resolve_invest_cap_krw(params, slot_money)
    fee_rate = float(params.get("fee_rate", 0.00015))
    sell_tax = float(params.get("sell_tax", 0.0018))

    skipped_micro_buys = 0
    tick_exit_count = 0
    ohlc_exit_count = 0
    entry_stats: Dict[str, int] = {}
    live_align = momentum_live_align_enabled(params)
    live_scan_queue = momentum_backtest_live_scan_queue_enabled(params)
    scan_sec = momentum_backtest_scan_sec(params)
    universe_timeline = params.get("_momentum_universe_timeline")
    use_scan_at = (
        momentum_backtest_universe_scan_at_enabled(params)
        and universe_timeline is not None
    )
    attach_indicator_caches_to_params(params, codes_candles)
    # eval 메모이즈 게이트 — 틱·호가·프로그램·log verdict 가 전무하면 스캔초마다
    # 매수신호 결과가 동일하므로 (code,신호봉idx,당일매수수,마지막청산시각) 으로 캐시 가능.
    # 하나라도 있으면 None → 메모 비활성(기존 경로 그대로, 결과 불변).
    _eval_memo_safe = (
        not ticks_by_code
        and not orderbook_by_code
        and not program_by_code
        and not params.get("_backtest_log_verdict_by_code")
    )
    eval_memo: Optional[Dict[Tuple, Any]] = {} if _eval_memo_safe else None
    ctx_by_code: Dict[str, Dict[str, Any]] = {}
    all_times_set = set()
    period_start = str(params.get("_backtest_period_start_key") or "")[:12]
    for code, raw_rows in codes_candles.items():
        if len(raw_rows) < min_bars:
            continue
        candles = [dict(r) for r in raw_rows]
        ctx_by_code[code] = {
            "code": code,
            "candles": candles,
            "time_index": {c["candle_time"]: idx for idx, c in enumerate(candles)},
            "last_exit_dt": {},
            "daily_cnt": {},
            "pending_entry": None,
        }
        for c in candles:
            ct = str(c.get("candle_time") or "")
            if period_start and ct < period_start:
                continue
            all_times_set.add(ct)

    all_times = sorted(all_times_set)
    portfolio: Dict[str, Dict[str, Any]] = {}
    all_trades: List[Dict] = []
    scan_events = 0
    scan_buys = 0

    if live_scan_queue and live_align:
        time_start_hm, time_end_hm = _time_bounds_hm(params)
        scan_keys = _build_scan_time_keys(all_times_set, scan_sec, time_start_hm, time_end_hm)
        for scan_key in scan_keys:
            scan_events += 1
            bar_t = scan_key[:12]
            slot_key = _slot_key(bar_t, int(params.get("scan_interval_min", 1)))

            tick_exit_count, ohlc_exit_count = _process_sells_for_scan(
                portfolio, ctx_by_code, scan_key,
                params=params,
                force_eod_exit=force_eod_exit,
                ticks_by_code=ticks_by_code,
                all_trades=all_trades,
                tick_exit_count=tick_exit_count,
                ohlc_exit_count=ohlc_exit_count,
                scan_sec=scan_sec,
            )

            if len(portfolio) >= max_stocks:
                continue
            if portfolio_exposure_krw(portfolio) >= total_budget - 1e-6:
                continue

            scan_univ = _universe_codes_for_scan(
                scan_key=scan_key,
                slot_key=slot_key,
                universe_by_slot=universe_by_slot,
                universe_timeline=universe_timeline,
                use_scan_at=use_scan_at,
            )
            if scan_univ is not None and not scan_univ:
                continue

            candidates = _collect_buy_candidates(
                bar_t=bar_t,
                slot_key=slot_key,
                ctx_by_code=ctx_by_code,
                portfolio=portfolio,
                params=params,
                universe_by_slot=universe_by_slot,
                universe_codes=scan_univ,
                live_align=True,
                ticks_by_code=ticks_by_code,
                orderbook_by_code=orderbook_by_code,
                program_by_code=program_by_code,
                sl_pct=sl_pct,
                tp_pct=tp_pct,
                min_tick_time=scan_key,
                eval_memo=eval_memo,
            )
            if not candidates:
                continue
            candidates.sort(key=lambda x: x[0])
            _pri, pick_code, pe = candidates[0]
            pick_ctx = ctx_by_code[pick_code]
            if _try_open_momentum_position(
                portfolio, pick_code, pe,
                invest_cap=invest_cap,
                total_budget=total_budget,
                min_invest_ratio=min_invest_ratio,
                max_stocks=max_stocks,
                entry_stats=entry_stats,
            ):
                pick_ctx["daily_cnt"][bar_t[:8]] = pick_ctx["daily_cnt"].get(bar_t[:8], 0) + 1
                scan_buys += 1
            else:
                skipped_micro_buys += 1
    else:
        for t in all_times:
            slot_key = _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_open = float(pe.get("entry_price") or 0)
                entry_price, entry_time_key, entry_src = align_momentum_entry_from_ticks(
                    ticks_by_code, code, t, entry_open, params,
                )
                pe = dict(pe)
                pe["entry_price"] = entry_price
                pe["entry_time"] = entry_time_key
                pe["entry_source"] = entry_src
                if not _try_open_momentum_position(
                    portfolio, code, pe,
                    invest_cap=invest_cap,
                    total_budget=total_budget,
                    min_invest_ratio=min_invest_ratio,
                    max_stocks=max_stocks,
                    entry_stats=entry_stats,
                ):
                    skipped_micro_buys += 1
                    continue
                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]
                if str(portfolio[code]["entry_time"])[:12] == str(t)[:12]:
                    continue
                is_eod_raw = (idx == len(candles) - 1) or (candles[idx + 1]["candle_time"][:8] != day)
                is_eod = is_eod_raw and force_eod_exit
                cur_c_info = {
                    "open": float(c["open"]),
                    "high": float(c["high"]),
                    "low": float(c["low"]),
                    "close": float(c["close"]),
                    "candle_time": t,
                }
                pos = portfolio[code]
                sell_res = resolve_momentum_sell_for_bar(
                    pos, cur_c_info, params,
                    is_eod=is_eod,
                    ticks_by_code=ticks_by_code,
                    code=code,
                )
                if not sell_res:
                    continue
                reason, exit_price, sell_time_key, hold_min, exit_source = sell_res
                tick_exit_count, ohlc_exit_count = _record_momentum_sell(
                    portfolio=portfolio, code=code, ctx=ctx, pos=pos,
                    reason=reason, exit_price=exit_price, sell_time_key=sell_time_key,
                    hold_min=hold_min, exit_source=exit_source,
                    all_trades=all_trades,
                    tick_exit_count=tick_exit_count, ohlc_exit_count=ohlc_exit_count,
                )

            if len(portfolio) >= max_stocks:
                continue
            if portfolio_exposure_krw(portfolio) >= total_budget - 1e-6:
                continue

            candidates = _collect_buy_candidates(
                bar_t=t,
                slot_key=slot_key,
                ctx_by_code=ctx_by_code,
                portfolio=portfolio,
                params=params,
                universe_by_slot=universe_by_slot,
                live_align=live_align,
                ticks_by_code=ticks_by_code,
                orderbook_by_code=orderbook_by_code,
                program_by_code=program_by_code,
                sl_pct=sl_pct,
                tp_pct=tp_pct,
            )
            if not candidates:
                continue
            candidates.sort(key=lambda x: x[0])
            _pri, pick_code, pe = candidates[0]
            pick_ctx = ctx_by_code[pick_code]
            if live_align:
                if _try_open_momentum_position(
                    portfolio, pick_code, pe,
                    invest_cap=invest_cap,
                    total_budget=total_budget,
                    min_invest_ratio=min_invest_ratio,
                    max_stocks=max_stocks,
                    entry_stats=entry_stats,
                ):
                    pick_ctx["daily_cnt"][t[:8]] = pick_ctx["daily_cnt"].get(t[:8], 0) + 1
                else:
                    skipped_micro_buys += 1
            else:
                pick_ctx["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
    if entry_stats:
        skip_stats.update(entry_stats)
    if live_scan_queue and live_align:
        skip_stats["buy_queue_mode"] = "live_scan"
        skip_stats["scan_sec"] = scan_sec
        skip_stats["scan_events"] = scan_events
        skip_stats["scan_buys"] = scan_buys
        if use_scan_at:
            skip_stats["universe_mode"] = "scan_at"
            utm = params.get("_universe_timeline_meta") or {}
            skip_stats["universe_debounce_sec"] = utm.get("debounce_sec")
        else:
            skip_stats["universe_mode"] = "minute_slot"
    else:
        skip_stats["buy_queue_mode"] = "minute_legacy"
    if skip_stats:
        params["_portfolio_skip_stats"] = skip_stats

    attach_scalp_trade_pnl(
        all_trades, fee_rate=fee_rate, sell_tax=sell_tax,
        slip_pct=backtest_slip_pct(params),
    )
    all_trades.sort(key=lambda x: x["sell_time"])
    return all_trades