tick-stock-panel/backend/app/backtest/engine.py

"""回测引擎 — 共享数据加载 + 撮合 + 统计计算。

纯 Polars/NumPy 实现，不依赖 pandas/vectorbt。
"""
from __future__ import annotations

import hashlib
import logging
import threading
import time
from collections import OrderedDict
from dataclasses import dataclass
from datetime import date
from typing import Callable

logger = logging.getLogger(__name__)
from typing import Literal

import numpy as np
import polars as pl

from app.parquet import scan_enriched_parquet
from app.tickflow.repository import KlineRepository

logger = logging.getLogger(__name__)


# ================================================================
# 数据结构
# ================================================================

@dataclass
class MatcherConfig:
    # matching 为向后兼容入口: 仅传 matching 时, entry_fill/exit_fill 都取 matching 的值。
    # 显式传入 entry_fill/exit_fill 时以二者为准 (允许建仓/清仓口径不同)。
    matching: Literal["close_t", "open_t+1"] = "close_t"
    entry_fill: Literal["close_t", "open_t+1"] | None = None
    exit_fill: Literal["close_t", "open_t+1"] | None = None
    # 成本模型: 优先使用拆分口径 (佣金双边 + 印花税仅卖出 + 滑点双边)。
    # 未设 commission_pct 时回退到 fees_pct 作为双边佣金 (向后兼容, 无印花税)。
    fees_pct: float = 0.0002
    commission_pct: float | None = None
    stamp_tax_pct: float | None = None
    slippage_bps: float = 5.0
    stop_loss_pct: float | None = None
    take_profit_pct: float | None = None
    trailing_stop_pct: float | None = None
    trailing_take_profit_activate_pct: float | None = None
    trailing_take_profit_drawdown_pct: float | None = None
    max_hold_days: int | None = None
    max_positions: int = 10
    max_exposure_pct: float = 1.0
    score_min: float | None = None
    score_max: float | None = None
    initial_capital: float = 1_000_000.0
    position_sizing: Literal["equal", "score_weight"] = "equal"
    # 分钟K精确成交: 开启后, 信号触发日的成交价用当日分钟K优化
    # (有参考线→穿越价, 无参考线→VWAP)。数据缺失时降级为日K口径。
    minute_fill: bool = False

    def __post_init__(self) -> None:
        # 解析最终口径: 优先 entry_fill/exit_fill, 否则回退到 matching (向后兼容)。
        if self.entry_fill is None:
            self.entry_fill = self.matching
        if self.exit_fill is None:
            self.exit_fill = self.matching

    def _commission_pct(self) -> float:
        # commission_pct 显式给出时优先, 否则回退 fees_pct (向后兼容双边佣金)。
        return self.commission_pct if self.commission_pct is not None else self.fees_pct

    def buy_cost_pct(self) -> float:
        # 买入腿: 佣金 + 滑点。
        return self._commission_pct() + self.slippage_bps / 10000.0

    def sell_cost_pct(self) -> float:
        # 卖出腿: 佣金 + 印花税 + 滑点。印花税未设时为 0 (向后兼容)。
        stamp = self.stamp_tax_pct if self.stamp_tax_pct is not None else 0.0
        return self._commission_pct() + stamp + self.slippage_bps / 10000.0


@dataclass
class TradeRecord:
    symbol: str
    entry_date: date
    exit_date: date
    entry_price: float
    exit_price: float
    pnl_pct: float
    duration: int
    exit_reason: str  # "signal" | "stop_loss" | "take_profit" | "trailing_stop" | "trailing_take_profit" | "max_hold" | "end"
    # 退出优先级 (高→低): pending_exit(历史挂单) > 风控(止损/移动止损/移动止盈) > signal(卖点) > max_hold(到期) > end
    name: str = ""
    shares: float = 0.0
    lots: float = 0.0
    position_pct: float = 0.0
    entry_value: float = 0.0
    exit_value: float = 0.0
    pnl_amount: float = 0.0
    entry_score: float | None = None
    entry_signal_date: date | str | None = None
    exit_signal_date: date | str | None = None
    blocked_exit_days: int = 0
    # 触发买入/卖出的具体信号列名 (如 signal_ma_golden_5_20 / csg_xxx);
    # 仅当该腿由信号触发时填充, 止损/止盈/到期等非信号退出时 exit_signal_id 为 None。
    entry_signal_id: str | None = None
    exit_signal_id: str | None = None


@dataclass
class SimResult:
    equity_curve: list[dict]       # [{date, value}]
    drawdown_curve: list[dict]     # [{date, value}]
    trades: list[TradeRecord]
    per_symbol_stats: list[dict]
    stats: dict


def _resolve_signal_id(panel: pl.DataFrame, idx: int, signal_ids: list[str] | None) -> str | None:
    """在触发行 idx 上, 从候选信号里找出 panel 列为 True 的那个, 返回其列名。

    多个信号同时为 True 时返回第一个匹配的 (信号 OR 关系, 回测只记录其一即可)。
    signal_ids 元素可能带 signal_/csg_ 前缀, 也可能是裸名 (如 "ma_golden_5_20")。
    """
    if not signal_ids:
        return None
    for sid in signal_ids:
        col = sid if (sid.startswith("signal_") or sid.startswith("csg_")) else f"signal_{sid}"
        if col not in panel.columns:
            continue
        try:
            if bool(panel[col][idx]):
                return col
        except (IndexError, TypeError):
            continue
    return None


# ================================================================
# PanelCache — 避免重复 scan_parquet + compute_all
# ================================================================

class _CacheEntry:
    __slots__ = ("df", "ts")

    def __init__(self, df: pl.DataFrame, ts: float):
        self.df = df
        self.ts = ts


class _InFlight:
    """同 key 正在计算的占位: leader 算完通过 done 唤醒所有跟随者复用结果。"""

    __slots__ = ("done", "df", "error")

    def __init__(self) -> None:
        self.done = threading.Event()
        self.df: pl.DataFrame | None = None
        self.error: BaseException | None = None


class PanelCache:
    """LRU + TTL 数据面板缓存。"""

    def __init__(self, max_size: int = 2, ttl_seconds: int = 180):
        self._cache: OrderedDict[str, _CacheEntry] = OrderedDict()
        self._max_size = max_size
        self._ttl = ttl_seconds
        # 跨请求单例, SSE 回测在各自 daemon 线程并发访问 OrderedDict。
        # 无锁的 move_to_end/del/popitem check-then-act 会抛 "OrderedDict mutated"。
        # 用实例锁守护所有 OrderedDict 变更; compute_fn (重扫盘) 放锁外避免串行化。
        self._lock = threading.Lock()
        # single-flight: 同 key 只让一个线程 compute, 其余等其结果复用。
        # 否则优化器等场景下 max_workers 个线程冷启动同时 miss, 会并行加载 N 份同一面板。
        self._inflight: dict[str, _InFlight] = {}

    def get_or_compute(
        self,
        symbols: list[str] | None,
        start: date,
        end: date,
        columns: list[str] | None,
        compute_fn,
        asset_type: str = "stock",
    ) -> pl.DataFrame:
        key = self._make_key(symbols, start, end, columns, asset_type)
        now = time.monotonic()

        with self._lock:
            entry = self._cache.get(key)
            if entry is not None:
                if now - entry.ts < self._ttl:
                    self._cache.move_to_end(key)
                    return entry.df
                del self._cache[key]  # 过期, 丢弃后重算
            # single-flight: 同 key 若已有线程在算, 登记为跟随者; 否则本线程当 leader。
            flight = self._inflight.get(key)
            leader = flight is None
            if leader:
                flight = _InFlight()
                self._inflight[key] = flight

        if not leader:
            # 跟随者: 等 leader 算完直接复用, 不重复 compute (消除缓存踩踏)。
            flight.done.wait()
            if flight.error is not None:
                raise flight.error
            return flight.df

        # leader: compute 放锁外 (不同 key 仍可并发, 保留原设计优点)。
        try:
            df = compute_fn(symbols, start, end, columns, asset_type)
        except BaseException as e:
            # 失败不缓存: 摘除 inflight 让后续线程重试, 并把异常透传给已在等的跟随者。
            with self._lock:
                self._inflight.pop(key, None)
            flight.error = e
            flight.done.set()
            raise
        with self._lock:
            self._cache[key] = _CacheEntry(df=df, ts=now)
            if len(self._cache) > self._max_size:
                self._cache.popitem(last=False)
            self._inflight.pop(key, None)
        flight.df = df
        flight.done.set()
        return df

    def invalidate(self) -> None:
        with self._lock:
            self._cache.clear()

    @staticmethod
    def _make_key(symbols: list[str] | None, start: date, end: date, columns: list[str] | None, asset_type: str = "stock") -> str:
        if symbols is None:
            h = "all"
        else:
            h = hashlib.md5(",".join(sorted(symbols)).encode()).hexdigest()[:12]
        cols = "all" if columns is None else hashlib.md5(",".join(sorted(columns)).encode()).hexdigest()[:8]
        return f"{asset_type}:{h}:{start}:{end}:{cols}"


# ================================================================
# BacktestEngine
# ================================================================

