feat(backtest): P2a 面板路径 prepare-once — 参数网格共享特征面板, 调参循环 1.6-1.9×

默认后端 (polars_expr/python_history_legacy) 的优化器此前每组参数都完整
重跑 载盘/指标计算/信号注入/扩展表 join (profile: 占每组成本 ~35-60%);
只有 matrix_native 有 prepare-once。本提交补齐面板路径:

- PreparedPanelBacktest: 逐组解析依赖计划取并集 (参数驱动的列差异由并集
  覆盖), 一次加载全组复用; run 逐组做 签名校验 + 所需列子集校验, 缺列
  显式报错不静默回退 (fail-closed, 掩盖计划漂移比失败更危险)。
- _merge_resolved_feature_plans 放宽为单一后端即可 (composite 的全
  matrix_native 语义不变); _panel_backend 谓词与 run 面板分支同口径。
- 优化器: polars_expr 分支 prepare 一次 + 全部试算与最优复跑复用同一面板;
  进度新增 optimizer_prepare 事件 (shared_panel_bytes)。
- 基准 scripts/bench_optimizer_prepare.py 同进程交错 A/B (消除页缓存
  顺序偏置): 15 组实测 trials 1.72×, 含 prepare 1.62× (6 组 1.92×/1.64×);
  stats 剔除计时字段后与逐组独立加载完全一致, 交易逐笔一致。
- 回归: tests/backtest 408 全过; ruff 零新增 (optimizer/strategy 告警
  与 HEAD 逐条相同)。

下一步 (P2a-2): build_market_matrix 的面板编码 (build_market_data_matrix
+_encode_axes, 参数无关, 占剩余每组成本 ~55%) 提到 prepare 复用, 预期
含 prepare 越过 2×; 其后 P2b L2 内容寻址物化。

Signed-off-by: shy3130 <shy3130@qq.com>

backend/app/backtest/optimizer.py

@@ -217,7 +217,11 @@ class StrategyOptimizer:
         rss_sampler: PhaseRssSampler | None = None,
         prepared_market_data=None,
     ) -> dict:
-        from app.backtest.strategy import BacktestResultPolicy, StrategyBacktestConfig
+        from app.backtest.strategy import (
+            BacktestResultPolicy,
+            StrategyBacktestConfig,
+            _panel_backend,
+        )
 
         t0 = time.perf_counter()
         if cfg.objective not in VALID_OBJECTIVES:
@@ -247,6 +251,7 @@ class StrategyOptimizer:
         ]
 
         prepared = None
+        prepared_panel = None
         prepare_ms = 0.0
         trials_ms = 0.0
         final_backtest_ms = 0.0
@@ -260,20 +265,13 @@ class StrategyOptimizer:
             if cancel_event is not None and cancel_event.is_set():
                 return None
             # 单组异常必须隔离: 一组失败不能丢弃全部已完成结果。
+            run_kwargs: dict = {"cancel_event": cancel_event, "result_policy": trial_policy}
+            if prepared is not None:
+                run_kwargs["prepared"] = prepared
+            elif prepared_panel is not None:
+                run_kwargs["prepared_panel"] = prepared_panel
             try:
-                if prepared is None:
-                    res = self.service.run(
-                        bt_cfg,
-                        cancel_event=cancel_event,
-                        result_policy=trial_policy,
-                    )
-                else:
-                    res = self.service.run(
-                        bt_cfg,
-                        cancel_event=cancel_event,
-                        prepared=prepared,
-                        result_policy=trial_policy,
-                    )
+                res = self.service.run(bt_cfg, **run_kwargs)
             except Exception as e:  # 隔离单组失败, 记录后继续, 不拖垮整批
                 logger.warning("参数组 %s 回测异常: %r", combo, e)
                 return {"params": combo, "error": repr(e), "objective_raw": None, "_sort": float("-inf")}
@@ -330,6 +328,24 @@ class StrategyOptimizer:
                         "shared_matrix_bytes": prepared.market_data.nbytes,
                         "elapsed_ms": prepare_ms,
                     })
