feat(backtest): P2a-2 共享矩阵编码 — 参数无关半边一次编码, 调参循环 3.14×

build_market_matrix 内部的参数无关成本 (build_market_data_matrix 行情
矩阵 + _encode_axes 轴 + _signal_code_matrix 信号码, 占剩余每组成本
~55%) 拆出复用:

- matrix.py 新增 PanelMatrixEncoding / encode_panel_matrix /
  build_market_matrix_from_encoding; build_market_matrix 改为两者的组合
  (单一实现, 无平行路径)。score 场随参数变化 (universe 置空), 不进共享
  编码 — 逐组散布进信号矩阵, 非有限值→nan 语义与 _float_matrix 一致。
- PreparedPanelBacktest.sim_encoding: prepare 期构建 sim 窗口编码 (失败
  回退逐组建矩阵, 只损失性能); run 侧列序 (忽略 score 列) 与行数比对
  通过才走共享编码, 否则整建 — 回退是优化放弃, 不是正确性闸门。
- 基准 (同进程交错 A/B, 15 组): 试算 4.24×, 含 prepare 3.14× (P2a 单
  面板共享为 1.8×); stats 剔除计时字段后完全一致。对应 P2 验收 ≥2×。
- 回归: tests/backtest 410 全过 (含新增矩阵级逐元素等价测试); ruff
  零新增 (与 HEAD 逐条相同)。

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

backend/app/backtest/matrix.py

@@ -2451,42 +2451,128 @@ def build_market_matrix(
     """Backward-compatible long-panel boundary used by legacy/Polars strategies."""
     if panel.is_empty():
         raise ValueError("cannot build MarketMatrix from an empty panel")
-    market = build_market_data_matrix(
+    encoding = encode_panel_matrix(
         panel,
-        field_columns={"score", "ma5", "ma10", "ma20"},
+        entry_signal_ids=entry_signal_ids,
+        exit_signal_ids=exit_signal_ids,
     )
+    score = (
+        panel.get_column("score")
+        if "score" in panel.columns and panel["score"].dtype.is_numeric()
+        else None
+    )
+    return build_market_matrix_from_encoding(
+        encoding,
+        score=score,
+        entries=entries,
+        exits=exits,
+        entry_delay_bars=entry_delay_bars,
+        exit_delay_bars=exit_delay_bars,
+        minute_exit_trigger=minute_exit_trigger,
+    )
+
+
+@dataclass(frozen=True)
+class PanelMatrixEncoding:
+    """长面板 → 矩阵的参数无关编码, 供优化器跨参数组复用 (P2a-2)。
+
+    行情矩阵不带 score 场 — score 随参数组的 universe 掩码变化 (_apply_score
+    把非候选行置空), 由 build_market_matrix_from_encoding 逐组散布进信号
+    矩阵; 全部数组只读共享, 复用零拷贝。"""
+
+    market: MarketDataMatrix
+    time_id: np.ndarray
+    asset_id: np.ndarray
+    panel_rows: int
+    entry_signal_code: np.ndarray
+    entry_signal_ids: tuple[str, ...]
+    exit_signal_code: np.ndarray
+    exit_signal_ids: tuple[str, ...]
+    # 编码所用的列序; 调用方与当组 sim_columns 比对, 不一致回退整建
+    # (回退是优化放弃, 不是正确性闸门 — 整建结果总是正确)。
+    sim_columns: tuple[str, ...]
+
+    @property
+    def shape(self) -> tuple[int, int]:
+        return self.market.shape
+
+
+def encode_panel_matrix(
+    panel: pl.DataFrame,
+    *,
+    entry_signal_ids: list[str] | None = None,
+    exit_signal_ids: list[str] | None = None,
+) -> PanelMatrixEncoding:
+    """编码面板的参数无关半边: 行情矩阵 + 轴 id + 信号码矩阵。
+
+    score/entry/exit 掩码随参数组变化, 不在编码内 — 见
+    build_market_matrix_from_encoding。"""
+    market = build_market_data_matrix(panel, field_columns={"ma5", "ma10", "ma20"})
     _, _, _, time_id, asset_id = _encode_axes(panel)
     shape = market.shape
-
-    raw_entry = _scatter_bool_series(entries, len(panel), shape, time_id, asset_id)
-    raw_exit = _scatter_bool_series(exits, len(panel), shape, time_id, asset_id)
     entry_codes, normalized_entry_ids = _signal_code_matrix(
-        panel,
-        entry_signal_ids,
-        shape,
-        time_id,
-        asset_id,
+        panel, entry_signal_ids, shape, time_id, asset_id,
     )
     exit_codes, normalized_exit_ids = _signal_code_matrix(
-        panel,
-        exit_signal_ids,
-        shape,
-        time_id,
-        asset_id,
+        panel, exit_signal_ids, shape, time_id, asset_id,
     )
-    score = market.fields.get("score")
+    return PanelMatrixEncoding(
+        market=market,
+        time_id=time_id,
+        asset_id=asset_id,
+        panel_rows=len(panel),
+        entry_signal_code=entry_codes,
+        entry_signal_ids=normalized_entry_ids,
+        exit_signal_code=exit_codes,
+        exit_signal_ids=normalized_exit_ids,
+        sim_columns=tuple(panel.columns),
+    )
+
+
+def build_market_matrix_from_encoding(
+    encoding: PanelMatrixEncoding,
+    *,
+    score: pl.Series | None,
+    entries: pl.Series | None,
+    exits: pl.Series | None,
+    entry_delay_bars: int = 0,
+    exit_delay_bars: int = 0,
+    minute_exit_trigger: bool = False,
+) -> MarketMatrix:
+    """共享编码 + 参数相关的 score/entry/exit → MarketMatrix (优化器试算路径)。
+
+    散布与合成和 build_market_matrix 走同一套件 (_scatter_bool_series /
+    make_signal_matrix / build_market_matrix_from_signals), score 的
+    非有限值→nan 语义与 _float_matrix 一致。"""
+    shape = encoding.shape
+    raw_entry = _scatter_bool_series(
+        entries, encoding.panel_rows, shape, encoding.time_id, encoding.asset_id,
+    )
+    raw_exit = _scatter_bool_series(
+        exits, encoding.panel_rows, shape, encoding.time_id, encoding.asset_id,
+    )
+    score_matrix = None
+    if score is not None and len(score) == encoding.panel_rows:
+        score_matrix = np.full(shape, np.nan, dtype=np.float32)
+        values = np.array(
+            score.cast(pl.Float32, strict=False).to_numpy(),
+            dtype=np.float32,
+            copy=True,
+        )
+        values[~np.isfinite(values)] = np.nan
+        score_matrix[encoding.time_id, encoding.asset_id] = values
     signals = make_signal_matrix(
         shape,
         entry=raw_entry,
         exit=raw_exit,
-        score=np.nan_to_num(score, nan=0.0) if score is not None else None,
-        entry_signal_code=entry_codes,
-        exit_signal_code=exit_codes,
-        entry_signal_ids=normalized_entry_ids,
-        exit_signal_ids=normalized_exit_ids,
+        score=np.nan_to_num(score_matrix, nan=0.0) if score_matrix is not None else None,
+        entry_signal_code=encoding.entry_signal_code,
+        exit_signal_code=encoding.exit_signal_code,
+        entry_signal_ids=encoding.entry_signal_ids,
+        exit_signal_ids=encoding.exit_signal_ids,
     )
     return build_market_matrix_from_signals(
-        market,
+        encoding.market,
         signals,
         entry_delay_bars=entry_delay_bars,
         exit_delay_bars=exit_delay_bars,

backend/app/backtest/strategy.py

@@ -28,11 +28,14 @@ from app.backtest.matrix import (
     MatrixPipelineConfig,
     MatrixPrewarmCancelledError,
     MatrixStrategyPipeline,
+    PanelMatrixEncoding,
     SignalMatrix,
     apply_time_masks,
     build_market_data_matrix,
     build_market_matrix,
+    build_market_matrix_from_encoding,
     build_market_matrix_from_signals,
+    encode_panel_matrix,
     slice_market_data_matrix,
     slice_signal_matrix,
 )
@@ -800,6 +803,8 @@ class PreparedPanelBacktest:
     load_start: date
     load_end: date
     sim_end: date
+    # sim 半边的参数无关矩阵编码 (P2a-2); None = 未启用 (回退逐组建矩阵)
+    sim_encoding: PanelMatrixEncoding | None = None
 
 
 class StrategyBacktestService:
@@ -1195,6 +1200,25 @@ class StrategyBacktestService:
         )
         if panel.is_empty():
             raise ValueError("共享特征面板无数据，请检查日期范围或先运行盘后管道")
+
+        # sim 半边的参数无关编码 (P2a-2): 行情矩阵/轴/信号码一次构建,
+        # 试算只散布 score/entry/exit。sim 窗口与列选择和 run() 完全同口径;
+        # 构建失败不阻断 (回退逐组建矩阵, 只损失性能)。
+        sim_encoding = None
+        try:
+            sim_range = self._date_range_mask(panel, first.start, sim_end)
+            sim_columns = [
+                column for column in feature_plan.matrix_columns if column in panel.columns
+            ]
+            sim_panel = panel.filter(sim_range).select(sorted(sim_columns))
+            sim_encoding = encode_panel_matrix(
+                sim_panel,
+                entry_signal_ids=list(entry_signals),
+                exit_signal_ids=list(exit_signals),
+            )
+        except Exception as e:  # noqa: BLE001 — 编码是纯优化, 失败回退整建
+            logger.warning("共享面板矩阵编码失败, 试算回退逐组建矩阵: %s", e)
+
         return PreparedPanelBacktest(
             signature=signature,
             panel=panel,
@@ -1202,6 +1226,7 @@ class StrategyBacktestService:
             load_start=load_start,
             load_end=load_end,
             sim_end=sim_end,
+            sim_encoding=sim_encoding,
         )
 
     def run(
@@ -1824,16 +1849,48 @@ class StrategyBacktestService:
             }
 
             t_matrix = time.perf_counter()
-            market_matrix = build_market_matrix(
-                sim_panel,
-                sim_entry_mask,
-                sim_exit_mask,
-                entry_delay_bars=1 if matcher_config.entry_fill == "open_t+1" else 0,
-                exit_delay_bars=1 if matcher_config.exit_fill == "open_t+1" else 0,
-                entry_signal_ids=entry_signals,
-                exit_signal_ids=exit_signals,
-                minute_exit_trigger=matcher_config.exit_fill == "signal_next_minute",
+            shared_encoding = (
+                prepared_panel.sim_encoding if prepared_panel is not None else None
             )
+            # score 列只在 run 侧存在 (_apply_score 产物), 编码面板没有 —
+            # 重构后两路径的市场矩阵都不含 score 场, 比对忽略它。
+            _enc_cols = (
+                tuple(c for c in shared_encoding.sim_columns if c != "score")
+                if shared_encoding is not None else None
+            )
+            if (
+                shared_encoding is not None
+                and _enc_cols == tuple(c for c in sorted(sim_columns) if c != "score")
+                and len(sim_entry_mask) == shared_encoding.panel_rows
+            ):
+                # P2a-2: 共享矩阵编码 — 只散布随参数变化的 score/entry/exit。
+                # 列序或行数不一致 (计划漂移) 时回退整建, 回退是优化放弃而非
+                # 正确性闸门, 整建结果总是正确。
+                score_series = (
+                    sim_panel.get_column("score")
+                    if "score" in sim_panel.columns and sim_panel["score"].dtype.is_numeric()
+                    else None
+                )
+                market_matrix = build_market_matrix_from_encoding(
+                    shared_encoding,
+                    score=score_series,
+                    entries=sim_entry_mask,
+                    exits=sim_exit_mask,
+                    entry_delay_bars=1 if matcher_config.entry_fill == "open_t+1" else 0,
+                    exit_delay_bars=1 if matcher_config.exit_fill == "open_t+1" else 0,
+                    minute_exit_trigger=matcher_config.exit_fill == "signal_next_minute",
+                )
+            else:
+                market_matrix = build_market_matrix(
+                    sim_panel,
+                    sim_entry_mask,
+                    sim_exit_mask,
+                    entry_delay_bars=1 if matcher_config.entry_fill == "open_t+1" else 0,
+                    exit_delay_bars=1 if matcher_config.exit_fill == "open_t+1" else 0,
+                    entry_signal_ids=entry_signals,
+                    exit_signal_ids=exit_signals,
+                    minute_exit_trigger=matcher_config.exit_fill == "signal_next_minute",
+                )
             timing_ms["matrix_build"] = round((time.perf_counter() - t_matrix) * 1000, 1)
             del panel, sim_panel, sim_entry_mask, sim_exit_mask
 

backend/tests/backtest/test_prepare_panel.py

@@ -203,3 +203,71 @@ def test_panel_backend_predicate() -> None:
     assert not _panel_backend("matrix_native")
     assert not _panel_backend("composite")
     assert not _panel_backend("minute_filter")
+
+
+# ── P2a-2: 共享矩阵编码 ──────────────────────────────────
+
+def test_matrix_encoding_equivalence() -> None:
+    """编码路径与整建路径逐元素一致 (含 score 的 null→nan→0 语义)。"""
+    import dataclasses
+
+    import numpy as np
+
+    from app.backtest.matrix import (
+        build_market_matrix,
+        build_market_matrix_from_encoding,
+        encode_panel_matrix,
+    )
+
+    rows = []
+    for i in range(30):
+        for order, sym in enumerate(("A", "B", "C")):
+            close = 10.0 + i * 0.1
+            rows.append({
+                "symbol": sym, "name": sym, "date": date(2024, 1, 1) + timedelta(days=i),
+                "open": close - 0.05, "high": close + 0.1, "low": close - 0.1, "close": close,
+                "volume": 1000.0 + i, "ma5": close - 0.1, "ma10": close - 0.2, "ma20": close - 0.3,
+                # null (非候选行) 与负值/越界值都要走到 — nan_to_num(0) 语义一致
+                "score": None if i % 7 == 0 else (i * 0.01 if order != 1 else -1.0),
+                "signal_limit_up": i % 11 == 0, "signal_limit_down": False,
+                "signal_golden": (i + order) % 9 == 0,
+            })
+    panel = pl.DataFrame(rows).sort(["symbol", "date"]).with_columns(
+        pl.col("score").cast(pl.Float64, strict=False)
+    )
+    n = len(panel)
+    entries = pl.Series([(i + j) % 13 == 0 for i in range(30) for j in range(3)][:n])
+    exits = pl.Series([(i * j) % 17 == 0 for i in range(30) for j in range(3)][:n])
+
+    direct = build_market_matrix(
+        panel, entries, exits,
+        entry_delay_bars=1, exit_delay_bars=1,
+        entry_signal_ids=["signal_golden"],
+    )
+    encoding = encode_panel_matrix(panel, entry_signal_ids=["signal_golden"])
+    via_encoding = build_market_matrix_from_encoding(
+        encoding,
+        score=panel.get_column("score"),
+        entries=entries, exits=exits,
+        entry_delay_bars=1, exit_delay_bars=1,
+    )
+
+    for field in dataclasses.fields(direct):
+        a, b = getattr(direct, field.name), getattr(via_encoding, field.name)
+        if isinstance(a, np.ndarray):
+            assert np.array_equal(a, b, equal_nan=True), f"{field.name} 数组不一致"
+        else:
+            assert a == b, f"{field.name} 不一致: {a!r} vs {b!r}"
+
+
+def test_prepare_builds_sim_encoding_and_runs_use_it() -> None:
+    service, engine = _service(_panel())
+    cfgs = [_cfg({"ma_proximity": 1}), _cfg({"ma_proximity": 2})]
+    prepared = service.prepare_panel_optimization(cfgs)
+
+    assert prepared.sim_encoding is not None
+    assert prepared.sim_encoding.panel_rows > 0
+    for cfg in cfgs:
+        result = service.run(cfg, prepared_panel=prepared)
+        assert result.error is None, result.error
+    assert engine.load_count == 1

docs/ext-architecture-plan.md

@@ -50,7 +50,7 @@
 | P0 基准 | `backend/scripts/bench_feature_resolution.py`; 依赖解析宽度/耗时基线 | 基线 JSON 落 `data/benchmarks/`(本地不入库) | ✅ |
 | P1a 依赖推导 | filter_history AST 列推导(`app/strategy/feature_deps.py`), 引擎加载接入 | 无声明且可证明的策略不再全量兜底 | ✅ |
 | P1b 契约统一 | signal/factor 统一 schema + 依赖提取公共化 | 新增 kind 复用同一校验/注册路径 | ✅ |
-| P2 L2 物化 | 按需物化 + 内容寻址缓存 + 失效传播; 指标成为一等定义 | 调参循环提速 ≥2×; 新增 10 个自定义指标热路径耗时不变 | ⬜ |
+| P2 L2 物化 | 按需物化 + 内容寻址缓存 + 失效传播; 指标成为一等定义 | 调参循环提速 ≥2×; 新增 10 个自定义指标热路径耗时不变 | 🔶 P2a 完成 |
 | P3 表达式级 Python | T2 中间层 (ctx 追踪依赖+安全) | SKDJ 类口径零 DSL 可表达 | ⬜ |
 | P4 体验与分发 | 统一选择器/自动表单/导入冒烟; 备份管线复用为扩展包 | 新增信号导入到可回测 ≤2 分钟 | ⬜ |
 | 条件: 沙箱 | T3 子进程执行 | 触发条件: 首次真实导入第三方扩展 | ⬜ |
@@ -67,6 +67,21 @@
 
 数据加载 IO 随列宽近似线性, 宽度比即回测数据准备阶段的近似提速上界。
 
+
+## P2a 实测(2026-10-26, 本机, consolidation_limit_pullback, 15 组参数网格)
+
+共享面板 + 共享矩阵编码 (prepare-once) vs 逐组独立准备, 同进程交错 A/B:
+
+| 臂 | 试算累计 | 含 prepare |
+|---|---|---|
+| 逐组独立 (基线) | 19.47s | 19.47s |
+| P2a 面板共享 | ~9.4s (2.1×) | ~11.0s (1.8×) |
+| P2a-2 面板+矩阵编码共享 | 4.59s (4.24×) | 6.19s (**3.14×**) |
+
+stats 剔除计时字段后逐组完全一致, 交易逐笔相同; tests/backtest 410 全过。
+剩余每组成本 = filter_history (参数相关, 不可共享) + 掩码散布。
+P2b (L2 内容寻址物化) 在此之上进一步覆盖跨 run/跨策略的因子复用。
+
 ## P1b 交付(2026-10-26)
 
 - `app/definitions.py`: `JsonDefinitionStore`(原子写 + 损坏跳过 + 目录 mtime 签名,