class BacktestEngine:
    """回测引擎 — 数据加载 + 撮合模拟 + 统计计算。"""

    def __init__(self, repo: KlineRepository) -> None:
        self.repo = repo
        self._cache = PanelCache()

    # ── 数据加载 ──────────────────────────────────────

    def load_panel(
        self,
        symbols: list[str] | None,
        start: date,
        end: date,
        columns: list[str] | None = None,
        asset_type: str = "stock",
    ) -> pl.DataFrame:
        """加载 enriched 数据面板，带缓存。asset_type='etf' 时读 ETF enriched。"""
        return self._cache.get_or_compute(symbols, start, end, columns, self._load_panel_inner, asset_type=asset_type)

    def _load_panel_inner(
        self,
        symbols: list[str] | None,
        start: date,
        end: date,
        columns: list[str] | None = None,
        asset_type: str = "stock",
    ) -> pl.DataFrame:
        t0 = time.perf_counter()

        # 近期区间优先复用 repository 的预计算 enriched 历史缓存 (仅 stock: 该缓存为股票专用)。
        try:
            if asset_type == "stock" and self.repo is not None and hasattr(self.repo, "get_enriched_range"):
                cached = self.repo.get_enriched_range(start, end, symbols=symbols, columns=columns)
                if cached is not None and not cached.is_empty():
                    elapsed = (time.perf_counter() - t0) * 1000
                    logger.info("load_panel(cache): %.0fms, %d rows, %d columns", elapsed, len(cached), len(cached.columns))
                    return cached
        except Exception as e:  # noqa: BLE001
            logger.debug("backtest load panel cache miss: %s", e)

        from app.tickflow.repository import enriched_dirname
        enriched_glob = str(self.repo.store.data_dir / enriched_dirname(asset_type) / "**" / "*.parquet")

        try:
            lf = scan_enriched_parquet(enriched_glob)
            if symbols is not None:
                lf = lf.filter(pl.col("symbol").is_in(symbols))
            if columns is not None:
                available = set(lf.collect_schema().names())
                selected = [c for c in columns if c in available]
                if "symbol" not in selected and "symbol" in available:
                    selected.insert(0, "symbol")
                if "date" not in selected and "date" in available:
                    selected.insert(1, "date")
                lf = lf.select(selected)
            df = (
                lf.filter(
                    (pl.col("date") >= start)
                    & (pl.col("date") <= end)
                )
                .sort(["symbol", "date"])
                .collect(streaming=True)
            )
        except Exception as e:
            logger.warning("backtest load panel failed: %s", e)
            return pl.DataFrame()

        if df.is_empty():
            return df

        if columns is not None:
            elapsed = (time.perf_counter() - t0) * 1000
            logger.info("load_panel: %.0fms, %d rows, %d columns", elapsed, len(df), len(df.columns))
            return df

        from app.indicators.pipeline import compute_all
        # 按 asset_type 取维表: ETF 回测须用 ETF 维表, 否则名称 JOIN 失败(全 null)、
        # 涨停信号算在错误的 instruments 上。
        instruments = self.repo.get_instruments_asset(asset_type)
        df = compute_all(df, instruments=instruments)
        if not instruments.is_empty() and "name" not in df.columns:
            inst_cols = [c for c in ["symbol", "name"] if c in instruments.columns]
            if len(inst_cols) == 2:
                df = df.join(
                    instruments.select(inst_cols).unique(subset=["symbol"]),
                    on="symbol",
                    how="left",
                )

        elapsed = (time.perf_counter() - t0) * 1000
        logger.info("load_panel: %.0fms, %d rows", elapsed, len(df))
        return df

    # ── 撮合模拟 ──────────────────────────────────────

    def simulate(
        self,
        panel: pl.DataFrame,
        entries: pl.Series | None,
        exits: pl.Series | None,
        config: MatcherConfig,
        entry_signal_ids: list[str] | None = None,
        exit_signal_ids: list[str] | None = None,
    ) -> SimResult:
        """纯 NumPy 撮合模拟 — 逐 symbol 状态机。"""
        if panel.is_empty():
            return self._empty_result()

        n = len(panel)
        panel_dates = panel["date"].to_numpy()
        panel_symbols = panel["symbol"].to_numpy()

        # 构建信号数组
        ent = np.zeros(n, dtype=bool)
        ext = np.zeros(n, dtype=bool)
        if entries is not None and len(entries) == n:
            ent = entries.to_numpy().astype(bool)
        if exits is not None and len(exits) == n:
            ext = exits.to_numpy().astype(bool)

        if not ent.any():
            return self._empty_result()

        # 成交口径: entry/exit 可分别配置 close_t (信号当日收盘) 或 open_t+1 (次日开盘)。
        # open_t+1 时信号右移 1 天 (用前一根的信号 + 当根的 open 成交)。
        open_prices = panel["open"].to_numpy()
        close_prices = panel["close"].to_numpy()

        # 同一 symbol 内相邻行掩码, 跨 symbol 边界不允许 shift (避免错配)。
        same_prev_symbol = np.zeros(n, dtype=bool)
        same_prev_symbol[1:] = panel_symbols[1:] == panel_symbols[:-1]

        entry_prices = open_prices if config.entry_fill == "open_t+1" else close_prices
        exit_prices = open_prices if config.exit_fill == "open_t+1" else close_prices

        if config.entry_fill == "open_t+1":
            ent_s = np.zeros(n, dtype=bool)
            ent_s[1:] = ent[:-1] & same_prev_symbol
            ent = ent_s
        if config.exit_fill == "open_t+1":
            ext_s = np.zeros(n, dtype=bool)
            ext_s[1:] = ext[:-1] & same_prev_symbol
            ext = ext_s

        # 逐 symbol 撮合
        trades: list[TradeRecord] = []
        unique_symbols = np.unique(panel_symbols)

        for sym in unique_symbols:
            mask = panel_symbols == sym
            sym_ent = ent[mask]
            sym_ext = ext[mask]
            sym_entry_prices = entry_prices[mask]
            sym_exit_prices = exit_prices[mask]
            sym_close = close_prices[mask]
            sym_dates = panel_dates[mask]

            holding = False
            entry_idx = -1
            entry_price = 0.0
            hold_days = 0

            for i in range(len(sym_ent)):
                if not holding:
                    if sym_ent[i]:
                        holding = True
                        entry_idx = i
                        entry_price = float(sym_entry_prices[i])
                        hold_days = 0
                else:
                    hold_days += 1
                    exit_triggered = False
                    exit_reason = ""

                    # 止损 — 用当日 close 检测 (优先级最高)
                    if config.stop_loss_pct is not None:
                        pnl = (float(sym_close[i]) - entry_price) / entry_price
                        if pnl <= -abs(config.stop_loss_pct):
                            exit_triggered = True
                            exit_reason = "stop_loss"

                    # 信号退出 (优先于 max_hold: 卖点信号是策略主动离场)
                    if not exit_triggered and sym_ext[i]:
                        exit_triggered = True
                        exit_reason = "signal"

                    # 最大持仓天数 (兜底: 无信号/未止损时强制平仓)
                    if not exit_triggered and config.max_hold_days is not None:
                        if hold_days >= config.max_hold_days:
                            exit_triggered = True
                            exit_reason = "max_hold"

                    if exit_triggered:
                        exit_price = float(sym_exit_prices[i])
                        pnl_pct = (exit_price - entry_price) / entry_price if entry_price > 0 else 0.0
                        fee_cost = config.buy_cost_pct() + config.sell_cost_pct()
                        pnl_pct -= fee_cost

                        e_date = sym_dates[entry_idx]
                        x_date = sym_dates[i]
                        trades.append(TradeRecord(
                            symbol=str(sym),
                            entry_date=e_date.item() if hasattr(e_date, "item") else e_date,
                            exit_date=x_date.item() if hasattr(x_date, "item") else x_date,
                            entry_price=round(entry_price, 4),
                            exit_price=round(exit_price, 4),
                            pnl_pct=round(pnl_pct, 6),
                            duration=int(hold_days),
                            exit_reason=exit_reason,
                        ))
                        holding = False

        # 净值曲线: 按出场日期归集收益
        all_dates_sorted = np.sort(np.unique(panel_dates))
        equity_curve, drawdown_curve = self._build_curves(trades, all_dates_sorted, config.initial_capital)

        # 统计
        date_min = panel_dates.min()
        date_max = panel_dates.max()
        d_min = date_min.item() if hasattr(date_min, "item") else date_min
        d_max = date_max.item() if hasattr(date_max, "item") else date_max
        stats = self._calc_stats(trades, config.initial_capital, d_min, d_max)
        per_symbol = self._calc_per_symbol(trades)

        return SimResult(
            equity_curve=equity_curve,
            drawdown_curve=drawdown_curve,
            trades=trades,
            per_symbol_stats=per_symbol,
            stats=stats,
        )

    def simulate_independent_candidates(
        self,
        panel: pl.DataFrame,
        entries: pl.Series | None,
        exits: pl.Series | None,
        config: MatcherConfig,
        progress_cb: "Callable[[dict], None] | None" = None,
        cancel_event: "threading.Event | None" = None,
        entry_signal_ids: list[str] | None = None,
        exit_signal_ids: list[str] | None = None,
    ) -> SimResult:
        """全量候选独立执行：每个买入信号都是独立样本, 不受资金/仓位限制。"""
        if panel.is_empty():
            return self._empty_result()

        n = len(panel)
        panel_dates = panel["date"].to_numpy()
        panel_symbols = panel["symbol"].to_numpy()

        ent_raw = np.zeros(n, dtype=bool)
        ext_raw = np.zeros(n, dtype=bool)
        if entries is not None and len(entries) == n:
            ent_raw = entries.to_numpy().astype(bool)
        if exits is not None and len(exits) == n:
            ext_raw = exits.to_numpy().astype(bool)
        n_candidates = int(ent_raw.sum())
        if n_candidates <= 0:
            return self._empty_result()

        entry_signal_dates = np.array([None] * n, dtype=object)
        exit_signal_dates = np.array([None] * n, dtype=object)
        same_prev_symbol = panel_symbols[1:] == panel_symbols[:-1]

        # 建仓口径: close_t 用信号日收盘, open_t+1 右移到次日 open 成交。
        ent = np.zeros(n, dtype=bool)
        if config.entry_fill == "open_t+1":
            ent[1:] = ent_raw[:-1] & same_prev_symbol
            for idx in np.flatnonzero(ent):
                entry_signal_dates[idx] = self._date_str(panel_dates[idx - 1])
        else:
            ent = ent_raw
            for idx in np.flatnonzero(ent):
                entry_signal_dates[idx] = self._date_str(panel_dates[idx])

        # 清仓口径: 独立于建仓, close_t 用信号日收盘, open_t+1 右移到次日 open。
        ext = np.zeros(n, dtype=bool)
        if config.exit_fill == "open_t+1":
            ext[1:] = ext_raw[:-1] & same_prev_symbol
            for idx in np.flatnonzero(ext):
                exit_signal_dates[idx] = self._date_str(panel_dates[idx - 1])
        else:
            ext = ext_raw
            for idx in np.flatnonzero(ext):
                exit_signal_dates[idx] = self._date_str(panel_dates[idx])

        open_prices = panel["open"].to_numpy()
        high_prices = panel["high"].to_numpy() if "high" in panel.columns else open_prices
        low_prices = panel["low"].to_numpy()
        close_prices = panel["close"].to_numpy()
        # 撮合价: 建仓/清仓各自独立选列。
        entry_prices = open_prices if config.entry_fill == "open_t+1" else close_prices
        exit_prices = open_prices if config.exit_fill == "open_t+1" else close_prices

        # ── 分钟K精确成交预加载 (同 simulate_portfolio) ──
        minute_cache: dict = {}
        if config.minute_fill:
            _trigger_dates: set[str] = set()
            _trigger_symbols: set[str] = set()
            for _idx in range(n):
                if ent[_idx] or ext[_idx]:
                    _trigger_dates.add(self._date_str(panel_dates[_idx]))
                    _trigger_symbols.add(str(panel_symbols[_idx]))
            if _trigger_dates and _trigger_symbols:
                _loaded = self._load_minute_for_fills(
                    self.repo, list(_trigger_symbols), _trigger_dates, "stock",
                )
                for _key, _marr in _loaded.items():
                    if _marr is not None and len(_marr) > 0:
                        minute_cache[_key] = _marr

        def _refill_price(idx: int, side: str, daily_price: float) -> float:
            if not config.minute_fill or not minute_cache:
                return daily_price
            _sym = str(panel_symbols[idx])
            _d = self._date_str(panel_dates[idx])
            _marr = minute_cache.get((_sym, _d))
            if _marr is None:
                return daily_price
            _ref = None
            for _col in ("ma5", "ma10", "ma20"):
                if _col in panel.columns:
                    try:
                        _fv = float(panel[_col][idx])
                        if _fv > 0 and np.isfinite(_fv):
                            _ref = _fv
                            break
                    except (TypeError, ValueError):
                        pass
            _precise = self._resolve_minute_fill(_marr, _ref, side)
            return _precise if _precise is not None else daily_price

        has_volume = "volume" in panel.columns
        volumes = panel["volume"].fill_null(0).to_numpy() if has_volume else np.ones(n, dtype=float)
        names = panel["name"].fill_null("").to_numpy() if "name" in panel.columns else np.array([""] * n)
        scores = panel["score"].fill_null(0).to_numpy() if "score" in panel.columns else np.zeros(n, dtype=float)
        trade_scores = scores.copy()
        # 评分跟随建仓口径 shift (评分在买入日生效)。
        if config.entry_fill == "open_t+1":
            trade_scores[1:] = np.where(panel_symbols[1:] == panel_symbols[:-1], scores[:-1], trade_scores[1:])
        limit_up_flags = (
            panel["signal_limit_up"].fill_null(False).to_numpy().astype(bool)
            if "signal_limit_up" in panel.columns else np.zeros(n, dtype=bool)
        )
        limit_down_flags = (
            panel["signal_limit_down"].fill_null(False).to_numpy().astype(bool)
            if "signal_limit_down" in panel.columns else np.zeros(n, dtype=bool)
        )

        symbol_rows: dict[str, list[int]] = {}
        row_pos_in_symbol = np.zeros(n, dtype=int)
        for i, sym_value in enumerate(panel_symbols):
            sym = str(sym_value)
            rows = symbol_rows.setdefault(sym, [])
            row_pos_in_symbol[i] = len(rows)
            rows.append(i)

        buy_cost_pct = config.buy_cost_pct()
        sell_cost_pct = config.sell_cost_pct()
        score_min = getattr(config, "score_min", None)
        score_max = getattr(config, "score_max", None)
        trades: list[TradeRecord] = []
        execution_stats: dict[str, int] = {
            "buy_invalid_price": 0,
            "buy_suspended": 0,
            "buy_limit_up": 0,
            "buy_score_filter": 0,
            "buy_no_next_bar": max(n_candidates - int(ent.sum()), 0),
            "sell_invalid_price": 0,
            "sell_suspended": 0,
            "sell_limit_down": 0,
            "sell_no_future": 0,
            "pending_exit": 0,
        }

        def _count(key: str) -> None:
            execution_stats[key] = execution_stats.get(key, 0) + 1

        def _valid_price(value) -> bool:
            try:
                v = float(value)
            except (TypeError, ValueError):
                return False
            return v > 0 and np.isfinite(v)

        def _is_suspended(idx: int) -> bool:
            o = float(open_prices[idx])
            h = float(high_prices[idx])
            l = float(low_prices[idx])
            c = float(close_prices[idx])
            valid_bar = any(_valid_price(x) for x in (o, h, l, c))
            if not valid_bar:
                return True
            if has_volume and float(volumes[idx] or 0) <= 0:
                same_price = max(o, h, l, c) - min(o, h, l, c) <= max(abs(c) * 1e-4, 0.01)
                if same_price:
                    return True
            return False

        def _is_one_price_limit(idx: int, direction: str) -> bool:
            if _is_suspended(idx):
                return False
            o = float(open_prices[idx])
            h = float(high_prices[idx])
            l = float(low_prices[idx])
            c = float(close_prices[idx])
            if not all(_valid_price(x) for x in (o, h, l, c)):
                return False
            same_price = max(o, h, l, c) - min(o, h, l, c) <= max(abs(c) * 1e-4, 0.01)
            if direction == "up":
                return bool(limit_up_flags[idx]) and same_price
            return bool(limit_down_flags[idx]) and same_price

        def _can_buy(idx: int) -> tuple[bool, str]:
            if _is_suspended(idx):
                return False, "buy_suspended"
            if not _valid_price(entry_prices[idx]):
                return False, "buy_invalid_price"
            if _is_one_price_limit(idx, "up"):
                return False, "buy_limit_up"
            return True, ""

        def _can_sell(idx: int, exit_price_override: float | None = None) -> tuple[bool, str]:
            if _is_suspended(idx):
                return False, "sell_suspended"
            exit_price = exit_price_override if exit_price_override is not None else exit_prices[idx]
            if not _valid_price(exit_price):
                return False, "sell_invalid_price"
            if _is_one_price_limit(idx, "down"):
                return False, "sell_limit_down"
            return True, ""

        def _risk_exit(pos: dict, idx: int) -> tuple[str | None, float | None]:
            if pos.get("pending_exit_reason") or pos.get("entry_idx") == idx:
                return None, None
            entry_price = float(pos["entry_price"])
            if entry_price <= 0:
                return None, None
            open_price = float(open_prices[idx])
            low_price = float(low_prices[idx])
            high_price = float(high_prices[idx])
            peak_price = float(pos.get("max_high", entry_price))
            risk_lines: list[tuple[float, str]] = []

            if config.stop_loss_pct is not None:
                risk_lines.append((entry_price * (1 - abs(config.stop_loss_pct)), "stop_loss"))
            if config.trailing_stop_pct is not None and peak_price > 0:
                risk_lines.append((peak_price * (1 - abs(config.trailing_stop_pct)), "trailing_stop"))

            activate_pct = getattr(config, "trailing_take_profit_activate_pct", None)
            drawdown_pct = getattr(config, "trailing_take_profit_drawdown_pct", None)
            if activate_pct is not None and drawdown_pct is not None and peak_price > entry_price:
                peak_profit = peak_price / entry_price - 1
                if peak_profit >= abs(float(activate_pct)):
                    # 回撤止盈触发线: 相对峰值价回撤 drawdown 个点 (纯峰值口径)
                    risk_lines.append((peak_price * (1 - abs(float(drawdown_pct))), "trailing_take_profit"))

            risk_lines = [(line, reason) for line, reason in risk_lines if _valid_price(line)]
            # 止损/移损/回撤止盈: 价格跌破风控线触发 (取最高优先级线)
            if risk_lines:
                stop_price, reason = max(risk_lines, key=lambda item: item[0])
                if _valid_price(open_price) and open_price <= stop_price:
                    return reason, open_price
                if _valid_price(low_price) and low_price <= stop_price:
                    return reason, stop_price

            # 固定止盈: 价格涨破止盈线触发
            tp_pct = getattr(config, "take_profit_pct", None)
            if tp_pct is not None:
                tp_line = entry_price * (1 + abs(float(tp_pct)))
                if _valid_price(tp_line):
                    # 开盘即超过止盈线 → 以开盘价成交; 否则当日触及高点止盈
                    if _valid_price(open_price) and open_price >= tp_line:
                        return "take_profit", open_price
                    if _valid_price(high_price) and high_price >= tp_line:
                        return "take_profit", tp_line
            return None, None

        def _try_close(pos: dict, idx: int, reason: str, signal_date: str, exit_price_override: float | None = None) -> bool:
            ok, block_reason = _can_sell(idx, exit_price_override)
            if not ok:
                if not pos.get("pending_exit_reason"):
                    pos["pending_exit_reason"] = reason
                    pos["pending_exit_signal_date"] = signal_date
                    _count("pending_exit")
                pos["blocked_exit_days"] = int(pos.get("blocked_exit_days", 0)) + 1
                _count(block_reason)
                return False

            if exit_price_override is not None:
                exit_price = float(exit_price_override)
            else:
                exit_price = _refill_price(idx, "sell", float(exit_prices[idx]))
            shares = 100.0
            entry_value = shares * float(pos["entry_price"]) * (1 + buy_cost_pct)
            exit_value = shares * exit_price * (1 - sell_cost_pct)
            pnl_amount = exit_value - entry_value
            pnl_pct = pnl_amount / entry_value if entry_value > 0 else 0.0
            trades.append(TradeRecord(
                symbol=str(pos["symbol"]),
                name=str(pos.get("name", "")),
                entry_date=pos["entry_date"],
                exit_date=self._date_str(panel_dates[idx]),
                entry_price=round(float(pos["entry_price"]), 4),
                exit_price=round(exit_price, 4),
                pnl_pct=round(float(pnl_pct), 6),
                duration=int(pos["hold_days"]),
                exit_reason=reason,
                shares=shares,
                lots=1.0,
                position_pct=0.0,
                entry_value=round(float(entry_value), 2),
                exit_value=round(float(exit_value), 2),
                pnl_amount=round(float(pnl_amount), 2),
                entry_score=round(float(pos["entry_score"]), 2) if pos.get("entry_score") is not None else None,
                entry_signal_date=pos.get("entry_signal_date"),
                exit_signal_date=signal_date,
                blocked_exit_days=int(pos.get("blocked_exit_days", 0)),
                entry_signal_id=pos.get("entry_signal_id"),
                exit_signal_id=_resolve_signal_id(panel, idx, exit_signal_ids) if reason == "signal" else None,
            ))
            return True

        candidate_indices = np.flatnonzero(ent)
        for seq, entry_idx in enumerate(candidate_indices, start=1):
            if cancel_event is not None and cancel_event.is_set():
                logger.info("全量模拟被用户取消 (第 %d/%d 个候选)", seq, len(candidate_indices))
                break
            if progress_cb is not None and (seq == 1 or seq % 500 == 0):
                try:
                    progress_cb({
                        "day": seq,
                        "total": len(candidate_indices),
                        "date": self._date_str(panel_dates[entry_idx]),
                        "equity": 0,
                    })
                except Exception:
                    pass

            ok, block_reason = _can_buy(entry_idx)
            if not ok:
                _count(block_reason)
                continue
            score = float(trade_scores[entry_idx] or 0.0)
            if score_min is not None and score < score_min:
                _count("buy_score_filter")
                continue
            if score_max is not None and score > score_max:
                _count("buy_score_filter")
                continue

            sym = str(panel_symbols[entry_idx])
            rows = symbol_rows.get(sym, [])
            start_pos = int(row_pos_in_symbol[entry_idx])
            if start_pos >= len(rows):
                _count("sell_no_future")
                continue

            entry_price = _refill_price(entry_idx, "buy", float(entry_prices[entry_idx]))
            pos = {
                "symbol": sym,
                "name": str(names[entry_idx] or ""),
                "entry_idx": entry_idx,
                "entry_date": self._date_str(panel_dates[entry_idx]),
                "entry_signal_date": entry_signal_dates[entry_idx] or self._date_str(panel_dates[entry_idx]),
                "entry_signal_id": _resolve_signal_id(panel, entry_idx, entry_signal_ids),
                "entry_price": entry_price,
                "entry_score": score,
                "hold_days": 0,
                "max_high": entry_price,
                "pending_exit_reason": None,
                "pending_exit_signal_date": None,
                "blocked_exit_days": 0,
            }
            hi = float(high_prices[entry_idx])
            if _valid_price(hi):
                pos["max_high"] = max(float(pos["max_high"]), hi)

            closed = False
            last_idx = entry_idx
            for idx in rows[start_pos + 1:]:
                last_idx = idx
                pos["hold_days"] = int(pos["hold_days"]) + 1
                d_str = self._date_str(panel_dates[idx])

                def _scheduled_reason() -> tuple[str | None, str]:
                    if pos.get("pending_exit_reason"):
                        return str(pos["pending_exit_reason"]), str(pos.get("pending_exit_signal_date") or d_str)
                    # 卖点信号优先于到期: 策略主动离场先于 max_hold 兜底。
                    if ext[idx]:
                        return "signal", str(exit_signal_dates[idx] or d_str)
                    if config.max_hold_days is not None and pos["hold_days"] >= config.max_hold_days:
                        return "max_hold", d_str
                    if idx == rows[-1]:
                        return "end", d_str
                    return None, d_str

                # 统一退出顺序: 风控(止损/移动止损/止盈)先于计划出场 (signal/max_hold/end)。
                # 无论 entry/exit 口径如何, 风控都是保护性离场, 必须最高优先级。
                reason, override_price = _risk_exit(pos, idx)
                if reason and _try_close(pos, idx, reason, d_str, override_price):
                    closed = True
                    break
                reason, signal_date = _scheduled_reason()
                if reason and _try_close(pos, idx, reason, signal_date):
                    closed = True
                    break

                hi = float(high_prices[idx])
                if _valid_price(hi):
                    pos["max_high"] = max(float(pos.get("max_high", entry_price)), hi)

            if not closed:
                if last_idx == entry_idx:
                    _count("sell_no_future")
                elif not pos.get("pending_exit_reason"):
                    _try_close(pos, last_idx, "end", self._date_str(panel_dates[last_idx]))

        return self._calc_independent_candidate_result(trades, n_candidates, execution_stats)

    # ── 分钟K精确成交 ──────────────────────────────────

    @staticmethod
    def _resolve_minute_fill(
        minute_arr: np.ndarray,
        ref_price: float | None,
        side: str,
    ) -> float | None:
        """用当日分钟K确定精确成交价。

        Args:
            minute_arr: float64 2D 数组, 列顺序 = _MINUTE_NUMERIC_COLS
                        [open(0), high(1), low(2), close(3), volume(4), amount(5)]
                        (缺失的尾部列直接不存在, 用 shape 判断)
            ref_price: 信号参考线价格 (如 MA5 值); None 表示无参考线
            side: "buy" 或 "sell", 决定穿越方向

        Returns:
            精确成交价, 或 None (降级到日K口径)
        """
        if minute_arr is None or len(minute_arr) == 0:
            return None

        ncols = minute_arr.shape[1] if minute_arr.ndim == 2 else 1
        opens = minute_arr[:, 0]
        highs = minute_arr[:, 1] if ncols > 1 else opens
        lows = minute_arr[:, 2] if ncols > 2 else opens
        closes = minute_arr[:, 3] if ncols > 3 else opens
        volumes = minute_arr[:, 4] if ncols > 4 else None
        amounts = minute_arr[:, 5] if ncols > 5 else None

        # 有参考线 → 穿越价成交 (逻辑同止损: 找价格穿越参考线的时刻)
        if ref_price is not None and ref_price > 0 and np.isfinite(ref_price):
            if side == "sell":
                # 卖出: 价格跌破参考线 → 开盘已低于则按开盘; 否则按参考线 (低点触及)
                if np.isfinite(opens[0]) and opens[0] <= ref_price:
                    return float(opens[0])
                if np.any(np.isfinite(lows) & (lows <= ref_price)):
                    return float(ref_price)
            else:
                # 买入: 价格涨破参考线 → 开盘已高于则按开盘; 否则按参考线 (高点触及)
                if np.isfinite(opens[0]) and opens[0] >= ref_price:
                    return float(opens[0])
                if np.any(np.isfinite(highs) & (highs >= ref_price)):
                    return float(ref_price)
            # 参考线存在但当日分钟K未穿越 → 用收盘 (信号确认)
            return float(closes[-1]) if np.isfinite(closes[-1]) else None

        # 无参考线 → VWAP (成交额/成交量), 退化到收盘价
        if volumes is not None and amounts is not None:
            total_vol = float(np.nansum(volumes))
            total_amt = float(np.nansum(amounts))
            if total_vol > 0 and total_amt > 0:
                return total_amt / total_vol

        return float(closes[-1]) if np.isfinite(closes[-1]) else None

    # 分钟K cache 存储的数值列及固定顺序 (_resolve_minute_fill 按此顺序整数索引)。
    _MINUTE_NUMERIC_COLS = ["open", "high", "low", "close", "volume", "amount"]

    @staticmethod
    def _load_minute_for_fills(
        repo,
        symbols: list[str],
        dates_needed: set,
        asset_type: str,
    ) -> dict:
        """按触发日加载分钟K, 返回 {(symbol, date_str): float64 2D ndarray}。

        dates_needed: 需要分钟数据的日期集合 (set of date strings "YYYY-MM-DD")

        按触发日分批读取对应分区文件 (get_minute_by_dates), 而非扫描整个区间
        (get_minute_range)。内存与回测区间长度解耦, 只随触发日数量增长 ——
        触发日稀疏时避免读取区间内大量无关日期导致爆内存。

        cache 值为 float64 紧凑 2D 数组 (列顺序见 _MINUTE_NUMERIC_COLS), 而非
        完整 DataFrame, 避免每条记录携带 polars 元数据开销导致内存膨胀。
        """
        if not symbols or not dates_needed:
            return {}
        from datetime import date as _date
        sorted_date_strs = sorted(dates_needed)
        date_objs = [_date.fromisoformat(s) for s in sorted_date_strs]

        cache: dict = {}
        # 分批读取: 每批 50 个交易日, 处理完拼进 cache, 避免单批过大。
        BATCH = 50
        numeric_cols = BacktestEngine._MINUTE_NUMERIC_COLS
        for i in range(0, len(date_objs), BATCH):
            batch = date_objs[i:i + BATCH]
            try:
                df = repo.get_minute_by_dates(symbols, batch, asset_type=asset_type)
            except Exception as e:  # noqa: BLE001
                logger.warning("minute fill data load failed (batch %d-%d): %s", i, i + len(batch), e)
                continue
            if df.is_empty():
                continue
            # 按 (symbol, 日期) 分组, 每组转紧凑 float64 数组存入 cache
            df = df.with_columns(
                pl.col("datetime").dt.strftime("%Y-%m-%d").alias("_d_str")
            )
            for sub in df.partition_by(["symbol", "_d_str"], as_dict=False):
                if sub.is_empty():
                    continue
                sym = sub["symbol"][0]
                d_str = sub["_d_str"][0]
                cols = [c for c in numeric_cols if c in sub.columns]
                cache[(sym, d_str)] = sub.select(
                    [pl.col(c).cast(pl.Float64) for c in cols]
                ).to_numpy()
        return cache

    def simulate_portfolio(
        self,
        panel: pl.DataFrame,
        entries: pl.Series | None,
        exits: pl.Series | None,
        config: MatcherConfig,
        progress_cb: "Callable[[dict], None] | None" = None,
        cancel_event: "threading.Event | None" = None,
        entry_signal_ids: list[str] | None = None,
        exit_signal_ids: list[str] | None = None,
    ) -> SimResult:
        """账户级组合回测：日线信号 → 成交约束 → 仓位/现金撮合。"""
        if panel.is_empty():
            return self._empty_result()

        n = len(panel)
        panel_dates = panel["date"].to_numpy()
        panel_symbols = panel["symbol"].to_numpy()

        ent_raw = np.zeros(n, dtype=bool)
        ext_raw = np.zeros(n, dtype=bool)
        if entries is not None and len(entries) == n:
            ent_raw = entries.to_numpy().astype(bool)
        if exits is not None and len(exits) == n:
            ext_raw = exits.to_numpy().astype(bool)
        if not ent_raw.any():
            return self._empty_result()

        entry_signal_dates = np.array([None] * n, dtype=object)
        exit_signal_dates = np.array([None] * n, dtype=object)
        same_prev_symbol = panel_symbols[1:] == panel_symbols[:-1]

        # 建仓口径: close_t 用信号日收盘, open_t+1 右移到次日 open 成交。
        ent = np.zeros(n, dtype=bool)
        if config.entry_fill == "open_t+1":
            ent[1:] = ent_raw[:-1] & same_prev_symbol
            for idx in np.flatnonzero(ent):
                entry_signal_dates[idx] = self._date_str(panel_dates[idx - 1])
        else:
            ent = ent_raw
            for idx in np.flatnonzero(ent):
                entry_signal_dates[idx] = self._date_str(panel_dates[idx])

        # 清仓口径: 独立于建仓。
        ext = np.zeros(n, dtype=bool)
        if config.exit_fill == "open_t+1":
            ext[1:] = ext_raw[:-1] & same_prev_symbol
            for idx in np.flatnonzero(ext):
                exit_signal_dates[idx] = self._date_str(panel_dates[idx - 1])
        else:
            ext = ext_raw
            for idx in np.flatnonzero(ext):
                exit_signal_dates[idx] = self._date_str(panel_dates[idx])

        open_prices = panel["open"].to_numpy()
        high_prices = panel["high"].to_numpy() if "high" in panel.columns else open_prices
        low_prices = panel["low"].to_numpy()
        close_prices = panel["close"].to_numpy()
        # 撮合价: 建仓/清仓各自独立选列。
        entry_prices = open_prices if config.entry_fill == "open_t+1" else close_prices
        exit_prices = open_prices if config.exit_fill == "open_t+1" else close_prices
        has_volume = "volume" in panel.columns
        volumes = panel["volume"].fill_null(0).to_numpy() if has_volume else np.ones(n, dtype=float)
        names = (
            panel["name"].fill_null("").to_numpy()
            if "name" in panel.columns else np.array([""] * n)
        )
        scores = (
            panel["score"].fill_null(0).to_numpy()
            if "score" in panel.columns else np.zeros(n, dtype=float)
        )
        trade_scores = scores.copy()
        # 评分跟随建仓口径 shift (评分在买入日生效)。
        if config.entry_fill == "open_t+1":
            trade_scores[1:] = np.where(panel_symbols[1:] == panel_symbols[:-1], scores[:-1], trade_scores[1:])
        limit_up_flags = (
            panel["signal_limit_up"].fill_null(False).to_numpy().astype(bool)
            if "signal_limit_up" in panel.columns else np.zeros(n, dtype=bool)
        )
        limit_down_flags = (
            panel["signal_limit_down"].fill_null(False).to_numpy().astype(bool)
            if "signal_limit_down" in panel.columns else np.zeros(n, dtype=bool)
        )

        date_to_indices: dict[str, list[int]] = {}
        for i, d in enumerate(panel_dates):
            d_str = self._date_str(d)
            date_to_indices.setdefault(d_str, []).append(i)
        all_dates = sorted(date_to_indices.keys())
        if not all_dates:
            return self._empty_result()

        buy_cost_pct = config.buy_cost_pct()
        sell_cost_pct = config.sell_cost_pct()
        cash = float(config.initial_capital)
        peak = cash
        max_positions = max(int(config.max_positions), 0)
        max_exposure_pct = min(max(float(getattr(config, "max_exposure_pct", 1.0)), 0.0), 1.0)
        score_min = getattr(config, "score_min", None)
        score_max = getattr(config, "score_max", None)
        positions: dict[str, dict] = {}
        last_close: dict[str, float] = {}
        trades: list[TradeRecord] = []

        # ── 分钟K精确成交预加载 ──
        # 信号触发日加载分钟K, 成交时用穿越价/VWAP替代收盘价
        minute_cache: dict = {}  # {(symbol, date_str): structured ndarray}
        if config.minute_fill:
            trigger_dates: set[str] = set()
            trigger_symbols: set[str] = set()
            for idx in range(n):
                if ent[idx] or ext[idx]:
                    trigger_dates.add(self._date_str(panel_dates[idx]))
                    trigger_symbols.add(str(panel_symbols[idx]))
            if trigger_dates and trigger_symbols:
                asset_type = "etf" if all(
                    str(s).endswith(".SH") and str(s).startswith("5") for s in list(trigger_symbols)[:5]
                ) else "stock"
                loaded = self._load_minute_for_fills(
                    self.repo, list(trigger_symbols), trigger_dates, asset_type,
                )
                for key, marr in loaded.items():
                    if marr is not None and len(marr) > 0:
                        minute_cache[key] = marr

        def _refill_price(idx: int, side: str, daily_price: float) -> float:
            """分钟K精确成交价; 无数据则降级为 daily_price。"""
            if not config.minute_fill or not minute_cache:
                return daily_price
            sym = str(panel_symbols[idx])
            d_str = self._date_str(panel_dates[idx])
            marr = minute_cache.get((sym, d_str))
            if marr is None:
                return daily_price
            # 参考线: 从 panel 取 ma5/ma10/ma20 作为近似 (均线类信号)
            ref = None
            for col in ("ma5", "ma10", "ma20"):
                if col in panel.columns:
                    val = panel[col][idx]
                    try:
                        fv = float(val)
                        if fv > 0 and np.isfinite(fv):
                            ref = fv
                            break
                    except (TypeError, ValueError):
                        pass
            precise = self._resolve_minute_fill(marr, ref, side)
            return precise if precise is not None else daily_price

        equity_curve: list[dict] = []
        drawdown_curve: list[dict] = []
        execution_stats: dict[str, int] = {
            "buy_invalid_price": 0,
            "buy_suspended": 0,
            "buy_limit_up": 0,
            "buy_no_slot": 0,
            "buy_cash": 0,
            "buy_lot_size": 0,
            "buy_same_day_reentry": 0,
            "buy_exposure": 0,
            "buy_score_filter": 0,
            "sell_invalid_price": 0,
            "sell_suspended": 0,
            "sell_limit_down": 0,
            "pending_exit": 0,
        }

        def _count(key: str) -> None:
            execution_stats[key] = execution_stats.get(key, 0) + 1

        def _valid_price(value) -> bool:
            try:
                v = float(value)
            except (TypeError, ValueError):
                return False
            return v > 0 and np.isfinite(v)

        def _market_value() -> float:
            value = 0.0
            for pos in positions.values():
                mark = last_close.get(pos["symbol"], pos["entry_price"])
                value += pos["shares"] * mark
            return value

        def _is_suspended(idx: int) -> bool:
            o = float(open_prices[idx])
            h = float(high_prices[idx])
            l = float(low_prices[idx])
            c = float(close_prices[idx])
            valid_bar = any(_valid_price(x) for x in (o, h, l, c))
            if not valid_bar:
                return True
            if has_volume and float(volumes[idx] or 0) <= 0:
                same_price = max(o, h, l, c) - min(o, h, l, c) <= max(abs(c) * 1e-4, 0.01)
                if same_price:
                    return True
            return False

        def _is_one_price_limit(idx: int, direction: str) -> bool:
            if _is_suspended(idx):
                return False
            o = float(open_prices[idx])
            h = float(high_prices[idx])
            l = float(low_prices[idx])
            c = float(close_prices[idx])
            if not all(_valid_price(x) for x in (o, h, l, c)):
                return False
            same_price = max(o, h, l, c) - min(o, h, l, c) <= max(abs(c) * 1e-4, 0.01)
            if direction == "up":
                return bool(limit_up_flags[idx]) and same_price
            return bool(limit_down_flags[idx]) and same_price

        def _can_buy(idx: int) -> tuple[bool, str]:
            if _is_suspended(idx):
                return False, "buy_suspended"
            if not _valid_price(entry_prices[idx]):
                return False, "buy_invalid_price"
            if _is_one_price_limit(idx, "up"):
                return False, "buy_limit_up"
            return True, ""

        def _can_sell(idx: int, exit_price_override: float | None = None) -> tuple[bool, str]:
            if _is_suspended(idx):
                return False, "sell_suspended"
            exit_price = exit_price_override if exit_price_override is not None else exit_prices[idx]
            if not _valid_price(exit_price):
                return False, "sell_invalid_price"
            if _is_one_price_limit(idx, "down"):
                return False, "sell_limit_down"
            return True, ""

        def _mark_pending(sym: str, reason: str, signal_date: str) -> None:
            pos = positions[sym]
            if not pos.get("pending_exit_reason"):
                pos["pending_exit_reason"] = reason
                pos["pending_exit_signal_date"] = signal_date
                _count("pending_exit")
            pos["blocked_exit_days"] = int(pos.get("blocked_exit_days", 0)) + 1

        def _sell(
            sym: str,
            idx: int,
            reason: str,
            signal_date: str,
            sold_today: set[str],
            exit_price_override: float | None = None,
        ) -> None:
            nonlocal cash
            pos = positions.pop(sym)
            if exit_price_override is not None:
                exit_price = float(exit_price_override)
            else:
                exit_price = _refill_price(idx, "sell", float(exit_prices[idx]))
            exit_value = pos["shares"] * exit_price * (1 - sell_cost_pct)
            cash += exit_value
            pnl_amount = exit_value - pos["entry_value"]
            pnl_pct = (exit_value - pos["entry_value"]) / pos["entry_value"] if pos["entry_value"] > 0 else 0.0
            sold_today.add(sym)
            trades.append(TradeRecord(
                symbol=sym,
                name=pos.get("name", ""),
                entry_date=pos["entry_date"],
                exit_date=self._date_str(panel_dates[idx]),
                entry_price=round(float(pos["entry_price"]), 4),
                exit_price=round(exit_price, 4),
                pnl_pct=round(float(pnl_pct), 6),
                duration=int(pos["hold_days"]),
                exit_reason=reason,
                shares=round(float(pos["shares"]), 4),
                lots=round(float(pos["lots"]), 2),
                position_pct=round(float(pos.get("position_pct", 0.0)), 6),
                entry_value=round(float(pos["entry_value"]), 2),
                exit_value=round(float(exit_value), 2),
                pnl_amount=round(float(pnl_amount), 2),
                entry_score=round(float(pos["entry_score"]), 2) if pos.get("entry_score") is not None else None,
                entry_signal_date=pos.get("entry_signal_date"),
                exit_signal_date=signal_date,
                blocked_exit_days=int(pos.get("blocked_exit_days", 0)),
                entry_signal_id=pos.get("entry_signal_id"),
                exit_signal_id=_resolve_signal_id(panel, idx, exit_signal_ids) if reason == "signal" else None,
            ))

        def _try_sell(
            sym: str,
            idx: int | None,
            reason: str,
            signal_date: str,
            sold_today: set[str],
            exit_price_override: float | None = None,
        ) -> bool:
            if idx is None:
                _mark_pending(sym, reason, signal_date)
                _count("sell_suspended")
                return False
            ok, block_reason = _can_sell(idx, exit_price_override)
            if not ok:
                _mark_pending(sym, reason, signal_date)
                _count(block_reason)
                return False
            _sell(sym, idx, reason, signal_date, sold_today, exit_price_override)
            return True

        def _process_scheduled_exits(
            d_idx: int,
            d_str: str,
            row_by_symbol: dict[str, int],
            sold_today: set[str],
        ) -> None:
            for sym in list(positions.keys()):
                pos = positions.get(sym)
                if pos is None:
                    continue
                idx = row_by_symbol.get(sym)
                reason = ""
                signal_date = d_str
                if pos.get("pending_exit_reason"):
                    reason = str(pos["pending_exit_reason"])
                    signal_date = str(pos.get("pending_exit_signal_date") or d_str)
                # 卖点信号优先于到期: 策略主动离场先于 max_hold 兜底。
                elif idx is not None and ext[idx]:
                    reason = "signal"
                    signal_date = str(exit_signal_dates[idx] or d_str)
                elif config.max_hold_days is not None and pos["hold_days"] >= config.max_hold_days:
                    reason = "max_hold"
                elif d_idx == len(all_dates) - 1:
                    reason = "end"
                if reason:
                    _try_sell(sym, idx, reason, signal_date, sold_today)

        def _process_risk_exits(d_str: str, row_by_symbol: dict[str, int], sold_today: set[str]) -> None:
            for sym in list(positions.keys()):
                pos = positions.get(sym)
                if pos is None or pos.get("pending_exit_reason"):
                    continue
                if pos.get("entry_date") == d_str:
                    continue
                idx = row_by_symbol.get(sym)
                if idx is None or pos["entry_price"] <= 0:
                    continue
                open_price = float(open_prices[idx])
                low_price = float(low_prices[idx])
                high_price = float(high_prices[idx])
                entry_price = float(pos["entry_price"])
                peak_price = float(pos.get("max_high", entry_price))
                risk_lines: list[tuple[float, str]] = []

                if config.stop_loss_pct is not None:
                    risk_lines.append((entry_price * (1 - abs(config.stop_loss_pct)), "stop_loss"))

                if config.trailing_stop_pct is not None and peak_price > 0:
                    risk_lines.append((peak_price * (1 - abs(config.trailing_stop_pct)), "trailing_stop"))

                activate_pct = getattr(config, "trailing_take_profit_activate_pct", None)
                drawdown_pct = getattr(config, "trailing_take_profit_drawdown_pct", None)
                if activate_pct is not None and drawdown_pct is not None and peak_price > entry_price:
                    peak_profit = peak_price / entry_price - 1
                    if peak_profit >= abs(float(activate_pct)):
                        # 回撤止盈触发线: 相对峰值价回撤 drawdown 个点 (纯峰值口径)
                        # 启动门槛用成本基准的浮盈率, 触发线用峰值基准, 与 trailing_stop 同口径
                        take_profit_line = peak_price * (1 - abs(float(drawdown_pct)))
                        risk_lines.append((take_profit_line, "trailing_take_profit"))

                # 止损/移损/回撤止盈: 价格跌破风控线触发
                risk_lines = [(line, reason) for line, reason in risk_lines if _valid_price(line)]
                if risk_lines:
                    stop_price, reason = max(risk_lines, key=lambda item: item[0])
                    exit_price_override = None
                    if _valid_price(open_price) and open_price <= stop_price:
                        exit_price_override = open_price
                    elif _valid_price(low_price) and low_price <= stop_price:
                        exit_price_override = stop_price
                    if exit_price_override is not None:
                        _try_sell(sym, idx, reason, d_str, sold_today, exit_price_override)
                        continue

                # 固定止盈: 价格涨破止盈线触发
                tp_pct = getattr(config, "take_profit_pct", None)
                if tp_pct is not None:
                    tp_line = entry_price * (1 + abs(float(tp_pct)))
                    if _valid_price(tp_line):
                        if _valid_price(open_price) and open_price >= tp_line:
                            _try_sell(sym, idx, "take_profit", d_str, sold_today, open_price)
                        elif _valid_price(high_price) and high_price >= tp_line:
                            _try_sell(sym, idx, "take_profit", d_str, sold_today, tp_line)

        def _process_entries(
            d_str: str,
            idxs: list[int],
            sold_today: set[str],
        ) -> None:
            nonlocal cash
            if max_positions <= 0:
                return
            candidates: list[tuple[int, str, float]] = []
            for idx in idxs:
                if not ent[idx]:
                    continue
                sym = str(panel_symbols[idx])
                if sym in positions:
                    continue
                if sym in sold_today:
                    _count("buy_same_day_reentry")
                    continue
                ok, block_reason = _can_buy(idx)
                if not ok:
                    _count(block_reason)
                    continue
                score = float(trade_scores[idx] or 0.0)
                if score_min is not None and score < score_min:
                    _count("buy_score_filter")
                    continue
                if score_max is not None and score > score_max:
                    _count("buy_score_filter")
                    continue
                candidates.append((idx, sym, score))
            if not candidates:
                return
            candidates.sort(key=lambda x: x[2], reverse=True)

            slots = max_positions - len(positions)
            if slots <= 0:
                execution_stats["buy_no_slot"] += len(candidates)
                return

            selected = candidates[:slots]
            market_value_before = _market_value()
            account_equity_before_buy = cash + market_value_before
            if account_equity_before_buy <= 0 or max_exposure_pct <= 0:
                execution_stats["buy_exposure"] += len(selected)
                return
            target_position_value = account_equity_before_buy * max_exposure_pct / max_positions
            max_exposure_value = account_equity_before_buy * max_exposure_pct
            exposure_capacity = max_exposure_value - market_value_before
            if exposure_capacity <= 0:
                execution_stats["buy_exposure"] += len(selected)
                return

            weights = np.repeat(1 / len(selected), len(selected))
            if config.position_sizing == "score_weight":
                raw = np.array([max(x[2], 0.0) for x in selected], dtype=float)
                if raw.sum() > 0:
                    weights = raw / raw.sum()
            total_budget = min(cash, exposure_capacity, target_position_value * len(selected))

            for (idx, sym, _score), weight in zip(selected, weights):
                if len(positions) >= max_positions:
                    _count("buy_no_slot")
                    break
                current_market_value = _market_value()
                current_equity = cash + current_market_value
                current_exposure_capacity = current_equity * max_exposure_pct - current_market_value
                allocation = min(total_budget * float(weight), target_position_value, cash, current_exposure_capacity)
                if allocation <= 0:
                    _count("buy_exposure")
                    continue
                entry_price = _refill_price(idx, "buy", float(entry_prices[idx]))
                shares = np.floor(allocation / (entry_price * (1 + buy_cost_pct)) / 100) * 100
                entry_value = shares * entry_price * (1 + buy_cost_pct)
                if shares <= 0:
                    _count("buy_lot_size")
                    continue
                if entry_value > cash + 1e-6:
                    _count("buy_cash")
                    continue
                if entry_value > current_exposure_capacity + 1e-6:
                    _count("buy_exposure")
                    continue
                cash -= entry_value
                positions[sym] = {
                    "symbol": sym,
                    "name": str(names[idx] or ""),
                    "entry_date": self._date_str(panel_dates[idx]),
                    "entry_signal_date": entry_signal_dates[idx] or self._date_str(panel_dates[idx]),
                    "entry_signal_id": _resolve_signal_id(panel, idx, entry_signal_ids),
                    "entry_price": entry_price,
                    "entry_value": entry_value,
                    "shares": shares,
                    "lots": shares / 100,
                    "position_pct": entry_value / account_equity_before_buy if account_equity_before_buy > 0 else 0.0,
                    "entry_score": _score,
                    "max_high": entry_price,
                    "hold_days": 0,
                    "pending_exit_reason": None,
                    "pending_exit_signal_date": None,
                    "blocked_exit_days": 0,
                }

        for d_idx, d_str in enumerate(all_dates):
            if d_idx % 20 == 0:
                if cancel_event is not None and cancel_event.is_set():
                    logger.info("回测被用户取消 (第 %d/%d 天)", d_idx, len(all_dates))
                    break
                if progress_cb is not None:
                    try:
                        progress_cb({
                            "day": d_idx + 1,
                            "total": len(all_dates),
                            "date": str(d_str)[:10],
                            "equity": round(cash + _market_value(), 2),
                        })
                    except Exception:
                        pass

            idxs = date_to_indices[d_str]
            row_by_symbol = {str(panel_symbols[i]): i for i in idxs}
            sold_today: set[str] = set()

            for pos in positions.values():
                pos["hold_days"] += 1

            # 统一执行顺序 (不分口径): 风控(止损/移动止损/止盈) → 计划出场(signal/max_hold/end) → 建仓。
            # 风控是保护性离场, 必须最先; 计划出场次之; 建仓最后 (卖出释放的现金/仓位先用于满足新买)。
            # 当天新建仓不会被风控误杀 (_process_risk_exits 跳过 entry_date == d_str 的仓位)。
            _process_risk_exits(d_str, row_by_symbol, sold_today)
            _process_scheduled_exits(d_idx, d_str, row_by_symbol, sold_today)
            if d_idx < len(all_dates) - 1:
                _process_entries(d_str, idxs, sold_today)

            for sym, pos in positions.items():
                idx = row_by_symbol.get(sym)
                if idx is not None:
                    hi = float(high_prices[idx])
                    if _valid_price(hi):
                        pos["max_high"] = max(float(pos.get("max_high", pos["entry_price"])), hi)

            for i in idxs:
                c = float(close_prices[i])
                if c > 0 and np.isfinite(c):
                    last_close[str(panel_symbols[i])] = c

            market_value = _market_value()
            equity = cash + market_value
            peak = max(peak, equity)
            dd = (equity - peak) / peak if peak > 0 else 0.0
            exposure = market_value / equity if equity > 0 else 0.0
            equity_curve.append({
                "date": d_str[:10],
                "value": round(float(equity), 2),
                "cash": round(float(cash), 2),
                "positions": len(positions),
                "exposure": round(float(exposure), 4),
            })
            drawdown_curve.append({"date": d_str[:10], "value": round(float(dd), 4)})

        stats = self._calc_portfolio_stats(equity_curve, trades, config.initial_capital)
        stats["execution"] = execution_stats
        stats["pending_exit_positions"] = sum(1 for p in positions.values() if p.get("pending_exit_reason"))
        per_symbol = self._calc_per_symbol(trades)
        return SimResult(
            equity_curve=equity_curve,
            drawdown_curve=drawdown_curve,
            trades=trades,
            per_symbol_stats=per_symbol,
            stats=stats,
        )

    # ── 净值曲线 ──────────────────────────────────────

    @staticmethod
    def _build_curves(
        trades: list[TradeRecord],
        all_dates: np.ndarray,
        initial_capital: float,
    ) -> tuple[list[dict], list[dict]]:
        """从交易记录构建日频净值曲线和回撤曲线。

        资金模型: 每笔交易等权分配 (1/N_capital)，N_capital = 同时持仓数上限。
        简化版: 按出场日归集所有已平仓交易的平均收益作为当日组合收益。
        """
        if not trades or len(all_dates) == 0:
            return [], []

        # 按出场日归集 pnl
        exit_pnl: dict[str, list[float]] = {}
        for t in trades:
            d_str = str(t.exit_date)
            exit_pnl.setdefault(d_str, []).append(t.pnl_pct)

        equity = initial_capital
        peak = initial_capital
        curve: list[dict] = []
        dd_curve: list[dict] = []

        for d in all_dates:
            d_str = str(d.item() if hasattr(d, "item") else d)
            pnls = exit_pnl.get(d_str, [])
            # 当日组合收益 = 该日所有出场交易的平均收益
            daily_ret = float(np.mean(pnls)) if pnls else 0.0
            equity *= (1 + daily_ret)
            peak = max(peak, equity)
            dd = (equity - peak) / peak if peak > 0 else 0.0
            curve.append({"date": d_str[:10], "value": round(equity, 2)})
            dd_curve.append({"date": d_str[:10], "value": round(dd, 4)})

        return curve, dd_curve

    # ── 统计计算 ──────────────────────────────────────

    @staticmethod
    def _sortino_ratio(returns: np.ndarray, periods_per_year: int = 252) -> float | None:
        """Sortino 比率: 用下行偏差 (仅惩罚负收益) 替代总标准差, 年化。

        下行偏差 = sqrt(mean(min(r, 0)^2)), MAR=0 的目标半方差 (对全部样本求均, 非仅负样本)。
        无下行波动 (无亏损) 时 Sortino 未定义, 返回 None (与 profit_factor 的 None 约定一致,
        不虚报 0 或 inf)。样本不足 (<2) 返回 0.0 (与 sharpe 的退化约定一致)。
        """
        returns = returns[np.isfinite(returns)]  # 剔除 inf/nan, 防止污染均值/序列化出非法 JSON
        if len(returns) < 2:
            return 0.0
        mean = float(np.mean(returns))
        downside = np.minimum(returns, 0.0)
        downside_dev = float(np.sqrt(np.mean(downside ** 2)))
        if downside_dev <= 0:
            return None
        return mean / downside_dev * float(np.sqrt(periods_per_year))

    @staticmethod
    def _mc_drawdown_percentiles(pnls: np.ndarray, n_sims: int = 1000) -> dict:
        """自助重抽样交易序列, 估计最大回撤的分布 — 回答"仅因成交顺序运气, 回撤能有多坏"。

        对每笔收益有放回重抽样 n_sims 次, 各自算最大回撤, 取分位:
        - mc_maxdd_p50: 中位场景最大回撤
        - mc_maxdd_p95: 95% 置信最坏场景 (= 分布 5 分位, 更负)

        固定种子保证可复现/可测。样本 <3 无统计意义, 返回 None。
        大样本 (如 full 模式数千笔) 时按 2M 单元上限压降模拟次数, 防止瞬时数组 OOM。
        """
        pnls = pnls[np.isfinite(pnls)]  # 剔除 inf/nan, 否则 cumprod 传播 nan 导致分位为 nan
        # 防御: 单笔 pnl <= -100% 时 (1+pnl) <= 0 会让 cumprod 符号翻转/得非正净值, 回撤失真。
        # 回测有止损, 实际不会发生; 兜底 clip 到 -99.99% 保证 (1+pnl) 恒正。
        pnls = np.clip(pnls, -0.9999, None)
        n = len(pnls)
        if n < 3:
            return {"mc_maxdd_p50": None, "mc_maxdd_p95": None}
        # 内存护栏: samples/equity/peak/dd 各占 eff_sims*n*8B, 控总单元 <= 2M (~64MB 峰值)
        eff_sims = min(n_sims, max(200, 2_000_000 // n))
        rng = np.random.default_rng(42)
        samples = rng.choice(pnls, size=(eff_sims, n), replace=True)
        equity = np.cumprod(1.0 + samples, axis=1)
        peak = np.maximum.accumulate(equity, axis=1)
        dd = (equity - peak) / peak
        maxdds = dd.min(axis=1)
        return {
            "mc_maxdd_p50": round(float(np.percentile(maxdds, 50)), 4),
            "mc_maxdd_p95": round(float(np.percentile(maxdds, 5)), 4),
        }

    @staticmethod
    def _per_trade_block(pnls: np.ndarray, durations: np.ndarray) -> dict:
        """per-trade 明细字段: best/worst/median_pnl/avg_holding_days。"""
        pnls = pnls[np.isfinite(pnls)]  # 剔除 inf/nan, 防 best/worst 出非法值
        durations = durations[np.isfinite(durations)] if len(durations) else durations
        if not len(pnls):
            return {"best": 0.0, "worst": 0.0, "median_pnl": 0.0, "avg_holding_days": 0.0}
        return {
            "best": round(float(np.max(pnls)), 4),
            "worst": round(float(np.min(pnls)), 4),
            "median_pnl": round(float(np.median(pnls)), 4),
            "avg_holding_days": round(float(np.mean(durations)), 1) if len(durations) else 0.0,
        }

    @staticmethod
    def _calc_stats(
        trades: list[TradeRecord],
        initial_capital: float,
        start: date,
        end: date,
    ) -> dict:
        if not trades:
            return {"total_return": 0, "n_trades": 0}

        pnls = np.array([t.pnl_pct for t in trades])
        n_trades = len(trades)

        # 从净值曲线推算总收益 (等权组合)
        cumulative = 1.0
        for p in pnls:
            cumulative *= (1 + p)
        # 修正: 等权组合的总收益不等于各笔复乘，用曲线终点更准
        # 但这里作为简化，用各笔复乘作为近似
        total_return = cumulative - 1.0

        # 年化
        n_days = max((end - start).days, 1)
        years = n_days / 365.25
        if total_return > -1.0 and years > 0:
            annual_return = (1 + total_return) ** (1 / years) - 1
        else:
            annual_return = total_return

        # 胜率
        wins = pnls[pnls > 0]
        losses = pnls[pnls <= 0]
        win_rate = len(wins) / n_trades

        # 盈亏比
        avg_win = float(np.mean(wins)) if len(wins) > 0 else 0.0
        avg_loss = abs(float(np.mean(losses))) if len(losses) > 0 else 0.0
        profit_factor = avg_win / avg_loss if avg_loss > 0 else (float("inf") if avg_win > 0 else 0.0)

        # 最大回撤 — 用交易序列近似
        equity = initial_capital
        peak = initial_capital
        max_dd = 0.0
        for p in pnls:
            equity *= (1 + p)
            peak = max(peak, equity)
            dd = (equity - peak) / peak
            max_dd = min(max_dd, dd)

        # 夏普 — 用交易收益标准差近似
        sharpe = float(np.mean(pnls) / np.std(pnls)) * np.sqrt(252) if np.std(pnls) > 0 else 0.0

        # Sortino: 刻意沿用本函数 sharpe 的逐笔收益 x sqrt(252) 基准。逐笔年化非严格正确,
        # 但保证同一函数内 sharpe/sortino 口径一致可比 (内部一致 > 局部绝对)。仅惩罚下行波动。
        sortino = BacktestEngine._sortino_ratio(pnls)

        # Calmar
        calmar = annual_return / abs(max_dd) if abs(max_dd) > 0.001 else 0.0

        durations = np.array([t.duration for t in trades], dtype=float)
        return {
            "total_return": round(float(total_return), 4),
            "annual_return": round(float(annual_return), 4),
            "max_drawdown": round(float(max_dd), 4),
            "sharpe": round(float(sharpe), 2),
            "sortino": round(float(sortino), 2) if sortino is not None else None,
            "calmar": round(float(calmar), 2),
            "win_rate": round(float(win_rate), 4),
            "profit_factor": round(float(profit_factor), 2) if np.isfinite(profit_factor) else None,
            "n_trades": n_trades,
            "avg_pnl": round(float(np.mean(pnls)), 4),
            "avg_win": round(avg_win, 4),
            "avg_loss": round(avg_loss, 4),
            **BacktestEngine._per_trade_block(pnls, durations),
            **BacktestEngine._mc_drawdown_percentiles(pnls),
        }

    @staticmethod
    def _calc_per_symbol(trades: list[TradeRecord]) -> list[dict]:
        if not trades:
            return []
        by_sym: dict[str, dict] = {}
        for t in trades:
            s = t.symbol
            d = by_sym.setdefault(s, {
                "symbol": s, "n_trades": 0, "total_return": 1.0,
                "best": -999.0, "worst": 999.0, "wins": 0, "pnls": [],
            })
            d["n_trades"] += 1
            d["pnls"].append(t.pnl_pct)
            d["total_return"] *= (1 + t.pnl_pct)
            d["best"] = max(d["best"], t.pnl_pct)
            d["worst"] = min(d["worst"], t.pnl_pct)
            if t.pnl_pct > 0:
                d["wins"] += 1

        result = []
        for d in by_sym.values():
            result.append({
                "symbol": d["symbol"],
                "n_trades": d["n_trades"],
                "total_return": round(d["total_return"] - 1.0, 4),
                "win_rate": round(d["wins"] / d["n_trades"], 4) if d["n_trades"] > 0 else 0.0,
                "best": round(d["best"], 4),
                "worst": round(d["worst"], 4),
            })
        return sorted(result, key=lambda x: x["total_return"], reverse=True)

    @staticmethod
    def _calc_independent_candidate_result(
        trades: list[TradeRecord],
        n_candidates: int,
        execution_stats: dict[str, int],
    ) -> SimResult:
        """全量独立候选统计：按每个候选样本的实际执行收益聚合。"""
        if not trades:
            return SimResult(
                equity_curve=[],
                drawdown_curve=[],
                trades=[],
                per_symbol_stats=[],
                stats={
                    "mode": "full",
                    "full_kind": "candidate_execution",
                    "error": "no executable trades",
                    "n_candidates": int(n_candidates),
                    "n_trades": 0,
                    "execution": execution_stats,
                },
            )

        pnls = np.array([t.pnl_pct for t in trades], dtype=float)
        durations = np.array([t.duration for t in trades], dtype=float)
        wins = pnls[pnls > 0]
        losses = pnls[pnls <= 0]
        avg_win = float(np.mean(wins)) if len(wins) else 0.0
        avg_loss = abs(float(np.mean(losses))) if len(losses) else 0.0

        # 按退出日聚合已实现样本收益, 构造“样本收益曲线”。它不是账户净值。
        daily_returns: dict[str, list[float]] = {}
        for t in trades:
            daily_returns.setdefault(str(t.exit_date)[:10], []).append(float(t.pnl_pct))

        equity_curve: list[dict] = []
        drawdown_curve: list[dict] = []
        equity = 1.0
        peak = 1.0
        daily_avg: list[float] = []
        for d_str in sorted(daily_returns.keys()):
            values = daily_returns[d_str]
            avg_ret = float(np.mean(values)) if values else 0.0
            daily_avg.append(avg_ret)
            equity *= (1 + avg_ret)
            peak = max(peak, equity)
            dd = (equity - peak) / peak if peak > 0 else 0.0
            equity_curve.append({
                "date": d_str,
                "value": round(float(equity), 4),
                "positions": len(values),
            })
            drawdown_curve.append({"date": d_str, "value": round(float(dd), 4)})

        values = np.array([r["value"] for r in equity_curve], dtype=float)
        total_return = float(values[-1] - 1.0) if len(values) else 0.0
        peaks = np.maximum.accumulate(values) if len(values) else np.array([])
        drawdowns = values / peaks - 1 if len(values) else np.array([])
        max_drawdown = float(drawdowns.min()) if len(drawdowns) else 0.0
        daily = np.array(daily_avg, dtype=float)
        sharpe = float(np.mean(daily) / np.std(daily) * np.sqrt(252)) if len(daily) > 1 and np.std(daily) > 0 else 0.0
        sortino = BacktestEngine._sortino_ratio(daily)

        lo, hi, nbins = -0.20, 0.20, 20
        clipped = np.clip(pnls, lo, hi)
        counts, edges = np.histogram(clipped, bins=nbins, range=(lo, hi))
        dist = [
            {
                "range": f"{(edges[i]*100):+.0f}~{(edges[i+1]*100):+.0f}%",
                "count": int(counts[i]),
                "ratio": round(float(counts[i] / pnls.size), 4) if pnls.size else 0.0,
            }
            for i in range(nbins)
        ]

        stats = {
            "mode": "full",
            "full_kind": "candidate_execution",
            "n_candidates": int(n_candidates),
            "n_trades": int(len(trades)),
            "n_days": int(len(daily_returns)),
            "avg_daily_candidates": round(float(len(trades) / max(len(daily_returns), 1)), 1),
            "avg_return": round(float(np.mean(pnls)), 4),
            "median_return": round(float(np.median(pnls)), 4),
            "win_rate": round(float(len(wins) / len(pnls)), 4) if len(pnls) else 0.0,
            "profit_factor": round(float(avg_win / avg_loss), 2) if avg_loss > 0 else None,
            "best": round(float(np.max(pnls)), 4),
            "worst": round(float(np.min(pnls)), 4),
            "avg_duration": round(float(np.mean(durations)), 1) if len(durations) else 0.0,
            "total_return": round(float(total_return), 4),
            "max_drawdown": round(float(max_drawdown), 4),
            "sharpe": round(float(sharpe), 2),
            "sortino": round(float(sortino), 2) if sortino is not None else None,
            "return_distribution": dist,
            "execution": execution_stats,
            **BacktestEngine._mc_drawdown_percentiles(pnls),
        }

        return SimResult(
            equity_curve=equity_curve,
            drawdown_curve=drawdown_curve,
            trades=trades,
            per_symbol_stats=BacktestEngine._calc_per_symbol(trades),
            stats=stats,
        )

    @staticmethod
    def _calc_portfolio_stats(
        equity_curve: list[dict],
        trades: list[TradeRecord],
        initial_capital: float,
    ) -> dict:
        if not equity_curve:
            return {"total_return": 0, "n_trades": 0}
        final_equity = float(equity_curve[-1]["value"])
        total_return = final_equity / initial_capital - 1 if initial_capital > 0 else 0.0
        values = np.array([float(r["value"]) for r in equity_curve], dtype=float)
        daily = values[1:] / values[:-1] - 1 if len(values) > 1 else np.array([])
        annual_return = (1 + total_return) ** (252 / max(len(equity_curve), 1)) - 1 if total_return > -1 else total_return
        peaks = np.maximum.accumulate(values)
        drawdowns = values / peaks - 1
        max_drawdown = float(drawdowns.min()) if len(drawdowns) else 0.0
        sharpe = float(np.mean(daily) / np.std(daily) * np.sqrt(252)) if len(daily) and np.std(daily) > 0 else 0.0
        sortino = BacktestEngine._sortino_ratio(daily)
        pnls = np.array([t.pnl_pct for t in trades], dtype=float) if trades else np.array([])
        durations = np.array([t.duration for t in trades], dtype=float) if trades else np.array([])
        exposures = np.array([float(r.get("exposure", 0.0)) for r in equity_curve], dtype=float)
        wins = pnls[pnls > 0]
        losses = pnls[pnls <= 0]
        avg_win = float(np.mean(wins)) if len(wins) else 0.0
        avg_loss = abs(float(np.mean(losses))) if len(losses) else 0.0
        return {
            "total_return": round(float(total_return), 4),
            "annual_return": round(float(annual_return), 4),
            "max_drawdown": round(float(max_drawdown), 4),
            "sharpe": round(float(sharpe), 2),
            "sortino": round(float(sortino), 2) if sortino is not None else None,
            "calmar": round(float(annual_return / abs(max_drawdown)), 2) if abs(max_drawdown) > 0.001 else 0.0,
            "win_rate": round(float(len(wins) / len(pnls)), 4) if len(pnls) else 0.0,
            "profit_factor": round(float(avg_win / avg_loss), 2) if avg_loss > 0 else None,
            "n_trades": len(trades),
            "avg_pnl": round(float(np.mean(pnls)), 4) if len(pnls) else 0.0,
            "avg_win": round(avg_win, 4),
            "avg_loss": round(avg_loss, 4),
            **BacktestEngine._per_trade_block(pnls, durations),
            **BacktestEngine._mc_drawdown_percentiles(pnls),
            "final_equity": round(final_equity, 2),
            "initial_capital": round(float(initial_capital), 2),
            "avg_exposure": round(float(np.mean(exposures)), 4) if len(exposures) else 0.0,
            "max_exposure": round(float(np.max(exposures)), 4) if len(exposures) else 0.0,
        }

    @staticmethod
    def _date_str(value) -> str:
        value = value.item() if hasattr(value, "item") else value
        return str(value)[:10]

    @staticmethod
    def _empty_result() -> SimResult:
        return SimResult(
            equity_curve=[], drawdown_curve=[], trades=[],
            per_symbol_stats=[], stats={"error": "no data or no signals"},
        )

    # ── 截面工具 (因子回测用) ─────────────────────────

    @staticmethod
    def cross_section_rank(panel: pl.DataFrame, col: str) -> pl.DataFrame:
        return panel.with_columns(
            pl.col(col).rank(method="random").over("date").alias(f"{col}_rank")
        )

    @staticmethod
    def cross_section_qcut(panel: pl.DataFrame, col: str, n_groups: int) -> pl.DataFrame:
        return panel.with_columns(
            pl.col(col).qcut(n_groups, labels=[f"Q{i+1}" for i in range(n_groups)])
            .over("date").alias("_group")
        )