+            elif (
+                _panel_backend(getattr(s, "execution_backend", "polars_expr"))
+                and not cancelled_before_prepare
+            ):
+                # P2a: 默认后端同样 prepare-once — 全组依赖计划取并集, 载一次
+                # 共享特征面板, 每组只跑过滤/评分/模拟。
+                prepare_started = time.perf_counter()
+                prepared_panel = self.service.prepare_panel_optimization(backtest_configs)
+                prepare_ms = round((time.perf_counter() - prepare_started) * 1000, 1)
+                if progress_cb is not None:
+                    progress_cb({
+                        "type": "optimizer_prepare",
+                        "done": 0,
+                        "total": n_total,
+                        "best_score": None,
+                        "shared_panel_bytes": prepared_panel.panel.estimated_size(),
+                        "elapsed_ms": prepare_ms,
+                    })
 
             trials_started = time.perf_counter()
             if rss_sampler is not None:
@@ -383,10 +399,14 @@ class StrategyOptimizer:
                 final_started = time.perf_counter()
                 if rss_sampler is not None:
                     rss_sampler.reset_phase()
+                final_run_kwargs: dict = {"cancel_event": cancel_event}
+                if prepared is not None:
+                    final_run_kwargs["prepared"] = prepared
+                elif prepared_panel is not None:
+                    final_run_kwargs["prepared_panel"] = prepared_panel
                 final_result = self.service.run(
                     best_config,
-                    cancel_event=cancel_event,
-                    prepared=prepared,
+                    **final_run_kwargs,
                 )
                 final_backtest_ms = round((time.perf_counter() - final_started) * 1000, 1)
                 if rss_sampler is not None:
@@ -411,6 +431,8 @@ class StrategyOptimizer:
                 "effective_workers": 1,
                 "shared_market_data": prepared is not None,
                 "shared_market_data_bytes": prepared.market_data.nbytes if prepared is not None else 0,
