feat(backtest): 分钟K精确回测 — 穿越价/VWAP 成交 + Pro+ 门控

- engine.py: MatcherConfig 加 minute_fill; _resolve_minute_fill (穿越价/VWAP/降级)
  + _load_minute_for_fills; simulate_portfolio/independent_candidates 接入
- strategy.py: StrategyBacktestConfig 加 minute_fill
- backtest.py: strategy_stream 加 minute_fill 参数 + Pro+ 门控 + 数据范围检查
- repository.py: 新增 get_minute_range (多symbol x 日期范围)
- backtestTask.ts/StrategyBacktest.tsx: 激活 highGranularity 开关 + Pro+ 门控

backend/app/api/backtest.py

@@ -294,8 +294,9 @@ def _make_job_key(
     mode: str = "position", holding_days: int = 5,
     commission_pct: float | None = None, stamp_tax_pct: float | None = None,
     asset_type: str = "stock",
+    minute_fill: bool = False,
 ) -> str:
-    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}"
+    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}|{minute_fill}"
     return hashlib.md5(raw.encode()).hexdigest()[:12]
 
 
@@ -322,6 +323,7 @@ async def strategy_stream(
     mode: str = "position",
     holding_days: int = 5,
     asset_type: str = "stock",
+    minute_fill: bool = False,
 ):
     """SSE 流式策略回测: 实时推送进度, 完成后推送结果, 支持重连 (刷新/切页后恢复)。
 
@@ -363,6 +365,7 @@ async def strategy_stream(
         mode, holding_days,
         commission_pct, stamp_tax_pct,
         asset_type=asset_type,
+        minute_fill=minute_fill,
     )
 
     _cleanup_stale_jobs()
@@ -383,6 +386,22 @@ async def strategy_stream(
             yield f"event: error\ndata: {json.dumps({'message': BACKTEST_SERVER_GUARD_MESSAGE}, ensure_ascii=False)}\n\n"
             return
 
+        # 分钟K精确回测: Pro+ 门控 + 数据范围检查
+        if minute_fill:
+            capset = request.app.state.capabilities
+            from app.tickflow.capabilities import Cap
+            if not capset.has(Cap.KLINE_MINUTE_BATCH):
+                yield f"event: error\ndata: {json.dumps({'message': '分钟K精确回测需要 Pro+ 权限 (kline.minute.batch)'}, ensure_ascii=False)}\n\n"
+                return
+            # 检查本地分钟K历史是否覆盖回测区间
+            repo = request.app.state.repo
+            earliest_minute = repo.earliest_minute_date() if hasattr(repo, "earliest_minute_date") else None
+            if earliest_minute is not None and start_date < earliest_minute:
+                msg = (f"本地分钟K历史最早到 {earliest_minute}, 无法覆盖回测起始日 {start_date}。"
+                       f"请先用「扩展分钟K历史」功能拉取更多数据, 或缩小回测区间。")
+                yield f"event: error\ndata: {json.dumps({'message': msg}, ensure_ascii=False)}\n\n"
+                return
+
         # 如果是新任务, 启动回测线程
         if is_new and not job.done:
             cfg = StrategyBacktestConfig(
@@ -406,6 +425,7 @@ async def strategy_stream(
                 mode=mode,
                 holding_days=int(holding_days),
                 asset_type=asset_type,
+                minute_fill=minute_fill,
             )
 
             def _run_backtest():

backend/app/backtest/engine.py

@@ -54,6 +54,9 @@ class MatcherConfig:
     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 (向后兼容)。
@@ -508,6 +511,45 @@ class BacktestEngine:
         # 撮合价: 建仓/清仓各自独立选列。
         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, _mdf in _loaded.items():
+                    if not _mdf.is_empty():
+                        minute_cache[_key] = _mdf.to_numpy()
+
+        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)
@@ -665,7 +707,10 @@ class BacktestEngine:
                 _count(block_reason)
                 return False
 
-            exit_price = float(exit_price_override) if exit_price_override is not None else float(exit_prices[idx])
+            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)
@@ -729,7 +774,7 @@ class BacktestEngine:
                 _count("sell_no_future")
                 continue
 
-            entry_price = float(entry_prices[entry_idx])
+            entry_price = _refill_price(entry_idx, "buy", float(entry_prices[entry_idx]))
             pos = {
                 "symbol": sym,
                 "name": str(names[entry_idx] or ""),
@@ -790,6 +835,102 @@ class BacktestEngine:
 
         return self._calc_independent_candidate_result(trades, n_candidates, execution_stats)
 
+    # ── 分钟K精确成交 ──────────────────────────────────
+
+    @staticmethod
+    def _resolve_minute_fill(
+        minute_rows: np.ndarray,
+        ref_price: float | None,
+        side: str,
+    ) -> float | None:
+        """用当日分钟K确定精确成交价。
+
+        Args:
+            minute_rows: structured numpy array, 字段含 open/high/low/close/volume/amount
+            ref_price: 信号参考线价格 (如 MA5 值); None 表示无参考线
+            side: "buy" 或 "sell", 决定穿越方向
+
+        Returns:
+            精确成交价, 或 None (降级到日K口径)
+        """
+        if minute_rows is None or len(minute_rows) == 0:
+            return None
+
+        opens = minute_rows["open"].astype(float)
+        highs = minute_rows["high"].astype(float)
+        lows = minute_rows["low"].astype(float)
+        closes = minute_rows["close"].astype(float)
+        volumes = minute_rows["volume"].astype(float) if "volume" in minute_rows.dtype.names else None
+        amounts = minute_rows["amount"].astype(float) if "amount" in minute_rows.dtype.names 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
+
+    @staticmethod
+    def _load_minute_for_fills(
+        repo,
+        symbols: list[str],
+        dates_needed: set,
+        asset_type: str,
+    ) -> dict:
+        """批量加载回测区间内触发日的分钟K, 返回 {(symbol, date_str): minute_df}。
+
+        dates_needed: 需要分钟数据的日期集合 (set of date strings "YYYY-MM-DD")
+        """
+        if not symbols or not dates_needed:
+            return {}
+        from datetime import date as _date
+        sorted_dates = sorted(dates_needed)
+        start = _date.fromisoformat(sorted_dates[0])
+        end = _date.fromisoformat(sorted_dates[-1])
+        try:
+            df = repo.get_minute_range(symbols, start, end, asset_type=asset_type)
+        except Exception as e:  # noqa: BLE001
+            logger.warning("minute fill data load failed: %s", e)
+            return {}
+        if df.is_empty():
+            return {}
+
+        cache: dict = {}
+        for row in df.iter_rows(named=True):
+            dt = row.get("datetime")
+            if dt is None:
+                continue
+            d_str = str(dt)[:10]
+            sym = row["symbol"]
+            key = (sym, d_str)
+            if key not in cache:
+                cache[key] = []
+            cache[key].append(row)
+        # 转 DataFrame per key
+        result: dict = {}
+        for key, rows in cache.items():
+            result[key] = pl.DataFrame(rows)
+        return result
+
     def simulate_portfolio(
         self,
         panel: pl.DataFrame,
@@ -891,6 +1032,52 @@ class BacktestEngine:
         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, mdf in loaded.items():
+                    if not mdf.is_empty():
+                        minute_cache[key] = mdf.to_numpy()
+
+        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] = {
@@ -991,7 +1178,10 @@ class BacktestEngine:
         ) -> None:
             nonlocal cash
             pos = positions.pop(sym)
-            exit_price = float(exit_price_override) if exit_price_override is not None else float(exit_prices[idx])
+            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"]
@@ -1192,7 +1382,7 @@ class BacktestEngine:
                 if allocation <= 0:
                     _count("buy_exposure")
                     continue
-                entry_price = float(entry_prices[idx])
+                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:

backend/app/backtest/strategy.py

@@ -45,6 +45,8 @@ class StrategyBacktestConfig:
     mode: Literal["position", "full"] = "position"
     asset_type: str = "stock"
     holding_days: int = 5
+    # 分钟K精确成交: 开启后用当日分钟K确定穿越价/VWAP (需 Pro+ 分钟K能力)
+    minute_fill: bool = False
 
     def __post_init__(self) -> None:
         if self.entry_fill is None:
@@ -216,6 +218,7 @@ class StrategyBacktestService:
             score_max=score_max,
             initial_capital=config.initial_capital,
             position_sizing=config.position_sizing,
+            minute_fill=config.minute_fill,
         )
         # 撮合 — full 为全候选独立执行；position 为账户级仓位模拟。
         if config.mode == "full":

backend/app/tickflow/repository.py

@@ -1275,6 +1275,38 @@ class KlineRepository:
             logger.warning("批量分钟K查询失败: %s", e)
             return pl.DataFrame()
 
+    def get_minute_range(
+        self,
+        symbols: list[str],
+        start: date,
+        end: date,
+        asset_type: str = "stock",
+    ) -> pl.DataFrame:
+        """多 symbol × 日期范围的分钟K查询 (分钟K精确回测用)。
+
+        一次 scan_parquet + predicate pushdown 读多只股票在 [start, end] 内的所有分钟K。
+        返回列: symbol, datetime, open, high, low, close, volume, amount。
+        """
+        if not symbols:
+            return pl.DataFrame()
+        try:
+            lf = pl.scan_parquet(self._minute_glob_for(asset_type))
+            available = set(lf.collect_schema().names())
+            select_cols = [c for c in ["symbol", "datetime", "open", "high", "low", "close", "volume", "amount"] if c in available]
+            return (
+                lf.select(select_cols)
+                .filter(
+                    pl.col("symbol").is_in(symbols)
+                    & (pl.col("datetime").dt.date() >= start)
+                    & (pl.col("datetime").dt.date() <= end)
+                )
+                .sort(["symbol", "datetime"])
+                .collect(streaming=True)
+            )
+        except Exception as e:  # noqa: BLE001
+            logger.warning("分钟K范围查询失败: %s", e)
+            return pl.DataFrame()
+
     # ================================================================
     # Polars 查询内部方法
     # ================================================================

frontend/src/lib/backtestTask.ts

@@ -180,6 +180,7 @@ export function startBacktest(params: {
   mode?: 'position' | 'full'
   holding_days?: number
   asset_type?: 'stock' | 'etf'
+  minute_fill?: boolean
 }): void {
   // 取消之前的任务状态
   if (eventSource) {
@@ -212,6 +213,7 @@ export function startBacktest(params: {
     mode: params.mode,
     holding_days: params.holding_days,
     asset_type: params.asset_type,
+    minute_fill: params.minute_fill,
   })
 
   // 存 reconnect 信息 (刷新后用)

frontend/src/pages/backtest/StrategyBacktest.tsx

@@ -10,7 +10,6 @@ import {
   type StrategyParamDef,
 } from '@/lib/api'
 import { QK } from '@/lib/queryKeys'
-import { tierRank } from '@/lib/capability-labels'
 import { storage } from '@/lib/storage'
 import { fmtPct, fmtPrice, priceColorClass } from '@/lib/format'
 import { boardTag } from '@/lib/board'
@@ -728,10 +727,10 @@ export function StrategyBacktest() {
   const [simMode, setSimMode] = useState<'position' | 'full'>(saved?.mode ?? 'position')
   const [holdingDays, setHoldingDays] = useState(saved?.holdingDays ?? '5')
   const [settingsOpen, setSettingsOpen] = useState(false)
-  // 高颗粒回测（分钟K精确回测）— 开发中，Starter+ 功能
+  // 分钟K精确回测: 用当日分钟K确定精确成交价 (穿越价/VWAP), 需 Pro+ 分钟K能力
   const [highGranularity, setHighGranularity] = useState(false)
   const { data: caps } = useCapabilities()
-  const isFreeTier = tierRank(caps?.label ?? '') < 1
+  const hasMinuteBatch = !!caps?.capabilities?.['kline.minute.batch']
   const [rangeSettingsOpen, setRangeSettingsOpen] = useState(false)
   const [quickRanges, setQuickRanges] = useState(loadQuickRanges)
   const [settingsTab, setSettingsTab] = useState<AdvancedSettingsTab>('params')
@@ -864,6 +863,7 @@ export function StrategyBacktest() {
       overrides,
       mode: simMode,
       holding_days: Number(holdingDays) || 5,
+      minute_fill: highGranularity,
     })
   }
 
@@ -1093,22 +1093,18 @@ export function StrategyBacktest() {
         <div>
           <div className="flex items-center justify-between mb-1.5">
             <label className="text-xs font-medium text-secondary">选择策略</label>
-            {/* 高颗粒回测（分钟K）— 开发中占位 */}
+            {/* 分钟K精确回测 */}
             <div className="flex items-center gap-1">
               <Gauge className={`h-3 w-3 ${highGranularity ? 'text-amber-400' : 'text-muted/50'}`} />
               <button
-                onClick={() => {
-                  if (isFreeTier) return
-                  // 功能开发中，暂不实际启用
-                  setHighGranularity(v => !v)
-                }}
-                disabled={isFreeTier}
-                title={isFreeTier
-                  ? '高颗粒回测（分钟K精确回测）：需 Starter+ 档位'
-                  : '高颗粒回测（分钟K精确回测）：切换后结合每日分钟K更精确回测。⚠️ 开发中，且会显著影响性能、回测很慢。'
+                onClick={() => { if (!hasMinuteBatch) return; setHighGranularity(v => !v) }}
+                disabled={!hasMinuteBatch}
+                title={!hasMinuteBatch
+                  ? '分钟K精确回测：需 Pro+ 权限 (分钟K批量)'
+                  : '分钟K精确回测：用当日分钟K确定精确成交价（穿越价/VWAP），比收盘价更真实。⚠️ 回测速度会变慢。'
                 }
                 className={`group relative inline-flex h-3.5 w-6 items-center rounded-full shrink-0 transition-colors duration-200 ${
-                  isFreeTier ? 'bg-elevated opacity-50 cursor-not-allowed'
+                  !hasMinuteBatch ? 'bg-elevated opacity-50 cursor-not-allowed'
                   : highGranularity ? 'bg-amber-500 cursor-pointer'
                   : 'bg-elevated cursor-pointer'
                 }`}
@@ -1118,19 +1114,18 @@ export function StrategyBacktest() {
                 }`} />
               </button>
               <span className={`text-[9px] font-medium ${highGranularity ? 'text-amber-400' : 'text-muted/50'}`}>分钟K</span>
-              {isFreeTier && (
-                <span className="text-[8px] text-accent/70 font-medium bg-accent/10 px-1 py-px rounded">Starter+</span>
+              {!hasMinuteBatch && (
+                <span className="text-[8px] text-accent/70 font-medium bg-accent/10 px-1 py-px rounded">Pro+</span>
               )}
             </div>
           </div>
-          {/* 高颗粒开启时的警告条 */}
-          {highGranularity && !isFreeTier && (
+          {/* 分钟K开启时的提示条 */}
+          {highGranularity && hasMinuteBatch && (
             <div className="mb-2 flex items-start gap-1.5 rounded-btn border border-amber-400/30 bg-amber-400/5 px-2 py-1.5">
               <Zap className="h-3 w-3 text-amber-400 shrink-0 mt-px" />
               <div className="text-[10px] leading-snug text-amber-400/90">
-                <span className="font-medium">高颗粒回测（开发中）</span>
-                ：将结合每日分钟K进行更精确的回测。
-                <span className="text-amber-400/70"> ⚠️ 此功能尚未完成，且开启后会显著拖慢回测速度、占用大量资源。</span>
+                <span className="font-medium">分钟K精确回测</span>
+                ：信号触发日用当日分钟K确定成交价（均线类信号按穿越价, 其他按 VWAP 均价）。需本地有足够的分钟K历史, 回测速度会变慢。
               </div>
             </div>
           )}