feat(backtest): 分钟策略回测 v1 — 逐交易日回放信号分钟收盘入场

- 新增 MinuteSignalReplayer: 枚举区间内分钟分区日, 逐日组装 StrategyDataContext
  (timeframe=1m, 日线窗口严格止于 T-1) 并复用 strategy_engine.run, 与实盘选股同路径
- 缺分区日显式跳过并记入 skipped_days, 不回退最近分区
- 涨停拒买: 信号分钟收盘 >= 当日涨停价(T-1 raw_close + 板块幅度)剔除并计数
- MarketMatrix 新增 entry_price 覆盖矩阵, 引擎在有限值处优先于 open/close 惯例
- repository.list_minute_dates 按目录名枚举分钟分区日, 零 parquet 扫描
- strategy.run 增加 minute_filter 分支: 入场 delay 0/离场沿用日K matcher,
  trades.entry_date 补全为 YYYY-MM-DD HH:MM (北京时间)

backend/app/backtest/engine.py

@@ -813,6 +813,14 @@ class BacktestEngine:
             matrix, raw_candidates, config, progress_cb, cancel_event,
         )
 
+    @staticmethod
+    def _resolve_entry_prices(matrix: "MarketMatrix", config: "MatcherConfig") -> np.ndarray:
+        """入场价矩阵: 分钟策略的逐格覆盖有限值处优先, 否则按 open/close 惯例。"""
+        base = matrix.open if config.entry_fill == "open_t+1" else matrix.close
+        if matrix.entry_price is None:
+            return base
+        return np.where(np.isfinite(matrix.entry_price), matrix.entry_price, base)
+
     def _simulate_independent_matrix(
         self,
         matrix: MarketMatrix,
@@ -823,7 +831,7 @@ class BacktestEngine:
         options: SimulationOptions | None = None,
     ) -> SimResult:
         options = options or SimulationOptions()
-        entry_prices = matrix.open if config.entry_fill == "open_t+1" else matrix.close
+        entry_prices = self._resolve_entry_prices(matrix, config)
         exit_prices = matrix.open if config.exit_fill == "open_t+1" else matrix.close
         buy_cost_pct = config.buy_cost_pct()
         sell_cost_pct = config.sell_cost_pct()
@@ -1722,7 +1730,7 @@ class BacktestEngine:
     ) -> SimResult:
         options = options or SimulationOptions()
         time_count, asset_count = matrix.shape
-        entry_prices = matrix.open if config.entry_fill == "open_t+1" else matrix.close
+        entry_prices = self._resolve_entry_prices(matrix, config)
         exit_prices = matrix.open if config.exit_fill == "open_t+1" else matrix.close
         buy_cost_pct = config.buy_cost_pct()
         sell_cost_pct = config.sell_cost_pct()

backend/app/backtest/matrix.py

@@ -551,6 +551,9 @@ class MarketMatrix:
     exit_signal_code: np.ndarray
     entry_signal_ids: tuple[str, ...]
     exit_signal_ids: tuple[str, ...]
+    # 逐格入场价覆盖 (time x asset, NaN=回退 open/close 惯例)。分钟策略回测用:
+    # 信号在盘中第 m 根触发, 入场价 = 触发分钟收盘价, 而非当日开盘/收盘。
+    entry_price: np.ndarray | None = None
 
     @property
     def shape(self) -> tuple[int, int]:
@@ -2302,11 +2305,14 @@ def build_market_matrix_from_signals(
     exit_delay_bars: int = 0,
     reference_price: np.ndarray | None = None,
     minute_exit_trigger: bool = False,
+    entry_price_override: np.ndarray | None = None,
 ) -> MarketMatrix:
     """Combine base data and strategy signals into the matcher input matrix."""
     if entry_delay_bars not in (0, 1) or exit_delay_bars not in (0, 1):
         raise ValueError("phase-two MarketMatrix supports only zero or one bar delay")
     validate_signal_matrix(signals, market.shape)