+                "shared_panel": prepared_panel is not None,
+                "shared_panel_bytes": prepared_panel.panel.estimated_size() if prepared_panel is not None else 0,
                 "prepare_ms": prepare_ms,
                 "timing_ms": {
                     "prepare": prepare_ms,

backend/app/backtest/strategy.py

@@ -102,8 +102,8 @@ def _merge_resolved_feature_plans(
     if not plans:
         raise ValueError("cannot merge an empty feature plan list")
     backends = {plan.execution_backend for plan in plans}
-    if backends != {"matrix_native"}:
-        raise ValueError("shared MarketDataMatrix preparation only supports matrix_native")
+    if len(backends) != 1:
+        raise ValueError("shared preparation requires a single execution backend")
 
     def _union(field: str) -> frozenset[str]:
         merged: set[str] = set()
@@ -120,7 +120,7 @@ def _merge_resolved_feature_plans(
         instrument_columns=_union("instrument_columns"),
         warmup_bars=max(plan.warmup_bars for plan in plans),
         full_feature_fallback=any(plan.full_feature_fallback for plan in plans),
-        execution_backend="matrix_native",
+        execution_backend=backends.pop(),
         fundamental_columns=_union("fundamental_columns"),
     )
 
@@ -780,6 +780,28 @@ class PreparedMatrixBacktest:
     compute_cache: MatrixComputeCache
 
 
+def _panel_backend(execution_backend: str) -> bool:
+    """该执行后端走 polars 面板路径 (run 的 else 分支): polars_expr 与
+    python_history_legacy (动态 filter_history) 等一切非矩阵/复合/分钟后端。"""
+    return execution_backend not in ("matrix_native", "composite", "minute_filter")
+
+
+@dataclass(frozen=True)
+class PreparedPanelBacktest:
+    """polars_expr 参数网格的共享特征面板 (P2a) — 一次加载, 全部试算复用。
+
+    面板列是全部参数组依赖计划的并集; run 逐组做子集校验 (fail-closed),
+    缺列直接报错而不是静默回退加载 — 静默回退会掩盖计划漂移, 且优化器
+    正在拿"部分组快、部分组慢"的混合结果, 不如整批显式失败。"""
+
+    signature: tuple
+    panel: pl.DataFrame
+    columns: frozenset[str]
+    load_start: date
+    load_end: date
+    sim_end: date
+
+
 class StrategyBacktestService:
     def __init__(
         self,
@@ -1107,6 +1129,81 @@ class StrategyBacktestService:
             compute_cache=compute_cache,
         )
 
+    def prepare_panel_optimization(
+        self,
+        configs: list[StrategyBacktestConfig],
+    ) -> PreparedPanelBacktest:
+        """polars_expr 参数网格的共享特征面板 (P2a): 逐组解析依赖计划取并集,
+        一次加载全组复用 — 消灭每组重复的 载盘/指标计算/信号注入。
+
+        与 prepare_matrix_optimization 同一套签名约束 (params 不入签名 — 参数
+        驱动的列差异由并集计划覆盖); load 窗口推导与 run() 完全同口径。
+        """
+        if not configs:
+            raise ValueError("optimizer preparation requires at least one backtest config")
+
+        signature = self._matrix_prepare_signature(configs[0])
+        if any(self._matrix_prepare_signature(config) != signature for config in configs[1:]):
+            raise ValueError("optimizer trials must share strategy, universe, range and overrides")
+
+        first = configs[0]
+        strategy = self.strategy_engine.get(first.strategy_id)
+        if not _panel_backend(strategy.execution_backend):
+            raise ValueError("shared feature panel preparation requires a panel-path strategy")
+        StrategyEngine.validate_context(
+            strategy,
+            StrategyDataContext(
+                asset_type=first.asset_type,
+                timeframe="1d",
+                as_of=first.end,
+            ),
+        )
+
+        overrides = first.overrides or {}
+        basic_filter = _basic_filter_for_asset(
+            self._effective_basic_filter(strategy, overrides), first.asset_type
+        )
+        entry_signals = self._effective_signals(overrides, "entry_signals", strategy.entry_signals)
+        exit_signals = self._effective_signals(overrides, "exit_signals", strategy.exit_signals)
+        resolver = StrategyDependencyResolver()
+        plans: list[ResolvedFeaturePlan] = []
+        for config in configs:
+            params = self._normalize_params(config.params or {}, strategy)
+            plans.append(resolver.resolve(
+                strategy,
+                params=params,
+                basic_filter=basic_filter,
+                entry_signals=entry_signals,
+                exit_signals=exit_signals,
+                overrides=overrides,
+                minute_fill=config.minute_fill,
+                asset_type=config.asset_type,
+            ))
+        feature_plan = _merge_resolved_feature_plans(plans)
+
+        max_hold_days = self._override_value(overrides, "max_hold_days", strategy.max_hold_days)
+        full_horizon_days = max(int(max_hold_days or first.holding_days or 5), 1)
+        warmup_days = max(120, int(max(feature_plan.warmup_bars, 1) * 1.6))
+        load_start = first.start - timedelta(days=warmup_days)
+        load_end = first.end
+        if first.mode == "full":
+            load_end = first.end + timedelta(days=(full_horizon_days + 5) * 2)
+        sim_end = load_end if first.mode == "full" else first.end
+
+        panel = self.engine.load_panel_for_backtest(
+            first.symbols, load_start, load_end, feature_plan, asset_type=first.asset_type,
+        )
+        if panel.is_empty():
+            raise ValueError("共享特征面板无数据，请检查日期范围或先运行盘后管道")
+        return PreparedPanelBacktest(
+            signature=signature,
+            panel=panel,
+            columns=frozenset(panel.columns),
+            load_start=load_start,
+            load_end=load_end,
+            sim_end=sim_end,
+        )
+
     def run(
         self,
         config: StrategyBacktestConfig,
@@ -1114,6 +1211,7 @@ class StrategyBacktestService:
         cancel_event: threading.Event | None = None,
         prepared: PreparedMatrixBacktest | None = None,
         result_policy: BacktestResultPolicy | None = None,
+        prepared_panel: PreparedPanelBacktest | None = None,
     ) -> StrategyBacktestResult:
         t0 = time.perf_counter()
         run_id = uuid.uuid4().hex[:10]
@@ -1347,18 +1445,38 @@ class StrategyBacktestService:
                 return _err("正式回测区间内无数据")
             feature_width = len(feature_plan.matrix_columns)
         else:
-            t_load = time.perf_counter()
-            try:
-                panel = self.engine.load_panel_for_backtest(
-                    config.symbols,
-                    load_start,
-                    load_end,
-                    feature_plan,
-                    asset_type=config.asset_type,
+            if prepared_panel is not None:
+                # P2a 共享面板: 优化器准备的超集面板直接复用, 跳过载盘/指标/信号注入。
+                if not _panel_backend(s.execution_backend):
+                    return _err("共享特征面板只能用于面板路径策略")
+                if prepared_panel.signature != self._matrix_prepare_signature(config):
+                    return _err("共享特征面板与当前回测配置不匹配")
+                required_columns = (
+                    set(feature_plan.base_columns) | set(feature_plan.instrument_columns)
+                    | set(feature_plan.signal_columns) | set(feature_plan.indicator_columns)
+                    | set(feature_plan.fundamental_columns)
                 )
-            except (ValueError, pl.exceptions.PolarsError) as e:
-                return _err(f"回测特征准备失败: {e}")
-            timing_ms["load_panel"] = round((time.perf_counter() - t_load) * 1000, 1)
+                missing_columns = required_columns - prepared_panel.columns
+                if missing_columns:
+                    return _err(f"共享特征面板缺少本参数组所需列: {sorted(missing_columns)}")
+                panel = prepared_panel.panel
+                load_start = prepared_panel.load_start
+                load_end = prepared_panel.load_end
+                sim_end = prepared_panel.sim_end
+                timing_ms["load_panel"] = 0.0
+            else:
+                t_load = time.perf_counter()
+                try:
+                    panel = self.engine.load_panel_for_backtest(
+                        config.symbols,
+                        load_start,
+                        load_end,
+                        feature_plan,
+                        asset_type=config.asset_type,
+                    )
+                except (ValueError, 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("无数据，请检查日期范围或先运行盘后管道")
             # 环境过滤下正式起点=面板首日时顺延 (首日让渡为预热)

backend/scripts/bench_optimizer_prepare.py

@@ -0,0 +1,122 @@
+"""P2 基准: 默认后端 (polars_expr) 参数网格优化的调参循环耗时。
+
+对同一策略 + 同一参数网格跑 StrategyOptimizer.optimize, 记录准备/试算/总耗时。
+P2a (prepare-once 面板复用) 前后各跑一次同一脚本对比, 验收标准:
+调参循环提速 ≥2× (docs/ext-architecture-plan.md P2)。
+
+用法:
+  cd backend && uv run python scripts/bench_optimizer_prepare.py [--strategy ID]
+  默认策略 custom_v1_kanban_strict (动态 filter_history, 默认执行后端)。
+  --json PATH 另存结果。
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+import time
+from datetime import date
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+logging.getLogger("app").setLevel(logging.WARNING)
+
+from app.backtest.engine import BacktestEngine  # noqa: E402
+from app.backtest.optimizer import StrategyOptimizer  # noqa: E402
+from app.backtest.strategy import StrategyBacktestConfig, StrategyBacktestService  # noqa: E402
+from app.config import settings  # noqa: E402
+from app.strategy.engine import StrategyEngine  # noqa: E402
+from app.tickflow.repository import DataStore, KlineRepository  # noqa: E402
+
+STRATEGY_ID = "consolidation_limit_pullback"
+PARAM_GRID = {
+    "box_days": [10, 15, 20, 25, 30],
+    "box_amplitude": [0.10, 0.20, 0.25],
+}
+START, END = date(2026, 1, 1), date(2026, 9, 30)
+
+
+def _build() -> StrategyOptimizer:
+    from app.strategy import config as strategy_config
+
+    data_dir = Path(settings.data_dir)
+    dirs = [
+        Path(__file__).resolve().parents[1] / "app" / "strategy" / "research",
+        data_dir / "strategies" / "custom",
+        data_dir / "strategies" / "ai",
+        data_dir / "strategies" / "composite",
+    ]
+    engine = StrategyEngine(
+        strategy_dirs=[d for d in dirs if d.exists()],
+        override_loader=lambda sid: strategy_config.load_override(data_dir, sid),
+    )
+    repo = KlineRepository(DataStore(data_dir))
+    service = StrategyBacktestService(BacktestEngine(repo), engine)
+    return StrategyOptimizer(service, engine)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--strategy", default=STRATEGY_ID)
+    parser.add_argument("--json", dest="json_path", default=None)
+    args = parser.parse_args()
+
+    optimizer = _build()
+    combos = [{"box_days": d, "box_amplitude": a} for d in PARAM_GRID["box_days"] for a in PARAM_GRID["box_amplitude"]]
+    cfgs = [
+        StrategyBacktestConfig(
+            strategy_id=args.strategy, symbols=None, start=START, end=END,
+            params=combo, mode="fast",
+        )
+        for combo in combos
+    ]
+
+    # 交错 A/B (同进程, 每组交替先后顺序): 消除页缓存/预热顺序偏置。
+    # plain 臂 = 基线行为 (每组独立全量准备); shared 臂 = P2a (一次 prepare + 复用)。
+    t0 = time.perf_counter()
+    prepared_panel = optimizer.service.prepare_panel_optimization(cfgs)
+    prepare_s = time.perf_counter() - t0
+
+    plain_times: list[float] = []
+    shared_times: list[float] = []
+    plain_stats: list = []
+    shared_stats: list = []
+    for i, cfg in enumerate(cfgs):
+        order = ("plain", "shared") if i % 2 == 0 else ("shared", "plain")
+        for arm in order:
+            t = time.perf_counter()
+            res = optimizer.service.run(cfg, prepared_panel=prepared_panel if arm == "shared" else None)
+            elapsed = time.perf_counter() - t
+            assert res.error is None, f"combo {i} {arm}: {res.error}"
+            (plain_times if arm == "plain" else shared_times).append(elapsed)
+            (plain_stats if arm == "plain" else shared_stats).append(res.stats or {})
+
+    plain_total = sum(plain_times)
+    shared_total = sum(shared_times)
+    row = {
+        "strategy": args.strategy,
+        "combos": len(cfgs),
+        "panel_shape": list(prepared_panel.panel.shape),
+        "prepare_s": round(prepare_s, 2),
+        "plain_trials_s": round(plain_total, 2),
+        "shared_trials_s": round(shared_total, 2),
+        "plain_plus_prepare_s": round(plain_total, 2),
+        "shared_plus_prepare_s": round(prepare_s + shared_total, 2),
+        "trials_speedup": round(plain_total / shared_total, 2) if shared_total else None,
+        "incl_prepare_speedup": round(plain_total / (prepare_s + shared_total), 2) if shared_total else None,
+        "stats_identical": (
+            [{k: v for k, v in st.items() if k != "timing_ms"} for st in plain_stats]
+            == [{k: v for k, v in st.items() if k != "timing_ms"} for st in shared_stats]
+        ),
+        "per_combo_plain_s": [round(t, 2) for t in plain_times],
+        "per_combo_shared_s": [round(t, 2) for t in shared_times],
+    }
+    print(json.dumps(row, ensure_ascii=False, indent=2))
+    if args.json_path:
+        Path(args.json_path).write_text(json.dumps(row, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()

backend/tests/backtest/test_optimizer_run.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import threading
 from dataclasses import dataclass
 from datetime import date
+from types import SimpleNamespace
 
 import pytest
 
@@ -53,6 +54,8 @@ class _FakeService:
         self.calls = []
         self.prepared_calls = []
         self.prepared_values = []
+        self.prepared_panel_values = []
+        self.panel_prepared_calls = []
         self.result_policies = []
         self.matrix_cache_max_bytes = None
         self.cache = _FakeCache()
@@ -66,6 +69,11 @@ class _FakeService:
             "compute_cache": self.cache,
         })()
 
+    def prepare_panel_optimization(self, configs):
+        self.panel = SimpleNamespace(estimated_size=lambda: 2048)
+        self.panel_prepared_calls.append(list(configs))
+        return SimpleNamespace(panel=self.panel)
+
     def run(
         self,
         config,
@@ -73,10 +81,12 @@ class _FakeService:
         cancel_event=None,
         prepared=None,
         result_policy=None,
+        prepared_panel=None,
     ):
         with self._lock:
             self.calls.append(dict(config.params or {}))
             self.prepared_values.append(prepared)
+            self.prepared_panel_values.append(prepared_panel)
             self.result_policies.append(result_policy)
         return self.score_fn(config.params or {})
 
@@ -235,12 +245,31 @@ def test_progress_callback_reports_done_total():
     def cb(msg):
         seen.append(msg)
     _optimizer(score).optimize(_cfg(), progress_cb=cb)
-    assert len(seen) == 4
+    # P2a: 面板路径先发一条 optimizer_prepare (含共享面板字节), 再逐组进度。
+    assert len(seen) == 5
+    assert seen[0]["type"] == "optimizer_prepare"
+    assert seen[0]["shared_panel_bytes"] == 2048
     assert seen[-1]["type"] == "optimizer_finalize"
     assert seen[-1]["done"] == 3
     assert all(m["total"] == 3 for m in seen)
 
 
+def test_panel_optimizer_prepares_once_and_passes_to_trials():
+    def score(p):
+        return _FakeResult(stats={"sortino": p["ma_proximity"]})
+
+    opt = _optimizer(score)  # 默认 polars_expr
+    out = opt.optimize(_cfg(max_workers=8))
+
+    assert len(opt.service.panel_prepared_calls) == 1
+    assert len(opt.service.panel_prepared_calls[0]) == 3  # 全部组合一次性解析
+    assert out["shared_panel"] is True
+    assert out["shared_panel_bytes"] == 2048
+    # 每个试算 + 最优复跑 (3+1) 都复用同一份共享面板
+    assert len(opt.service.prepared_panel_values) == 4
+    assert all(v is not None and v.panel is opt.service.panel for v in opt.service.prepared_panel_values)
+
+
 def test_matrix_optimizer_prepares_once_and_reuses_same_market_data():
     def score(p):
         return _FakeResult(stats={"sortino": p["ma_proximity"]})

backend/tests/backtest/test_prepare_panel.py

@@ -0,0 +1,205 @@
+"""P2a 共享特征面板 (PreparedPanelBacktest) 契约测试。
+
+验收对象 (docs/ext-architecture-plan.md P2):
+  - polars_expr/python_history_legacy 参数网格 prepare-once — 全组依赖计划
+    取并集, 一次加载, 每组复用 (load 只发生一次);
+  - 结果与逐组独立加载完全一致 (stats 剔除计时字段后相等, 交易一致);
+  - fail-closed: 签名不匹配 / 组合所需列缺失 → 显式报错, 不静默回退。
+"""
+from __future__ import annotations
+
+from datetime import date, timedelta
+from types import SimpleNamespace
+
+import polars as pl
+import pytest
+
+from app.backtest.engine import SimResult
+from app.backtest.strategy import (
+    PreparedPanelBacktest,
+    StrategyBacktestConfig,
+    StrategyBacktestService,
+    _panel_backend,
+)
+from app.strategy.engine import StrategyDef
+
+
+def _strategy(**kwargs) -> StrategyDef:
+    defaults = dict(
+        meta={"id": "test", "name": "test", "scoring": {}, "params": [], "limit": 100},
+        basic_filter=None,
+        entry_signals=[],
+        exit_signals=[],
+        stop_loss=None,
+        trailing_stop=None,
+        trailing_take_profit_activate=None,
+        trailing_take_profit_drawdown=None,
+        max_hold_days=None,
+        # 参数驱动依赖: ma_proximity=1 引用 ma5, 否则 ma10 — 并集必须两者都载
+        filter_fn=lambda df, params: pl.col("close") > pl.col(
+            "ma5" if int(params.get("ma_proximity", 1)) == 1 else "ma10"
+        ),
+        filter_history_fn=None,
+        lookback_days=1,
+        source="custom",
+        file_path=None,
+    )
+    defaults.update(kwargs)
+    return StrategyDef(**defaults)
+
+
+class _StrategyEngineStub:
+    def __init__(self, strategy: StrategyDef) -> None:
+        self.strategy = strategy
+
+    def get(self, strategy_id: str) -> StrategyDef:
+        return self.strategy
+
+
+class _RepoStub:
+    def __init__(self, data_dir=None) -> None:
+        self.store = SimpleNamespace(data_dir=data_dir)
+
+    def get_index_daily(self, *args, **kwargs) -> pl.DataFrame:
+        return pl.DataFrame()
+
+
+class _EngineStub:
+    """记录加载次数的最小 BacktestEngine 替身 (面板路径只需 load_panel 家族)。"""
+
+    def __init__(self, panel: pl.DataFrame) -> None:
+        self.panel = panel
+        self.repo = _RepoStub()
+        self.load_count = 0
+
+    def load_panel(self, symbols, start, end, columns=None, asset_type="stock") -> pl.DataFrame:
+        self.load_count += 1
+        return self.panel
+
+    def load_panel_for_backtest(self, symbols, start, end, feature_plan, asset_type="stock"):
+        return self.load_panel(symbols, start, end, columns=sorted(feature_plan.base_columns))
+
+    def simulate_market_matrix(self, matrix, config, progress_cb=None, cancel_event=None, options=None):
+        return SimResult(
+            equity_curve=[{"date": "2024-01-01", "value": config.initial_capital}],
+            drawdown_curve=[{"date": "2024-01-01", "value": 0.0}],
+            trades=[],
+            per_symbol_stats=[],
+            stats={"total_return": 0.0, "n_trades": 0},
+        )
+
+
+def _panel(n_days: int = 40) -> pl.DataFrame:
+    base = date(2024, 1, 1)
+    rows = []
+    for i in range(n_days):
+        for sym in ("A", "B"):
+            close = 10.0 + i * 0.1
+            rows.append({
+                "symbol": sym, "name": sym, "date": base + timedelta(days=i),
+                "open": close - 0.05, "high": close + 0.1, "low": close - 0.1, "close": close,
+                "volume": 100_000.0, "amount": close * 100_000.0,
+                "raw_close": close, "raw_high": close + 0.1, "raw_low": close - 0.1,
+                "ma5": close - 0.2, "ma10": close - 0.5,
+                "signal_limit_up": False, "signal_limit_down": False,
+            })
+    return pl.DataFrame(rows).sort(["symbol", "date"])
+
+
+def _service(panel: pl.DataFrame, strategy: StrategyDef | None = None):
+    engine = _EngineStub(panel)
+    strategy = strategy or _strategy()
+    return StrategyBacktestService(engine, _StrategyEngineStub(strategy)), engine
+
+
+def _cfg(params: dict | None = None, **kw) -> StrategyBacktestConfig:
+    base = dict(
+        strategy_id="test",
+        symbols=["A", "B"],
+        start=date(2024, 1, 20),
+        end=date(2024, 2, 28),
+        params=params or {},
+        mode="fast",
+    )
+    base.update(kw)
+    return StrategyBacktestConfig(**base)
+
+
+@pytest.fixture(autouse=True)
+def _migrated_signal_definitions(monkeypatch):
+    from tests.signal_seeds import install_pipeline_caches
+    install_pipeline_caches(monkeypatch)
+
+
+def test_prepare_loads_once_and_trials_reuse_panel() -> None:
+    service, engine = _service(_panel())
+    cfgs = [_cfg({"ma_proximity": 1}), _cfg({"ma_proximity": 2}), _cfg({"ma_proximity": 1})]
+    prepared = service.prepare_panel_optimization(cfgs)
+
+    assert engine.load_count == 1  # 全组一次加载
+    # 参数驱动列差异由并集计划覆盖: ma5 与 ma10 都在面板里
+    assert {"ma5", "ma10"} <= prepared.columns
+
+    for cfg in cfgs:
+        result = service.run(cfg, prepared_panel=prepared)
+        assert result.error is None, result.error
+    assert engine.load_count == 1  # 试算不再触发加载
+
+
+def test_shared_panel_results_match_independent_runs() -> None:
+    service, _ = _service(_panel())
+    cfgs = [_cfg({"ma_proximity": 1}), _cfg({"ma_proximity": 2})]
+    prepared = service.prepare_panel_optimization(cfgs)
+
+    for cfg in cfgs:
+        plain = service.run(cfg)
+        shared = service.run(cfg, prepared_panel=prepared)
+        assert plain.error is None and shared.error is None
+        # 计时字段天然不同 (load_panel=0), 金融口径必须一致
+        assert {k: v for k, v in (plain.stats or {}).items() if k != "timing_ms"} == \
+               {k: v for k, v in (shared.stats or {}).items() if k != "timing_ms"}
+        assert plain.trades == shared.trades
+
+
+def test_shared_panel_signature_mismatch_rejected() -> None:
+    service, _ = _service(_panel())
+    prepared = service.prepare_panel_optimization([_cfg({"ma_proximity": 1})])
+    drifted = service.run(_cfg({"ma_proximity": 1}, end=date(2024, 3, 15)), prepared_panel=prepared)
+    assert drifted.error is not None and "不匹配" in drifted.error
+
+
+def test_shared_panel_missing_columns_fail_closed() -> None:
+    """共享面板缺本组所需列 → 显式报错 (不静默回退加载)。"""
+    service, _ = _service(_panel())
+    prepared = service.prepare_panel_optimization([_cfg({"ma_proximity": 1})])
+
+    drifted = _strategy(
+        filter_fn=lambda df, params: pl.col("close") > pl.col("ma30"),
+    )
+    drifted_service, _engine = _service(_panel(), strategy=drifted)
+    # 用第一个 service 的面板喂第二个 service (模拟计划漂移)
+    fake_prepared = PreparedPanelBacktest(
+        signature=drifted_service._matrix_prepare_signature(_cfg({"ma_proximity": 9})),
+        panel=prepared.panel,
+        columns=prepared.columns,
+        load_start=prepared.load_start,
+        load_end=prepared.load_end,
+        sim_end=prepared.sim_end,
+    )
+    result = drifted_service.run(_cfg({"ma_proximity": 9}), prepared_panel=fake_prepared)
+    assert result.error is not None and "所需列" in result.error
+
+
+def test_prepare_rejects_matrix_backend() -> None:
+    matrix_strategy = _strategy(execution_backend="matrix_native", filter_fn=None)
+    matrix_service = StrategyBacktestService(_EngineStub(_panel()), _StrategyEngineStub(matrix_strategy))
+    with pytest.raises(ValueError, match="panel-path"):
+        matrix_service.prepare_panel_optimization([_cfg()])
+
+
+def test_panel_backend_predicate() -> None:
+    assert _panel_backend("polars_expr")
+    assert _panel_backend("python_history_legacy")
+    assert not _panel_backend("matrix_native")
+    assert not _panel_backend("composite")
+    assert not _panel_backend("minute_filter")