From 660ce71fcad0189cba610301edde11967b768468 Mon Sep 17 00:00:00 2001 From: shy3130 Date: Mon, 26 Oct 2026 14:21:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(backtest):=20P2a=20=E9=9D=A2=E6=9D=BF?= =?UTF-8?q?=E8=B7=AF=E5=BE=84=20prepare-once=20=E2=80=94=20=E5=8F=82?= =?UTF-8?q?=E6=95=B0=E7=BD=91=E6=A0=BC=E5=85=B1=E4=BA=AB=E7=89=B9=E5=BE=81?= =?UTF-8?q?=E9=9D=A2=E6=9D=BF,=20=E8=B0=83=E5=8F=82=E5=BE=AA=E7=8E=AF=201.?= =?UTF-8?q?6-1.9=C3=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 默认后端 (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 --- backend/app/backtest/optimizer.py | 54 +++-- backend/app/backtest/strategy.py | 146 +++++++++++-- backend/scripts/bench_optimizer_prepare.py | 122 +++++++++++ backend/tests/backtest/test_optimizer_run.py | 31 ++- backend/tests/backtest/test_prepare_panel.py | 205 +++++++++++++++++++ 5 files changed, 527 insertions(+), 31 deletions(-) create mode 100644 backend/scripts/bench_optimizer_prepare.py create mode 100644 backend/tests/backtest/test_prepare_panel.py diff --git a/backend/app/backtest/optimizer.py b/backend/app/backtest/optimizer.py index 8fd875db..0c389c8b 100644 --- a/backend/app/backtest/optimizer.py +++ b/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, diff --git a/backend/app/backtest/strategy.py b/backend/app/backtest/strategy.py index 35566dfe..638c3962 100644 --- a/backend/app/backtest/strategy.py +++ b/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("无数据，请检查日期范围或先运行盘后管道") # 环境过滤下正式起点=面板首日时顺延 (首日让渡为预热) diff --git a/backend/scripts/bench_optimizer_prepare.py b/backend/scripts/bench_optimizer_prepare.py new file mode 100644 index 00000000..5e7cc2ad --- /dev/null +++ b/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() diff --git a/backend/tests/backtest/test_optimizer_run.py b/backend/tests/backtest/test_optimizer_run.py index f9e54b54..e60f69af 100644 --- a/backend/tests/backtest/test_optimizer_run.py +++ b/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"]}) diff --git a/backend/tests/backtest/test_prepare_panel.py b/backend/tests/backtest/test_prepare_panel.py new file mode 100644 index 00000000..693c1661 --- /dev/null +++ b/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")