+    if entry_price_override is not None and entry_price_override.shape != market.shape:
+        raise ValueError("entry_price_override shape does not match MarketDataMatrix")
 
     present = _present_matrix(market.open, market.high, market.low, market.close, market.volume)
     entry, entry_signal_time, entry_signal_code = _delay_signal_matrix(
@@ -2379,6 +2385,11 @@ def build_market_matrix_from_signals(
         exit_signal_code=exit_signal_code,
         entry_signal_ids=signals.entry_signal_ids,
         exit_signal_ids=signals.exit_signal_ids,
+        entry_price=(
+            np.array(entry_price_override, dtype=np.float32, copy=True)
+            if entry_price_override is not None
+            else None
+        ),
     )
 
 

backend/app/backtest/minute_replay.py

@@ -0,0 +1,286 @@
+"""分钟策略回测回放器 — 逐交易日回放 filter_minute_history 产生入场信号。
+
+与实盘选股 (ScreenerService 1m context) 走同一条 StrategyEngine.run 执行路径,
+消除回测/实盘偏差。语义铁律:
+
+- 分钟侧: 传入当日全量分钟分区, 策略函数自身因果 (第 m 根只用 <=m 的K线);
+- 日线侧: T 日的日线条件窗口只含 T-1 及更早的完成态日K — 与实盘盘中行为一致
+  (当日成形K不进窗口), 杜绝未来函数;
+- 按交易日精确对日: 缺分钟分区的日子显式跳过, 不做"回退最近分区" (那是实盘语义)。
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from datetime import date, timedelta
+
+import polars as pl
+
+from app.price_limits import is_risk_warning_name, price_limit_pct
+from app.strategy.engine import StrategyDataContext, StrategyDef, StrategyEngine
+
+# 日线面板列: 基础行情 + 涨停/炸板信号 (策略日线窗口契约) + 基础过滤/展示列。
+# raw_close 用于涨停价计算 (分钟价是未复权真实价, 涨停规则定义在原始价上)。
+MINUTE_DAILY_PANEL_COLUMNS = frozenset({
+    "open", "high", "low", "close", "volume", "amount",
+    "raw_close", "raw_high", "raw_low",
+    "turnover_rate",
+    "signal_limit_up", "signal_limit_down", "signal_broken_limit_up",
+})
+MINUTE_INSTRUMENT_COLUMNS = frozenset({"name", "total_shares", "float_shares"})
+
+
+def minute_replay_feature_plan(daily_bars: int):
+    """分钟回测的日线面板加载计划。
+
+    execution_backend 用 polars_expr 走"按需计算信号"路径: enriched 分区只落
+    基础列, 涨停/炸板信号由 load_panel_for_backtest 的 compute_limit_signals
+    按 signal_columns 需求现算 (matrix_native 路径会跳过通用信号计算)。
+    """
+    # 函数级导入规避与 strategy.py 的循环依赖 (strategy 顶层导入本模块)。
+    from app.backtest.strategy import ResolvedFeaturePlan
+
+    return ResolvedFeaturePlan(
+        base_columns=MINUTE_DAILY_PANEL_COLUMNS,
+        intermediate_columns=frozenset(),
+        indicator_columns=frozenset(),
+        signal_columns=frozenset({
+            "signal_limit_up", "signal_limit_down", "signal_broken_limit_up",
+        }),
+        matrix_columns=frozenset(),
+        instrument_columns=MINUTE_INSTRUMENT_COLUMNS,
+        warmup_bars=max(daily_bars, 1),
+        full_feature_fallback=False,
+        execution_backend="polars_expr",
+    )
+
+
+def minute_panel_start(start: date, daily_bars: int) -> date:
+    """日线面板加载起点: 覆盖首个回测日的 daily_bars 交易日窗口。
+
+    N 个交易日约需 N*2 自然日 (周末/节假日), 再留 warmup 余量。
+    """
+    calendar_days = max(daily_bars, 1) * 2 + 30
+    return start - timedelta(days=calendar_days)
+
+
+def _trigger_hhmm(value) -> str:
+    """从 last_datetime 提取北京时间 "HH:MM" 触发分钟。
+
+    分区 datetime 为 UTC 存储 (tz-aware 或 naive-UTC), 统一折算到北京时区。
+    """
+    from app.market_time import CN_TZ
+
+    if hasattr(value, "astimezone"):
+        if value.tzinfo is None:
+            from datetime import timezone
+
+            value = value.replace(tzinfo=timezone.utc)
+        return value.astimezone(CN_TZ).strftime("%H:%M")
+    text = str(value or "")
+    if len(text) >= 16 and text[13] == ":":
+        return text[11:16]
+    return text[-5:] if text else ""
+
+
+def _scalar_limit_up_price(prev_close: float, limit_pct: float) -> float:
+    """与 polars_limit_price 同口径的标量涨停价 (整数分半进位)。"""
+    cents = int(prev_close * 100 + 0.5)
+    numerator = round((1 + limit_pct) * 100)
+    return ((cents * numerator + 50) // 100) / 100
+
+
+@dataclass
+class MinuteReplayHit:
+    """一个盘中入场信号: 触发分钟收盘买入。"""
+
+    trade_date: date
+    symbol: str
+    # 已按当日 复权close/原始close 比例折算到复权价系的入场价, 与日线出场价同尺度。
+    entry_price: float
+    trigger_time: str  # "HH:MM" — 触发分钟K的时间戳
+    score: float = 0.0
+
+
+@dataclass
+class MinuteReplayResult:
+    hits: list[MinuteReplayHit] = field(default_factory=list)
+    skipped_days: list[date] = field(default_factory=list)
+    replayed_days: int = 0
+    strategy_matches: int = 0
+    buy_limit_up: int = 0
+    elapsed_ms: float = 0.0
+
+
+class MinuteSignalReplayer:
+    """逐交易日回放分钟策略, 产出与实盘选股同源的入场命中。"""
+
+    def __init__(self, engine, strategy_engine: StrategyEngine) -> None:
+        # engine: BacktestEngine — 只用其 repo (分钟分区读取)。
+        self.engine = engine
+        self.strategy_engine = strategy_engine
+
+    def replay(
+        self,
+        strategy: StrategyDef,
+        *,
+        panel: pl.DataFrame,
+        start: date,
+        end: date,
+        params: dict,
+        overrides: dict,
+        pool: list[str] | None = None,
+        symbols: list[str] | None = None,
+        progress_cb: Callable[[dict], None] | None = None,
+        cancel_event=None,
+    ) -> MinuteReplayResult:
+        t0 = time.perf_counter()
+        result = MinuteReplayResult()
+        repo = self.engine.repo
+        if panel.is_empty():
+            return result
+
+        universe = symbols if symbols else panel.get_column("symbol").unique().to_list()
+        daily_bars = int(strategy.minute_daily_bars or 0)
+
+        # 面板交易日序列 (升序) — 日线窗口切片与缺分区日判定的基准。
+        panel_dates = panel.get_column("date").unique().sort().to_list()
+        date_to_window: dict[date, tuple[date, date]] = {}
+        for i, day in enumerate(panel_dates):
+            window_start = panel_dates[max(0, i - daily_bars)] if daily_bars > 0 else day
+            date_to_window[day] = (window_start, day)
+
+        # 逐分区日回放: 只回放 [start, end] 内有分钟分区的交易日。
+        minute_days = repo.list_minute_dates(start, end, "stock")
+        minute_day_set = set(minute_days)
+        replay_days = [day for day in panel_dates if start <= day <= end]
+        result.skipped_days = [day for day in replay_days if day not in minute_day_set]
+        total = len(minute_days)
+
+        # 逐标的的 T-1 原始收盘/复权收盘查表 (涨停价与复权折算用)。
+        prev_raw_close: dict[str, float] = {}
+        prev_name: dict[str, str] = {}
+        adj_factor: dict[str, float] = {}
+
+        for i, day in enumerate(minute_days):
+            if cancel_event is not None and cancel_event.is_set():
+                break
+            if progress_cb is not None:
+                progress_cb({
+                    "day": i + 1,
+                    "total": max(total, 1),
+                    "date": str(day),
+                })
+
+            history = repo.get_minute_by_dates(universe, [day], "stock")
+            if history.is_empty():
+                result.skipped_days.append(day)
+                continue
+
+            # 日线窗口: 截至 T-1 的完成态日K (index of last panel date < day)。
+            prior = [d for d in panel_dates if d < day]
+            if not prior:
+                # 面板起点之前的分区日 (窗口数据不足), 策略按数据不足自然不命中。
+                daily_history = pl.DataFrame()
+                current = pl.DataFrame()
+            else:
+                last_prior = prior[-1]
+                window_start, _ = date_to_window[last_prior]
+                daily_history = panel.filter(
+                    (pl.col("date") >= window_start) & (pl.col("date") <= last_prior)
+                ) if daily_bars > 0 else pl.DataFrame()
+                current = panel.filter(pl.col("date") == last_prior)
+
+                # T-1 收盘/名称 + T 日复权因子查表。
+                _refresh_day_lookups(prev_raw_close, prev_name, current, prior)
+                day_rows = panel.filter(pl.col("date") == day).select(
+                    "symbol", "close", "raw_close",
+                )
+                adj_factor.clear()
+                adj_factor.update(_adj_factors(day_rows))
+
+            context = StrategyDataContext(
+                asset_type="stock",
+                timeframe="1m",
+                as_of=day,
+                current=current if not current.is_empty() else None,
+                history=history,
+                daily_history=daily_history if not daily_history.is_empty() else None,
+            )
+            try:
+                run_result = self.strategy_engine.run(
+                    strategy.meta.get("id", ""),
+                    context,
+                    pool,
+                    params,
+                    overrides,
+                )
+            except ValueError:
+                # 单日执行失败 (如窗口缺列) 记为跳过, 不中断整个回放。
+                result.skipped_days.append(day)
+                continue
+
+            result.replayed_days += 1
+            result.strategy_matches += len(run_result.rows)
+            for row in run_result.rows:
+                symbol = row.get("symbol")
+                close = row.get("close")
+                if not symbol or close is None or float(close) <= 0:
+                    continue
+                raw_close = float(close)
+                name = prev_name.get(str(symbol), "")
+                prev = prev_raw_close.get(str(symbol))
+                # 涨停拒买: 触发分钟收盘已达当日涨停价 (按 T-1 原始收盘 + 板块规则)。
+                if prev is not None and prev > 0:
+                    limit_up = _scalar_limit_up_price(
+                        prev, price_limit_pct(str(symbol), day, is_risk_warning=is_risk_warning_name(name)),
+                    )
+                    if raw_close >= limit_up - 1e-9:
+                        result.buy_limit_up += 1
+                        continue
+                trigger = row.get("last_datetime")
+                trigger_time = _trigger_hhmm(trigger)
+                result.hits.append(MinuteReplayHit(
+                    trade_date=day,
+                    symbol=str(symbol),
+                    entry_price=raw_close * adj_factor.get(str(symbol), 1.0),
+                    trigger_time=trigger_time,
+                    score=float(run_result.scores.get(str(symbol), 0.0) or 0.0),
+                ))
+
+        result.elapsed_ms = round((time.perf_counter() - t0) * 1000, 1)
+        return result
+
+
+def _refresh_day_lookups(
+    prev_raw_close: dict[str, float],
+    prev_name: dict[str, str],
+    prior_snapshot: pl.DataFrame,
+    prior: list[date],
+) -> None:
+    """从 T-1 快照刷新逐标的原始收盘与名称查表 (涨停价/ST 判定用)。"""
+    if prior_snapshot.is_empty():
+        return
+    frame = prior_snapshot
+    if "raw_close" not in frame.columns:
+        frame = frame.with_columns(pl.col("close").alias("raw_close"))
+    if "name" not in frame.columns:
+        frame = frame.with_columns(pl.lit("").alias("name"))
+    prev_raw_close.clear()
+    prev_name.clear()
+    for symbol, raw_close, name in frame.select("symbol", "raw_close", "name").iter_rows():
+        prev_raw_close[str(symbol)] = float(raw_close) if raw_close is not None else 0.0
+        prev_name[str(symbol)] = str(name or "")
+
+
+def _adj_factors(day_rows: pl.DataFrame) -> dict[str, float]:
+    """T 日 复权close/原始close 比例: 把分钟原始价折算到复权价系。"""
+    factors: dict[str, float] = {}
+    if day_rows.is_empty() or "raw_close" not in day_rows.columns:
+        return factors
+    for symbol, close, raw_close in day_rows.select("symbol", "close", "raw_close").iter_rows():
+        if close and raw_close and float(raw_close) > 0:
+            factors[str(symbol)] = float(close) / float(raw_close)
+    return factors

backend/app/backtest/strategy.py

@@ -28,13 +28,20 @@ from app.backtest.matrix import (
     MatrixPipelineConfig,
     MatrixPrewarmCancelledError,
     MatrixStrategyPipeline,
+    SignalMatrix,
     apply_time_masks,
+    build_market_data_matrix,
     build_market_matrix,
     build_market_matrix_from_signals,
     rolling_mean,
     slice_market_data_matrix,
     slice_signal_matrix,
 )
+from app.backtest.minute_replay import (
+    MinuteSignalReplayer,
+    minute_panel_start,
+    minute_replay_feature_plan,
+)
 from app.backtest.minute_trigger import unsupported_minute_exit_signals
 from app.config import settings
 from app.indicators.pipeline import (
@@ -995,7 +1002,7 @@ class StrategyBacktestService:
                 s,
                 StrategyDataContext(
                     asset_type=config.asset_type,
-                    timeframe="1d",
+                    timeframe="1m" if s.execution_backend == "minute_filter" else "1d",
                     as_of=config.end,
                 ),
             )
@@ -1044,6 +1051,26 @@ class StrategyBacktestService:
             overrides.get("score_max"),
         )
 
+        if s.execution_backend == "minute_filter":
+            # 分钟策略回测: 逐交易日回放 filter_minute_history (与实盘选股同源),
+            # 信号分钟收盘价入场, 之后复用日K矩阵模拟的离场与组合管理。
+            return self._run_minute_backtest(
+                config, s, params, overrides,
+                stop_loss=stop_loss,
+                take_profit=take_profit,
+                trailing_stop=trailing_stop,
+                trailing_take_profit_activate=trailing_take_profit_activate,
+                trailing_take_profit_drawdown=trailing_take_profit_drawdown,
+                max_hold_days=max_hold_days,
+                score_min=score_min,
+                score_max=score_max,
+                progress_cb=progress_cb,
+                cancel_event=cancel_event,
+                result_policy=result_policy,
+                run_id=run_id,
+                t0=t0,
+            )
+
         try:
             if s.execution_backend == "composite":
                 # composite 回测: 子策略必须全为 matrix_native(否则 fail-closed),
@@ -1629,6 +1656,300 @@ class StrategyBacktestService:
             elapsed_ms=round(elapsed, 1),
         )
 
+    # ── 分钟策略回测: 逐日回放入场 + 日K矩阵离场 ──
+
+    def _run_minute_backtest(
+        self,
+        config: StrategyBacktestConfig,
+        s: StrategyDef,
+        params: dict,
+        overrides: dict,
+        *,
+        stop_loss,
+        take_profit,
+        trailing_stop,
+        trailing_take_profit_activate,
+        trailing_take_profit_drawdown,
+        max_hold_days,
+        score_min,
+        score_max,
+        progress_cb,
+        cancel_event,
+        result_policy: BacktestResultPolicy,
+        run_id: str,
+        t0: float,
+    ) -> StrategyBacktestResult:
+        def _err(msg: str) -> StrategyBacktestResult:
+            return StrategyBacktestResult(
+                run_id=run_id,
+                config=self._config_to_dict(config),
+                error=msg,
+                elapsed_ms=(time.perf_counter() - t0) * 1000,
+            )
+
+        if config.asset_type != "stock":
+            return _err("分钟策略回测当前仅支持 A 股 (stock)")
+        if config.exit_fill == "signal_next_minute":
+            return _err("分钟策略回测暂不支持「信号触发卖出」离场口径")
+
+        minute_days = self.engine.repo.list_minute_dates(config.start, config.end, "stock")
+        if not minute_days:
+            earliest = self.engine.repo.earliest_minute_date()
+            hint = f"本地分钟K最早到 {earliest}, " if earliest else "本地无分钟K数据, "
+            return _err(
+                f"回测区间内无分钟K数据: {hint}请先用「扩展分钟K历史」拉取, 或开启盘中分钟增量"
+            )
+
+        # 日线面板一次加载: 覆盖首个回测日的日线窗口 + 模拟区间 (含 full 模式尾部)。
+        daily_bars = int(s.minute_daily_bars or 0)
+        feature_plan = minute_replay_feature_plan(daily_bars)
+        load_start = minute_panel_start(config.start, daily_bars)
+        full_horizon_days = int(max_hold_days or config.holding_days or 5)
+        load_end = config.end
+        if config.mode == "full":
+            load_end = config.end + timedelta(days=(full_horizon_days + 5) * 2)
+        sim_end = load_end if config.mode == "full" else config.end
+
+        timing_ms: dict[str, float] = {}
+        t_load = time.perf_counter()
+        try:
+            panel = self.engine.load_panel_for_backtest(
+                config.symbols,
+                load_start,
+                load_end,
+                feature_plan,
+                asset_type="stock",
+            )
+        except (ValueError, OSError, pl.exceptions.PolarsError) as e:
+            return _err(f"回测特征准备失败: {e}")
+        timing_ms["load_panel"] = round((time.perf_counter() - t_load) * 1000, 1)
+        if panel.is_empty():
+            return _err("无日线数据, 请检查日期范围或先运行盘后管道")
+
+        replayer = MinuteSignalReplayer(self.engine, self.strategy_engine)
+        replay = replayer.replay(
+            s,
+            panel=panel,
+            start=config.start,
+            end=config.end,
+            params=params,
+            overrides=overrides,
+            symbols=config.symbols,
+            progress_cb=progress_cb,
+            cancel_event=cancel_event,
+        )
+        timing_ms["minute_replay"] = replay.elapsed_ms
+        if cancel_event is not None and cancel_event.is_set():
+            return StrategyBacktestResult(
+                run_id=run_id,
+                config=self._config_to_dict(config),
+                error="cancelled",
+                elapsed_ms=round((time.perf_counter() - t0) * 1000, 1),
+            )
+        if not replay.hits:
+            skipped_hint = (
+                f" (区间内 {len(replay.skipped_days)} 个交易日缺分钟K分区被跳过)"
+                if replay.skipped_days else ""
+            )
+            return _err("在指定区间内未产生买入信号" + skipped_hint)
+
+        # 日频信号网格: 正式区间面板 → time x asset 矩阵, 命中格写入入场价覆盖。
+        sim_panel = panel.filter(
+            (pl.col("date") >= config.start) & (pl.col("date") <= sim_end)
+        )
+        if sim_panel.is_empty():
+            return _err("正式回测区间内无数据")
+        axis_dates = sim_panel.get_column("date").unique().sort().to_list()
+        # 轴顺序必须与 build_market_data_matrix 的 _encode_axes 一致 (unique().sort()),
+        # 否则 (time, asset) 下标指向错误的标的。
+        axis_symbols = sim_panel.get_column("symbol").cast(pl.Utf8).unique().sort().to_list()
+        time_index = {day: i for i, day in enumerate(axis_dates)}
+        asset_index = {sym: i for i, sym in enumerate(axis_symbols)}
+        shape = (len(axis_dates), len(axis_symbols))
+
+        entry = np.zeros(shape, dtype=np.uint8)
+        score = np.zeros(shape, dtype=np.float32)
+        entry_price_override = np.full(shape, np.nan, dtype=np.float32)
+        trigger_times: dict[tuple[str, date], str] = {}
+        dropped_axis_hits = 0
+        for hit in replay.hits:
+            time_id = time_index.get(hit.trade_date)
+            asset_id = asset_index.get(hit.symbol)
+            if time_id is None or asset_id is None:
+                dropped_axis_hits += 1
+                continue
+            entry[time_id, asset_id] = 1
+            score[time_id, asset_id] = hit.score
+            entry_price_override[time_id, asset_id] = hit.entry_price
+            trigger_times[(hit.symbol, hit.trade_date)] = hit.trigger_time
+        raw_candidates = int(entry.sum())
+        entry.setflags(write=False)
+        score.setflags(write=False)
+        entry_price_override.setflags(write=False)
+        exit_mask = np.zeros(shape, dtype=np.uint8)
+        exit_mask.setflags(write=False)
+        codes = np.zeros(shape, dtype=np.int16)
+        codes.setflags(write=False)
+        signals = SignalMatrix(
+            entry=entry,
+            exit=exit_mask,
+            score=score,
+            entry_signal_code=codes,
+            exit_signal_code=codes,
+            entry_signal_ids=(),
+            exit_signal_ids=(),
+        )
+
+        matcher_config = MatcherConfig(
+            matching=config.matching,
+            entry_fill="close_t",
+            exit_fill=config.exit_fill,
+            fees_pct=config.fees_pct,
+            commission_pct=config.commission_pct,
+            stamp_tax_pct=config.stamp_tax_pct,
+            slippage_bps=config.slippage_bps,
+            stop_loss_pct=stop_loss,
+            take_profit_pct=take_profit,
+            trailing_stop_pct=trailing_stop,
+            trailing_take_profit_activate_pct=trailing_take_profit_activate,
+            trailing_take_profit_drawdown_pct=trailing_take_profit_drawdown,
+            max_hold_days=max_hold_days,
+            max_positions=config.max_positions,
+            max_exposure_pct=config.max_exposure_pct,
+            score_min=score_min,
+            score_max=score_max,
+            initial_capital=config.initial_capital,
+            position_sizing=config.position_sizing,
+            # 分钟策略的成交价由 entry_price_override 提供 (触发分钟收盘),
+            # 不再叠加日线口径的分钟成交细化。
+            minute_fill=False,
+        )
+
+        t_matrix = time.perf_counter()
+        market_data = build_market_data_matrix(sim_panel)
+        market_matrix = build_market_matrix_from_signals(
+            market_data,
+            signals,
+            # 入场即信号日盘中 (分钟价覆盖), 离场沿用日K口径。
+            entry_delay_bars=0,
+            exit_delay_bars=1 if matcher_config.exit_fill == "open_t+1" else 0,
+            entry_price_override=entry_price_override,
+        )
+        timing_ms["matrix_build"] = round((time.perf_counter() - t_matrix) * 1000, 1)
+        del sim_panel, market_data
+
+        t_sim = time.perf_counter()
+        if config.mode == "full":
+            result = self.engine.simulate_independent_market_matrix(
+                market_matrix,
+                raw_candidates,
+                matcher_config,
+                progress_cb,
+                cancel_event,
+                result_policy.simulation_options(),
+            )
+        else:
+            result = self.engine.simulate_market_matrix(
+                market_matrix,
+                matcher_config,
+                progress_cb,
+                cancel_event,
+                result_policy.simulation_options(),
+            )
+        timing_ms["simulate"] = round((time.perf_counter() - t_sim) * 1000, 1)
+        timing_ms["statistics"] = float(result.stats.pop("statistics_ms", 0.0))
+
+        if cancel_event is not None and cancel_event.is_set():
+            return StrategyBacktestResult(
+                run_id=run_id,
+                config=self._config_to_dict(config),
+                error="cancelled",
+                elapsed_ms=round((time.perf_counter() - t0) * 1000, 1),
+            )
+        if result.stats.get("error"):
+            return _err(result.stats["error"])
+
+        execution = result.stats.get("execution") or {}
+        execution["buy_limit_up"] = int(execution.get("buy_limit_up", 0)) + replay.buy_limit_up
+        result.stats["execution"] = execution
+        timing_ms["total"] = round((time.perf_counter() - t0) * 1000, 1)
+        result.stats["timing_ms"] = timing_ms
+        result.stats["panel_rows"] = int(len(axis_dates) * len(axis_symbols))
+        result.stats["panel_columns"] = 0
+        result.stats["feature_columns"] = 0
+        result.stats["execution_backend"] = s.execution_backend
+        result.stats["selection"] = {
+            "strategy_matches": replay.strategy_matches,
+            "entry_candidates": raw_candidates,
+            "entry_trigger_filtered": max(replay.strategy_matches - raw_candidates, 0),
+            "entry_trigger_enabled": False,
+        }
+        result.stats["minute_replay"] = {
+            "replayed_days": replay.replayed_days,
+            "skipped_days": [str(day) for day in replay.skipped_days[:50]],
+            "skipped_day_count": len(replay.skipped_days),
+            "dropped_axis_hits": dropped_axis_hits,
+        }
+
+        benchmark_curve = (
+            self._build_benchmark_curve(config.start, config.end)
+            if result_policy.include_benchmark
+            else []
+        )
+        strategy_info = {
+            "id": s.meta.get("id", config.strategy_id),
+            "name": s.meta.get("name", config.strategy_id),
+            "description": s.meta.get("description", ""),
+            "entry_signals": [],
+            "exit_signals": [],
+            "stop_loss": stop_loss,
+            "take_profit": take_profit,
+            "trailing_stop": trailing_stop,
+            "trailing_take_profit_activate": trailing_take_profit_activate,
+            "trailing_take_profit_drawdown": trailing_take_profit_drawdown,
+            "max_hold_days": max_hold_days,
+            "full_horizon_days": full_horizon_days,
+            "score_min": score_min,
+            "score_max": score_max,
+            "source": s.source,
+            "execution_backend": s.execution_backend,
+        } if result_policy.include_strategy_info else {}
+
+        trades = (
+            [self._trade_to_dict(t) for t in result.trades]
+            if result_policy.include_trades
+            else []
+        )
+        # 入场时间戳补分钟: 交易记录携带触发分钟 (HH:MM), 与日线回测的纯日期区分。
+        for trade in trades:
+            entry_text = str(trade.get("entry_date") or "")
+            try:
+                key = (str(trade.get("symbol")), date.fromisoformat(entry_text[:10]))
+            except ValueError:
+                continue
+            trigger = trigger_times.get(key)
+            if trigger:
+                trade["entry_date"] = f"{entry_text[:10]} {trigger}"
+
+        selected_stats = result_policy.select_stats(result.stats)
+        elapsed = (time.perf_counter() - t0) * 1000
+        return StrategyBacktestResult(
+            run_id=run_id,
+            config=self._config_to_dict(config),
+            stats=selected_stats,
+            equity_curve=result.equity_curve if result_policy.include_curves else [],
+            drawdown_curve=result.drawdown_curve if result_policy.include_curves else [],
+            benchmark_curve=benchmark_curve,
+            trades=trades,
+            per_symbol_stats=(
+                result.per_symbol_stats
+                if result_policy.include_per_symbol_stats
+                else []
+            ),
+            strategy_info=strategy_info,
+            elapsed_ms=round(elapsed, 1),
+        )
+
     # ── 全量模拟 (选股能力统计, 不建组合不算净值) ──
 
     def _run_full_simulation(

backend/app/tickflow/repository.py

@@ -1866,6 +1866,29 @@ class KlineRepository:
             return None
         return None
 
+    def list_minute_dates(self, start: date, end: date, asset_type: str = "stock") -> list[date]:
+        """枚举 [start, end] 内存在的分钟K分区日 (目录名直读, 零 parquet 扫描)。
+
+        分钟回测按交易日精确对日: 缺分区的日子由调用方显式跳过,
+        不做"回退最近分区" (那是实盘选股的语义, 回放会串日)。
+        """
+        dirname = "kline_minute" if asset_type == "stock" else f"kline_{asset_type}_minute"
+        minute_dir = self.store.data_dir / dirname
+        if not minute_dir.exists():
+            return []
+        out: list[date] = []
+        for entry in minute_dir.iterdir():
+            if not (entry.is_dir() and entry.name.startswith("date=")):
+                continue
+            try:
+                day = date.fromisoformat(entry.name[5:])
+            except ValueError:
+                continue
+            if start <= day <= end:
+                out.append(day)
+        out.sort()
+        return out
+
     def latest_daily_date(self) -> date | None:
         """本地日K数据的最新日期。"""
         try: