fix(backtest): 步进优化支持 python_history_legacy/polars_expr 策略

问题: WalkForwardService._prepare_shared_matrix 对非 matrix_native 策略直接抛
"步进优化暂仅支持矩阵策略", filter_history 类自定义策略 (如 everbloom 系列)
无法做步进优化。

根因: 闸门把非矩阵后端一刀切, 但 run() 内部本就存在 shared_market_data=None
的通用路径 (每折独立优化 + 独立 OOS 回测), 只是永远不会被走到。

方案: 闸门改分流 —— matrix_native 仍走共享矩阵加速路径; python_history_legacy
与 polars_expr 返回 None 走通用每折回测路径 (正确但无矩阵加速); composite /
minute_filter 维持 fail-closed 拒绝, 错误文案同步更新。

兼容: matrix_native 行为不变; 结果字段 shared_market_data=false 标识通用路径。
性能: 非矩阵策略每组合×每折一次完整回测, 大网格耗时线性放大 (UI 已有耗时提示)。

验证: uv run --frozen pytest tests/backtest/test_walkforward.py -q (23 passed,
含新增 3 例: legacy 走通用路径/polars_expr 放行/minute_filter 拒绝);
tests/backtest 全套 282 passed; ruff 无新增告警;
真实 API 端到端: everbloom_tupengpan 2x2 网格 120/30 步进 8/8 折全部完成。
This commit is contained in:
shy3130
2026-09-05 20:33:31 +08:00
parent c1ad4880fe
commit c289ac5767
2 changed files with 62 additions and 3 deletions
+10 -3
View File
@@ -149,14 +149,21 @@ class WalkForwardService:
self.strategy_engine = strategy_engine self.strategy_engine = strategy_engine
def _prepare_shared_matrix(self, cfg: WalkForwardConfig, folds: list[Fold]): def _prepare_shared_matrix(self, cfg: WalkForwardConfig, folds: list[Fold]):
"""Build one immutable superset matrix for every matrix-native fold.""" """Build one immutable superset matrix for every matrix-native fold.
返回 None 时 run() 走通用路径: 每折独立优化 + OOS 回测, 正确但无共享矩阵加速。
python_history_legacy (filter_history) 与 polars_expr (内置) 策略无法装入
共享矩阵, 走通用路径; composite / minute_filter 仍不支持, 保持 fail-closed。
"""
if self.strategy_engine is None or not folds: if self.strategy_engine is None or not folds:
return None return None
strategy = self.strategy_engine.get(cfg.strategy_id) strategy = self.strategy_engine.get(cfg.strategy_id)
if strategy.execution_backend != "matrix_native": if strategy.execution_backend != "matrix_native":
if strategy.execution_backend in ("python_history_legacy", "polars_expr"):
return None
raise ValueError( raise ValueError(
f"步进优化暂仅支持矩阵(matrix_native)策略; " f"步进优化暂仅支持矩阵(matrix_native)/日线历史(python_history_legacy/"
f"{cfg.strategy_id}{strategy.execution_backend}" f"polars_expr)策略; {cfg.strategy_id}{strategy.execution_backend}"
) )
from app.backtest.optimizer import expand_param_grid from app.backtest.optimizer import expand_param_grid
@@ -239,6 +239,58 @@ def test_walkforward_reports_degradation():
assert abs(out["summary"]["degradation"] - 1.0) < 1e-9 assert abs(out["summary"]["degradation"] - 1.0) < 1e-9
# ---------------------------------------------------------------
# 非矩阵后端的闸门: filter_history/内置走通用路径, 其余 fail-closed
# ---------------------------------------------------------------
class _FakeStrategyDef:
def __init__(self, backend):
self.execution_backend = backend
class _FakeStrategyEngine:
def get(self, strategy_id):
return _FakeStrategyDef(self._backend)
def __init__(self, backend):
self._backend = backend
def test_walkforward_supports_filter_history_strategy():
"""python_history_legacy (filter_history) 策略走通用路径:
不建共享矩阵, 每折在训练区间独立优化、测试区间独立 OOS。"""
opt, svc = _FakeOptimizer(), _FakeService()
wf = WalkForwardService(opt, svc, strategy_engine=_FakeStrategyEngine("python_history_legacy"))
out = wf.run(_wf_cfg())
assert out["n_folds"] > 0
assert out["shared_market_data"] is False
assert len(opt.train_ranges) == out["n_folds"]
assert len(svc.calls) == out["n_folds"]
first_fold = out["folds"][0]
assert svc.calls[0]["params"] == first_fold["best_params"]
assert svc.calls[0]["start"] >= opt.train_ranges[0][1]
def test_walkforward_supports_polars_expr_strategy():
"""内置 polars_expr 策略同样走通用路径 (与矩阵策略共用同一套折编排)。"""
opt, svc = _FakeOptimizer(), _FakeService()
wf = WalkForwardService(opt, svc, strategy_engine=_FakeStrategyEngine("polars_expr"))
out = wf.run(_wf_cfg())
assert out["n_folds"] > 0
assert out["shared_market_data"] is False
def test_walkforward_rejects_minute_filter_strategy():
"""minute_filter / composite 无法用日线折编排评估: 保持 fail-closed。"""
wf = WalkForwardService(
_FakeOptimizer(), _FakeService(),
strategy_engine=_FakeStrategyEngine("minute_filter"),
)
with pytest.raises(ValueError, match=r"步进优化暂仅支持|minute_filter"):
wf.run(_wf_cfg())
class _NoParamsOptimizer(_FakeOptimizer): class _NoParamsOptimizer(_FakeOptimizer):
"""模拟训练区间全组失败: best_params=None。""" """模拟训练区间全组失败: best_params=None。"""
def optimize(self, cfg, progress_cb=None, cancel_event=None): def optimize(self, cfg, progress_cb=None, cancel_event=None):