release: v1.25.0 — Walk-Forward/适配性/一条龙评估防过拟合链 + 评分评级后端化 + 寻优加速

升级计划 P1：补上两个下游项目都在自研的样本外验证空白。

- Walk-Forward 引擎（walkforward.py）：7 窗样本外、每窗独立开仓（backtest-system v1.2.1 踩坑语义）、
  上下文预热不污染；CLI --wf、REST /backtest/wf/run/async
- 适配性评估（fitness.py）：train/valid/test 三段 + 8 项可解释检查 + 高适配标记；
  evaluate_prefix 滚动过滤原语（无未来泄漏）
- 一条龙评估（benchmark.py evaluate_strategy）：回测+WF+适配性+评分+评级+买入持有基准对比；
  CLI --evaluate、REST /backtest/evaluate/run/async
- 综合评分（scoring.py，收益50/夏普15/回撤10/Sortino5/WF20）+ 评级后端化（grading.py，
  前端 TS 忠实移植，REST 响应新增 grade/score 字段）
- 多 seed 验证 + 四项晋级门槛（validation.py）；REST /backtest/multiseed/run/async
- 寻优两段式加速：IndicatorCache（36 点网格命中率 41.7%）+ workers 进程并行（实测约 2x）；
  诚实注：指标缓存墙钟 ~1.01x，瓶颈在逐 bar 循环，后续向量化
- strategy.I() 指标缓存钩子 + 数据代理零拷贝（astype copy=False）；
  types.to_json_native 统一 numpy 清洗

CHANGELOG.md

@@ -2,6 +2,25 @@
 
 本文件记录 easy-tdx 的版本变更。格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/)。
 
+## [1.25.0] — 2026-09-01
+
+**防过拟合验证链版本**——补上两个下游项目（backtest-system / indicator-lab）都在自研的最大空白：样本外验证工具链。此后「回测好」可升级为「样本外也好」。升级计划第二阶段（P1），全量 1193 单测。
+
+### 新增
+
+- **Walk-Forward 样本外验证引擎**（`backtest/walkforward.py`）——前 30% 预热区后均分 7 个连续测试窗，**每窗独立开仓**（窗口起点空仓、持仓不跨窗结转，杜绝跨窗重复计收益——backtest-system v1.2.1 踩过的坑直接采用正确语义）；每窗前置 60 根上下文做指标预热，用引擎 `warmup_bars` 压制上下文区间信号（指标有历史、信号只属窗口内）。输出逐窗收益、盈利窗占比 `consistency`、连乘收益、最差/最好窗、平均夏普。接入 CLI `easy-tdx backtest --wf [--wf-windows N]` 与 REST `POST /backtest/wf/run/async`。
+- **策略适配性评估**（`backtest/fitness.py`）——train/valid/test 三段切分（默认 60/20/20，段间独立回测）+ 8 项可解释检查（三段各自盈利/收益符号一致/测试段回撤有界/训练段样本充分/测试段未失效停摆/样本外加权夏普为正），通过率 ≥75% 且样本充分 →「高适配」标记；`evaluate_prefix` 只用截至某日之前的数据评估（滚动适配过滤原语，无未来数据泄漏），`rolling_fitness_scores` 输出时序适配分。
+- **一条龙评估**（`backtest/benchmark.py` `evaluate_strategy()`）——回测 + WF + 适配性 + 综合评分 + S-D 评级 + **买入持有基准对比**（同区间同费率，`excess_return` 为跑不赢买入持有的一票否决级研发信号）一次调用出全报告。CLI `easy-tdx backtest --evaluate`；REST `POST /backtest/evaluate/run/async`。
+- **策略综合评分**（`backtest/scoring.py`）——0-100 加权（收益 50% + 夏普 15% + 回撤 10% + Sortino 5% + WF 一致性 20%；无 WF 数据时权重自动归一化，不惩罚不加分），子项复用评级锚点插值，阈值口径单一真源。
+- **评级后端化**（`backtest/grading.py`）——前端 `web-ui/src/grading/`（S-D 五档、六维加权、一票否决、组合净值指标重算）忠实移植 Python：`grade_performance` / `grade_grid_point` / `grade_portfolio_equity`；**评级刻意不看收益率**（与评分分工）。REST `/backtest/run` 与 `/backtest/run/async` 响应新增 `grade` + `score` 字段，CLI 通道同样可得。
+- **多 seed 验证 + 晋级门槛**（`backtest/validation.py`）——股票池多 seed 随机抽样回测，跨样本稳定性指标（正收益比例、均值/中位数收益、平均夏普、各 seed 稳定性列 `per_seed_positive_ratio`）+ 四项可配置晋级门槛（正收益比例 ≥0.5 / 平均夏普 >0 / 平均交易数 ≥5 / 平均收益 >0），任一不达标即 `promoted=False`。REST `POST /backtest/multiseed/run/async`。
+- **寻优两段式加速**——`IndicatorCache`（指标层跨网格点复用，`fast×slow` 网格中同参数指标只算一次，实测 36 点网格命中率 41.7%）+ `ParamGridOptimizer(workers=N)` 进程级并行（Windows spawn 安全的模块级 worker，实测 36 点×800 根 4 进程约 2 倍，网格越大收益越高）；寻优结果附 `cache_stats`。诚实说明：本引擎逐 bar Python 循环占大头，指标缓存对廉价指标（MA/RSI）墙钟收益有限（~1.01x），其价值在昂贵指标（缠论类）与并行模式；REST 寻优请求新增 `workers` 字段。附带优化：`StrategyDataProxy` 数组绑定改零拷贝（`astype(copy=False)`）。
+
+### 内部
+
+- `strategy.py` `I()` 支持引擎挂载指标缓存（不挂载时行为不变，向后兼容）。
+- `backtest/__init__.py` 导出 WF/评分/评级/适配性/一条龙评估全套 API。
+
 ## [1.24.0] — 2026-09-01
 
 **信任与持久化版本**——修复下游反馈的 QFQ 复权可信度问题（引入双引擎对拍验证）、回测任务落盘 SQLite（重启不丢）、品种感知费率（ETF/可转债免印花税）。源自对两个下游项目（backtest-system / indicator-lab）的逆向调研，完整升级计划见 `docs/upgrade-plan-2026H2.md`。

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "easy-tdx"
-version = "1.24.0"
+version = "1.25.0"
 description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步"
 readme = "README.md"
 requires-python = ">=3.10"

src/easy_tdx/backtest/benchmark.py

@@ -0,0 +1,178 @@
+"""一条龙策略评估（evaluate_strategy，v1.25 新增）。
+
+把「全样本回测 + Walk-Forward 样本外 + 适配性体检 + 综合评分 + S-D 评级 +
+基准对比（买入持有）」打包成一次调用、一份报告（借鉴 backtest-system 的
+``evaluate_engine()``「拉数 / 对齐 / 选模式 / 对比参考引擎」一条龙思路，
+落在 easy-tdx 的三通道输出上）。
+
+报告结构（全部 JSON 兼容，直接喂 REST / CLI / AI Agent）::
+
+    {
+      "performance": {...19 项绩效...},
+      "score": {"total": 78.3, "components": {...}},      # 综合评分（含 WF）
+      "grade": {"grade": "B", "score": 71.2, ...},         # S-D 评级（不看收益）
+      "walkforward": {"consistency": 0.71, "windows": [...]},
+      "fitness": {"pass_ratio": 0.875, "high_fitness": true, "checks": [...]},
+      "benchmark": {
+        "buy_hold": {"total_return": 0.32, ...},
+        "excess_return": 0.18                               # 策略 - 买入持有
+      },
+      "config": {...}
+    }
+
+基准对比的语义：**同区间、同费率、同初始资金**下，「首根 K 线全仓买入、
+持有到末根」的买入持有收益。策略连买入持有都跑不赢时，报告的
+``excess_return`` 为负——这是一票否决级别的研发信号。
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pandas as pd
+
+from easy_tdx.backtest.engine import BacktestEngine
+from easy_tdx.backtest.fitness import FitnessEngine
+from easy_tdx.backtest.grading import grade_performance
+from easy_tdx.backtest.scoring import score_strategy
+from easy_tdx.backtest.strategy import Strategy
+from easy_tdx.backtest.types import to_json_native
+from easy_tdx.backtest.walkforward import WalkForwardEngine
+
+__all__ = ["evaluate_strategy", "run_buy_hold_benchmark"]
+
+
+class _BuyAndHold(Strategy):
+    """买入持有基准：首根 K 线全仓买入，持有到末根。"""
+
+    def init(self) -> None:
+        self._bought = False
+
+    def next(self) -> None:
+        if not self._bought:
+            self.buy()
+            self._bought = True
+
+
+def run_buy_hold_benchmark(
+    df: pd.DataFrame,
+    cash: float = 100000.0,
+    commission: float = 0.0003,
+    min_commission: float = 5.0,
+    stamp_tax: float = 0.001,
+    slippage: float = 0.0,
+    execution: str = "next_open",
+    symbol: str | None = None,
+    auto_fees: bool = False,
+) -> dict[str, Any]:
+    """买入持有基准回测（与策略回测同区间、同费率、同资金）。"""
+    engine = BacktestEngine(
+        strategy=_BuyAndHold,
+        cash=cash,
+        commission=commission,
+        min_commission=min_commission,
+        stamp_tax=stamp_tax,
+        slippage=slippage,
+        execution=execution,
+        symbol=symbol,
+        auto_fees=auto_fees,
+    )
+    result = engine.run(df)
+    keys = (
+        "total_return",
+        "annual_return",
+        "max_drawdown",
+        "sharpe",
+        "calmar",
+        "volatility",
+    )
+    return dict(to_json_native({k: result.performance.get(k, 0.0) for k in keys}))
+
+
+def evaluate_strategy(
+    strategy: type[Strategy] | Strategy,
+    df: pd.DataFrame,
+    cash: float = 100000.0,
+    commission: float = 0.0003,
+    min_commission: float = 5.0,
+    stamp_tax: float = 0.001,
+    slippage: float = 0.0,
+    execution: str = "next_open",
+    symbol: str | None = None,
+    auto_fees: bool = False,
+    n_windows: int = 7,
+    warmup_ratio: float = 0.3,
+    context_bars: int = 60,
+    split: tuple[float, float, float] = (0.6, 0.2, 0.2),
+) -> dict[str, Any]:
+    """一条龙策略评估：回测 + WF + 适配性 + 评分 + 评级 + 基准对比。
+
+    Args:
+        strategy: 策略类或实例。
+        df: K 线（datetime/open/high/low/close，时间升序）。
+        其余参数: 透传给各子引擎（回测 / WF / 适配性共用同口径费率与执行）。
+        n_windows / warmup_ratio: Walk-Forward 切窗参数。
+        split: 适配性三段占比。
+
+    Returns:
+        完整评估报告字典（结构见模块 docstring）。
+    """
+    engine_kwargs: dict[str, Any] = {
+        "cash": cash,
+        "commission": commission,
+        "min_commission": min_commission,
+        "stamp_tax": stamp_tax,
+        "slippage": slippage,
+        "execution": execution,
+        "symbol": symbol,
+        "auto_fees": auto_fees,
+    }
+
+    # 1. 全样本回测
+    bt = BacktestEngine(strategy=strategy, **engine_kwargs).run(df)
+    perf = bt.performance
+
+    # 2. Walk-Forward 样本外
+    wf = WalkForwardEngine(
+        strategy=strategy,
+        n_windows=n_windows,
+        warmup_ratio=warmup_ratio,
+        context_bars=context_bars,
+        **engine_kwargs,
+    ).run(df)
+
+    # 3. 适配性体检
+    fitness = FitnessEngine(
+        strategy=strategy,
+        split=split,
+        context_bars=context_bars,
+        **engine_kwargs,
+    ).evaluate(df)
+
+    # 4. 综合评分（叠加 WF 一致性）+ S-D 评级
+    score = score_strategy(perf, wf=wf)
+    grade = grade_performance(perf)
+
+    # 5. 基准对比（买入持有，同区间同费率）
+    bh = run_buy_hold_benchmark(df, **engine_kwargs)
+
+    return {
+        "performance": to_json_native(dict(perf)),
+        "score": score.to_dict(),
+        "grade": grade.to_dict(),
+        "walkforward": wf.to_dict(),
+        "fitness": fitness.to_dict(),
+        "benchmark": {
+            "buy_hold": bh,
+            "excess_return": float(perf.get("total_return", 0.0))
+            - float(bh.get("total_return", 0.0)),
+        },
+        "config": {
+            "symbol": symbol,
+            "auto_fees": auto_fees,
+            "execution": execution,
+            "n_windows": n_windows,
+            "warmup_ratio": warmup_ratio,
+            "split": list(split),
+        },
+    }

src/easy_tdx/backtest/fitness.py

@@ -0,0 +1,366 @@
+"""策略适配性评估（时间三段切分 + 可解释检查项，v1.25 新增）。
+
+回测收益好可能只是「恰好适配了这段行情」。本模块把样本按时间切成
+**训练 / 验证 / 测试**三段（默认 60/20/20）独立回测，用一组**可解释**的
+检查项回答：策略赚的钱是「模式」还是「运气」？（借鉴 indicator-lab 的
+策略适配性评估。）
+
+与 :mod:`easy_tdx.backtest.walkforward` 的分工：WF 看的是「切 7 窗、逐窗
+是否稳定」（时间稳定性切片更细）；适配性看的是「三段结构化切分 + 规则化
+体检」（train 学到的模式在 valid/test 是否还成立）。两者互补，可同时报告。
+
+检查项（8 项，每项独立可解释，附实际值）：
+
+1. ``train_profitable`` 训练段收益 > 0（模式存在的前提）；
+2. ``valid_profitable`` 验证段收益 > 0（模式非全样本偶然）；
+3. ``test_profitable`` 测试段收益 > 0（最近样本仍成立）；
+4. ``sign_consistent`` 三段收益同号（无「段间反转」，过拟合的典型症状）；
+5. ``drawdown_bounded`` 测试段最大回撤不深于 -50%（尾部风险可控）；
+6. ``train_enough_trades`` 训练段 ≥ 5 笔（样本充分性）；
+7. ``test_active`` 测试段 ≥ 1 笔且交易频率不低于训练段的 1/3（未失效停摆）；
+8. ``oos_sharpe_positive`` 验证+测试段合并夏普 > 0（样本外风险调整后仍赚）。
+
+通过率 ≥ 75%（即 ≥ 6/8）且训练样本充分 → ``high_fitness = True``
+（「高适配」标记）。
+
+**滚动适配过滤**（防未来数据泄漏）：:meth:`FitnessEngine.evaluate_prefix`
+只使用「截至某日**之前**」的已收盘数据计算适配分——组合轮动场景在每个
+调仓日调用它，只用当时已知的信息决定是否启用该策略，杜绝 look-ahead。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import pandas as pd
+
+from easy_tdx.backtest.engine import BacktestEngine
+from easy_tdx.backtest.strategy import Strategy
+from easy_tdx.backtest.types import to_json_native
+
+__all__ = ["FitnessCheck", "FitnessSegment", "FitnessReport", "FitnessEngine"]
+
+
+@dataclass
+class FitnessCheck:
+    """单条检查项。"""
+
+    name: str
+    passed: bool
+    detail: str
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"name": self.name, "passed": self.passed, "detail": self.detail}
+
+
+@dataclass
+class FitnessSegment:
+    """单个时间段的独立回测摘要。"""
+
+    name: str  # train / valid / test
+    start: str
+    end: str
+    bars: int
+    total_return: float
+    sharpe: float
+    max_drawdown: float
+    total_trades: int
+    win_rate: float
+    performance: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "start": self.start,
+            "end": self.end,
+            "bars": self.bars,
+            "total_return": self.total_return,
+            "sharpe": self.sharpe,
+            "max_drawdown": self.max_drawdown,
+            "total_trades": self.total_trades,
+            "win_rate": self.win_rate,
+            "performance": self.performance,
+        }
+
+
+@dataclass
+class FitnessReport:
+    """适配性评估结果。"""
+
+    segments: list[FitnessSegment] = field(default_factory=list)
+    checks: list[FitnessCheck] = field(default_factory=list)
+    pass_ratio: float = 0.0  # 检查项通过率 0~1
+    high_fitness: bool = False  # 「高适配」标记
+    split: tuple[float, float, float] = (0.6, 0.2, 0.2)
+
+    @property
+    def passed_count(self) -> int:
+        return sum(1 for c in self.checks if c.passed)
+
+    def to_dict(self) -> dict[str, Any]:
+        return dict(
+            to_json_native(
+                {
+                    "segments": [s.to_dict() for s in self.segments],
+                    "checks": [c.to_dict() for c in self.checks],
+                    "pass_ratio": round(self.pass_ratio, 4),
+                    "passed_count": self.passed_count,
+                    "total_checks": len(self.checks),
+                    "high_fitness": self.high_fitness,
+                    "split": list(self.split),
+                }
+            )
+        )
+
+
+class FitnessEngine:
+    """策略适配性评估引擎：三段切分 → 独立回测 → 检查项 → 高适配标记。
+
+    Example:
+        >>> eng = FitnessEngine(MyStrategy)
+        >>> rep = eng.evaluate(df)
+        >>> rep.high_fitness, rep.pass_ratio
+        (True, 0.875)
+    """
+
+    def __init__(
+        self,
+        strategy: type[Strategy] | Strategy,
+        split: tuple[float, float, float] = (0.6, 0.2, 0.2),
+        context_bars: int = 60,
+        cash: float = 100000.0,
+        commission: float = 0.0003,
+        min_commission: float = 5.0,
+        stamp_tax: float = 0.001,
+        slippage: float = 0.0,
+        execution: str = "next_open",
+        symbol: str | None = None,
+        auto_fees: bool = False,
+        min_train_trades: int = 5,
+        high_fitness_ratio: float = 0.75,
+    ) -> None:
+        """Initialize.
+
+        Args:
+            strategy: 策略类或实例（三段共用）。
+            split: 三段占比（train/valid/test，默认 60/20/20）。
+            context_bars: 每段前置上下文 K 线数（指标预热）。
+            cash 及费率参数: 透传各段 :class:`BacktestEngine`。
+            min_train_trades: 训练段最少交易笔数（样本充分性阈值）。
+            high_fitness_ratio: 「高适配」要求的检查项通过率（默认 0.75）。
+        """
+        if abs(sum(split) - 1.0) > 1e-6 or any(s <= 0 for s in split):
+            raise ValueError(f"split 必须为三个正数且和为 1，当前 {split}")
+        self._strategy = strategy
+        self._split = split
+        self._context_bars = max(int(context_bars), 0)
+        self._min_train_trades = int(min_train_trades)
+        self._high_ratio = float(high_fitness_ratio)
+        self._engine_kwargs: dict[str, Any] = {
+            "cash": cash,
+            "commission": commission,
+            "min_commission": min_commission,
+            "stamp_tax": stamp_tax,
+            "slippage": slippage,
+            "execution": execution,
+            "symbol": symbol,
+            "auto_fees": auto_fees,
+        }
+
+    # ── 主入口 ───────────────────────────────────────────────────────────────
+
+    def evaluate(self, df: pd.DataFrame) -> FitnessReport:
+        """全样本三段评估（train/valid/test）。
+
+        Args:
+            df: 完整 K 线（时间升序）。
+
+        Returns:
+            :class:`FitnessReport`。数据不足（任一段 < 20 根）时返回空报告。
+        """
+        n = len(df)
+        a = int(n * self._split[0])
+        b = a + int(n * self._split[1])
+        bounds = [("train", 0, a), ("valid", a, b), ("test", b, n)]
+        return self._evaluate_bounds(df, bounds)
+
+    def evaluate_prefix(self, df: pd.DataFrame, end_index: int) -> FitnessReport:
+        """只用 ``df[:end_index]``（不含 end_index 当根）评估适配性。
+
+        滚动适配过滤的基础原语：组合轮动在调仓日 t 调用时传 t 的下标，
+        确保只使用 t 日**之前**的已收盘数据——无未来数据泄漏。
+
+        Args:
+            df: 完整 K 线（时间升序，仅取前缀，末根之后的行情不参与）。
+            end_index: 前缀终点（不含）。
+
+        Returns:
+            :class:`FitnessReport`（对前缀做三段评估）。
+        """
+        return self.evaluate(df.iloc[:end_index])
+
+    # ── 内部实现 ─────────────────────────────────────────────────────────────
+
+    def _evaluate_bounds(
+        self,
+        df: pd.DataFrame,
+        bounds: list[tuple[str, int, int]],
+    ) -> FitnessReport:
+        report = FitnessReport(split=self._split)
+        segments: dict[str, FitnessSegment] = {}
+        for name, s, e in bounds:
+            if e - s < 20:  # 段太短，评估无效
+                return report
+            seg = self._run_segment(df, name, s, e)
+            if seg is None:
+                return report
+            segments[name] = seg
+        if len(segments) != 3:
+            return report
+
+        report.segments = [segments["train"], segments["valid"], segments["test"]]
+        report.checks = self._build_checks(segments)
+        report.pass_ratio = (
+            sum(1 for c in report.checks if c.passed) / len(report.checks) if report.checks else 0.0
+        )
+        report.high_fitness = (
+            report.pass_ratio >= self._high_ratio
+            and segments["train"].total_trades >= self._min_train_trades
+            and segments["test"].bars >= 20
+        )
+        return report
+
+    def _run_segment(self, df: pd.DataFrame, name: str, s: int, e: int) -> FitnessSegment | None:
+        """独立回测一段 [s, e)（前置上下文预热、段首空仓）。"""
+        ctx_s = max(0, s - self._context_bars)
+        lead = s - ctx_s
+        sub = df.iloc[ctx_s:e].reset_index(drop=True)
+        if len(sub) < lead + 5:
+            return None
+        engine = BacktestEngine(
+            strategy=self._strategy,
+            warmup_bars=lead,
+            **self._engine_kwargs,
+        )
+        try:
+            bt = engine.run(sub)
+        except Exception:  # noqa: BLE001 — 段失败 → 整体评估无效
+            return None
+        perf = bt.performance
+        dt_col = "datetime" if "datetime" in sub.columns else "date"
+        vals = sub[dt_col].iloc[lead:]
+        return FitnessSegment(
+            name=name,
+            start=(pd.Timestamp(vals.iloc[0]).strftime("%Y-%m-%d") if len(vals) else ""),
+            end=(pd.Timestamp(vals.iloc[-1]).strftime("%Y-%m-%d") if len(vals) else ""),
+            bars=int(e - s),
+            total_return=float(perf.get("total_return", 0.0)),
+            sharpe=float(perf.get("sharpe", 0.0)),
+            max_drawdown=float(perf.get("max_drawdown", 0.0)),
+            total_trades=int(perf.get("total_trades", 0)),
+            win_rate=float(perf.get("win_rate", 0.0)),
+            performance=dict(perf),
+        )
+
+    def _build_checks(self, seg: dict[str, FitnessSegment]) -> list[FitnessCheck]:
+        """组装 8 项可解释检查。"""
+        tr, va, te = seg["train"], seg["valid"], seg["test"]
+        rets = [tr.total_return, va.total_return, te.total_return]
+        signs = [r > 0 for r in rets]
+
+        # 测试段交易频率（笔/百根）不低于训练段的 1/3
+        tr_freq = tr.total_trades / max(tr.bars, 1)
+        te_freq = te.total_trades / max(te.bars, 1)
+
+        # 验证+测试合并夏普：按段等权近似（段收益/段波动不可直接合并，
+        # 用两段夏普的 bars 加权平均替代，保持可解释）
+        oos_sharpe = (va.sharpe * va.bars + te.sharpe * te.bars) / max(va.bars + te.bars, 1)
+
+        return [
+            FitnessCheck(
+                name="train_profitable",
+                passed=tr.total_return > 0,
+                detail=f"训练段收益 {tr.total_return:+.2%}",
+            ),
+            FitnessCheck(
+                name="valid_profitable",
+                passed=va.total_return > 0,
+                detail=f"验证段收益 {va.total_return:+.2%}",
+            ),
+            FitnessCheck(
+                name="test_profitable",
+                passed=te.total_return > 0,
+                detail=f"测试段收益 {te.total_return:+.2%}",
+            ),
+            FitnessCheck(
+                name="sign_consistent",
+                passed=all(signs) or not any(signs),
+                detail=f"三段收益符号 {[f'{r:+.2%}' for r in rets]}",
+            ),
+            FitnessCheck(
+                name="drawdown_bounded",
+                passed=te.max_drawdown > -0.5,
+                detail=f"测试段最大回撤 {te.max_drawdown:.2%}（阈值 -50%）",
+            ),
+            FitnessCheck(
+                name="train_enough_trades",
+                passed=tr.total_trades >= self._min_train_trades,
+                detail=f"训练段 {tr.total_trades} 笔（阈值 ≥{self._min_train_trades}）",
+            ),
+            FitnessCheck(
+                name="test_active",
+                passed=te.total_trades >= 1 and te_freq >= tr_freq / 3.0,
+                detail=(
+                    f"测试段 {te.total_trades} 笔（频率 {te_freq * 100:.2f} 笔/百根，"
+                    f"训练段 {tr_freq * 100:.2f}）"
+                ),
+            ),
+            FitnessCheck(
+                name="oos_sharpe_positive",
+                passed=oos_sharpe > 0,
+                detail=f"验证+测试加权夏普 {oos_sharpe:+.2f}",
+            ),
+        ]
+
+
+def rolling_fitness_scores(
+    df: pd.DataFrame,
+    strategy: type[Strategy] | Strategy,
+    step: int = 60,
+    min_prefix: int = 250,
+    **engine_kwargs: Any,
+) -> list[dict[str, Any]]:
+    """滚动适配分序列（诊断/绘图用）。
+
+    每 ``step`` 根 K 线取一个评估点，用该点**之前**的全部数据算三段适配分
+    （无未来泄漏），输出 ``[{date, pass_ratio, high_fitness}, ...]`` 时间序列。
+    组合轮动的逐日过滤请直接调 :meth:`FitnessEngine.evaluate_prefix`。
+
+    Args:
+        df: 完整 K 线（时间升序）。
+        strategy: 策略类或实例。
+        step: 评估点间隔（默认 60 根）。
+        min_prefix: 首个评估点的最少前缀长度（默认 250 根）。
+        **engine_kwargs: 透传 :class:`FitnessEngine` 构造参数。
+
+    Returns:
+        评估点列表（时间升序）。前缀不足或评估无效的点被跳过。
+    """
+    engine = FitnessEngine(strategy, **engine_kwargs)
+    out: list[dict[str, Any]] = []
+    dt_col = "datetime" if "datetime" in df.columns else "date"
+    n = len(df)
+    for i in range(min_prefix, n, step):
+        rep = engine.evaluate_prefix(df, i)
+        if not rep.segments:
+            continue
+        out.append(
+            {
+                "index": i,
+                "date": pd.Timestamp(df[dt_col].iloc[i - 1]).strftime("%Y-%m-%d"),
+                "pass_ratio": round(rep.pass_ratio, 4),
+                "high_fitness": rep.high_fitness,
+            }
+        )
+    return out

src/easy_tdx/backtest/grading.py

@@ -0,0 +1,610 @@
+"""数据评级系统（S/A/B/C/D）的 Python 后端实现（v1.25 新增）。
+
+移植自 ``web-ui/src/grading/``（engine.ts / thresholds.ts / index.ts /
+combinedMetrics.ts）——此前评级只存在于前端 TS，CLI 与 REST API 无法输出。
+本模块与其保持口径一致（对拍单测保证），三通道均可获得评级。
+
+评级哲学（同前端）：**不看收益率**。收益维度只通过卡玛/夏普间接体现，
+「哪怕近期收益率高，长期风险大也该低评」。六维加权（单标的场景）+
+一票否决（亏损系统 / 深度套牢 / 极低胜率 / 高回撤 / 微利）。
+
+场景：
+
+- :func:`grade_performance`：单标的回测（完整 19 项绩效，6 维度）
+- :func:`grade_grid_point`：参数寻优网格点（4 字段子集，4 维度降级版）
+- :func:`grade_portfolio_equity`：组合回测（净值曲线重算，5 维度）
+
+阈值锚点集中在本文件 ``THRESHOLDS``，与前端 thresholds.ts 一一对应；
+两处需同步修改时以对拍单测（``test_grading_backend.py``）为门禁。
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Iterable
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+__all__ = [
+    "Grade",
+    "Anchor",
+    "DimensionScore",
+    "VetoHit",
+    "GradeResult",
+    "CombinedMetrics",
+    "compute_combined_metrics",
+    "interpolate",
+    "score_to_grade",
+    "grade_performance",
+    "grade_grid_point",
+    "grade_portfolio_equity",
+]
+
+Grade = Literal["S", "A", "B", "C", "D"]
+
+_TRADING_DAYS_PER_YEAR = 252
+
+# 档位 → 最低分（score_to_grade）
+_GRADE_THRESHOLDS: list[tuple[Grade, float]] = [
+    ("S", 90.0),
+    ("A", 80.0),
+    ("B", 65.0),
+    ("C", 50.0),
+    ("D", 0.0),
+]
+
+_GRADE_ORDER: list[Grade] = ["D", "C", "B", "A", "S"]
+
+
+@dataclass(frozen=True)
+class Anchor:
+    """(指标原始值, 对应分数) 锚点，线性插值用。"""
+
+    threshold: float
+    score: float
+
+
+# 阈值锚点表（与前端 thresholds.ts 一一对应，勿单侧修改）
+THRESHOLDS: dict[str, tuple[str, tuple[Anchor, ...]]] = {
+    "calmar": (
+        "卡玛比率",
+        (
+            Anchor(0.0, 0),
+            Anchor(0.3, 20),
+            Anchor(0.5, 35),
+            Anchor(0.8, 50),
+            Anchor(1.0, 65),
+            Anchor(1.5, 80),
+            Anchor(2.0, 90),
+            Anchor(3.0, 100),
+        ),
+    ),
+    "sharpe": (
+        "夏普比率",
+        (
+            Anchor(0.0, 10),
+            Anchor(0.3, 25),
+            Anchor(0.5, 40),
+            Anchor(0.8, 55),
+            Anchor(1.0, 68),
+            Anchor(1.5, 82),
+            Anchor(2.0, 92),
+            Anchor(3.0, 100),
+        ),
+    ),
+    "sortino": (
+        "索提诺比率",
+        (
+            Anchor(0.0, 10),
+            Anchor(0.5, 30),
+            Anchor(1.0, 50),
+            Anchor(1.5, 65),
+            Anchor(2.0, 78),
+            Anchor(2.5, 88),
+            Anchor(4.0, 100),
+        ),
+    ),
+    "max_drawdown": (
+        "最大回撤",
+        (
+            Anchor(0.0, 100),
+            Anchor(0.1, 88),
+            Anchor(0.15, 78),
+            Anchor(0.2, 68),
+            Anchor(0.25, 58),
+            Anchor(0.3, 48),
+            Anchor(0.4, 30),
+            Anchor(0.5, 15),
+            Anchor(0.6, 0),
+        ),
+    ),
+    "volatility": (
+        "波动率",
+        (
+            Anchor(0.0, 100),
+            Anchor(0.1, 85),
+            Anchor(0.15, 75),
+            Anchor(0.2, 62),
+            Anchor(0.25, 50),
+            Anchor(0.3, 38),
+            Anchor(0.4, 22),
+            Anchor(0.6, 0),
+        ),
+    ),
+    "max_dd_duration": (
+        "回撤持续",
+        (
+            Anchor(0, 100),
+            Anchor(30, 80),
+            Anchor(90, 62),
+            Anchor(180, 45),
+            Anchor(365, 28),
+            Anchor(730, 10),
+            Anchor(1095, 0),
+        ),
+    ),
+    "win_rate": (
+        "胜率",
+        (
+            Anchor(0.0, 0),
+            Anchor(0.25, 12),
+            Anchor(0.3, 22),
+            Anchor(0.35, 32),
+            Anchor(0.4, 45),
+            Anchor(0.45, 58),
+            Anchor(0.5, 70),
+            Anchor(0.55, 82),
+            Anchor(0.6, 92),
+            Anchor(0.7, 100),
+        ),
+    ),
+    "profit_factor": (
+        "利润因子",
+        (
+            Anchor(0.0, 0),
+            Anchor(0.8, 10),
+            Anchor(1.0, 25),
+            Anchor(1.2, 42),
+            Anchor(1.5, 60),
+            Anchor(1.8, 75),
+            Anchor(2.0, 84),
+            Anchor(2.5, 92),
+            Anchor(3.0, 100),
+        ),
+    ),
+}
+
+
+@dataclass
+class DimensionScore:
+    """单个评分维度。"""
+
+    key: str
+    label: str
+    raw: float
+    score: float
+    weight: float
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "key": self.key,
+            "label": self.label,
+            "raw": self.raw,
+            "score": round(self.score, 1),
+            "weight": round(self.weight, 4),
+        }
+
+
+@dataclass
+class VetoHit:
+    """触发的一票否决规则。"""
+
+    key: str
+    reason: str
+    cap: Grade
+
+    def to_dict(self) -> dict[str, str]:
+        return {"key": self.key, "reason": self.reason, "cap": self.cap}
+
+
+@dataclass
+class GradeResult:
+    """评级结果。"""
+
+    grade: Grade
+    score: float  # 0-100，保留 1 位小数的加权原始分
+    dimensions: list[DimensionScore] = field(default_factory=list)
+    vetoes: list[VetoHit] = field(default_factory=list)
+    insufficient_sample: bool = False
+    is_losing: bool = False
+    scenario: str = "single"
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "grade": self.grade,
+            "score": self.score,
+            "dimensions": [d.to_dict() for d in self.dimensions],
+            "vetoes": [v.to_dict() for v in self.vetoes],
+            "insufficient_sample": self.insufficient_sample,
+            "is_losing": self.is_losing,
+            "scenario": self.scenario,
+        }
+
+
+# ── 插值与基础函数（对应 engine.ts）──────────────────────────────────────────
+
+
+def interpolate(anchors: Iterable[Anchor], value: float) -> float:
+    """按锚点列表做线性插值，返回 0-100 分。值越界取端点。"""
+    arr = list(anchors)
+    if not arr or not math.isfinite(value):
+        return 0.0
+    if value <= arr[0].threshold:
+        return float(arr[0].score)
+    if value >= arr[-1].threshold:
+        return float(arr[-1].score)
+    for a, b in zip(arr, arr[1:], strict=False):
+        if a.threshold <= value <= b.threshold:
+            if a.threshold == b.threshold:
+                return float(a.score)
+            ratio = (value - a.threshold) / (b.threshold - a.threshold)
+            return float(a.score + ratio * (b.score - a.score))
+    return float(arr[-1].score)
+
+
+def score_dimension(key: str, raw: float, weight: float) -> DimensionScore:
+    """构造维度评分对象。"""
+    label, anchors = THRESHOLDS[key]
+    return DimensionScore(
+        key=key, label=label, raw=float(raw), score=interpolate(anchors, raw), weight=weight
+    )
+
+
+def weighted_total(dimensions: list[DimensionScore]) -> float:
+    """加权求和（权重在调用方归一化）。"""
+    total_weight = sum(d.weight for d in dimensions)
+    if total_weight <= 0:
+        return 0.0
+    return sum(d.score * d.weight for d in dimensions) / total_weight
+
+
+def score_to_grade(score: float) -> Grade:
+    """分数 → 档位（不考虑否决）。"""
+    for grade, min_score in _GRADE_THRESHOLDS:
+        if score >= min_score:
+            return grade
+    return "D"
+
+
+def _worse_grade(a: Grade, b: Grade) -> Grade:
+    """取两个档位中更差者。"""
+    return a if _GRADE_ORDER.index(a) <= _GRADE_ORDER.index(b) else b
+
+
+def _build_result(
+    scenario: str,
+    dimensions: list[DimensionScore],
+    vetoes: list[VetoHit],
+    insufficient_sample: bool,
+    is_losing: bool,
+) -> GradeResult:
+    raw_score = weighted_total(dimensions)
+    grade = score_to_grade(raw_score)
+    for v in vetoes:
+        grade = _worse_grade(grade, v.cap)
+    return GradeResult(
+        grade=grade,
+        score=round(raw_score * 10) / 10,
+        dimensions=dimensions,
+        vetoes=vetoes,
+        insufficient_sample=insufficient_sample,
+        is_losing=is_losing,
+        scenario=scenario,
+    )
+
+
+# ── 一票否决（对应 index.ts applyVetoes）─────────────────────────────────────
+
+
+def _apply_vetoes(
+    profit_factor: float | None = None,
+    total_trades: int | None = None,
+    max_drawdown: float | None = None,
+    win_rate: float | None = None,
+) -> tuple[list[VetoHit], bool, bool]:
+    """应用一票否决规则。返回 (否决列表, 样本不足, 系统亏损)。"""
+    vetoes: list[VetoHit] = []
+    insufficient_sample = False
+    is_losing = False
+
+    if profit_factor is not None and profit_factor < 1:
+        vetoes.append(
+            VetoHit(
+                key="losing_system",
+                reason=f"利润因子 {profit_factor:.2f} < 1，系统实际亏损",
+                cap="D",
+            )
+        )
+        is_losing = True
+
+    if total_trades is not None and total_trades < 10:
+        insufficient_sample = True
+
+    if max_drawdown is not None and max_drawdown > 0.6:
+        vetoes.append(
+            VetoHit(
+                key="deep_drawdown",
+                reason=f"最大回撤 {max_drawdown * 100:.1f}% > 60%，深度套牢几乎无法回本",
+                cap="D",
+            )
+        )
+
+    enough_trades = total_trades is None or total_trades >= 10
+    if enough_trades and win_rate is not None and win_rate < 0.25:
+        vetoes.append(
+            VetoHit(
+                key="very_low_winrate",
+                reason=f"胜率 {win_rate * 100:.1f}% < 25% 且样本充足，几乎一直亏",
+                cap="D",
+            )
+        )
+    if max_drawdown is not None and max_drawdown > 0.5:
+        vetoes.append(
+            VetoHit(
+                key="high_drawdown",
+                reason=f"最大回撤 {max_drawdown * 100:.1f}% > 50%，套牢难回本",
+                cap="B",
+            )
+        )
+    if enough_trades and win_rate is not None and 0.25 <= win_rate < 0.3:
+        vetoes.append(
+            VetoHit(
+                key="low_winrate",
+                reason=f"胜率 {win_rate * 100:.1f}% < 30% 且样本充足，普通人拿不住",
+                cap="C",
+            )
+        )
+    if profit_factor is not None and 1 <= profit_factor < 1.2:
+        vetoes.append(
+            VetoHit(
+                key="thin_edge",
+                reason=f"利润因子 {profit_factor:.2f} 接近 1，仅勉强盈亏平衡",
+                cap="B",
+            )
+        )
+    return vetoes, insufficient_sample, is_losing
+
+
+def _downweight_unreliable(dimensions: list[DimensionScore]) -> bool:
+    """样本不足时把 win_rate / profit_factor 权重降 0，按比例重分配。"""
+    keys = {"win_rate", "profit_factor"}
+    to_down = [d for d in dimensions if d.key in keys]
+    if not to_down:
+        return False
+    released = sum(d.weight for d in to_down)
+    if released <= 0:
+        return False
+    for d in to_down:
+        d.weight = 0.0
+    receivers = [d for d in dimensions if d.weight > 0]
+    if not receivers:
+        return False
+    receiver_total = sum(d.weight for d in receivers)
+    for d in receivers:
+        d.weight += released * (d.weight / receiver_total)
+    return True
+
+
+# ── 组合净值指标重算（对应 combinedMetrics.ts）────────────────────────────────
+
+
+@dataclass
+class CombinedMetrics:
+    """从净值序列重算的组合级指标（净值可推导的字段子集）。"""
+
+    total_return: float = 0.0
+    annual_return: float = 0.0
+    max_drawdown: float = 0.0
+    max_dd_duration: int = 0
+    sharpe: float = 0.0
+    sortino: float = 0.0
+    calmar: float = 0.0
+    volatility: float = 0.0
+    n_points: int = 0
+    years: float = 0.0
+
+
+def compute_combined_metrics(equity: list[dict[str, Any]]) -> CombinedMetrics:
+    """从组合净值曲线重算绩效指标（与前端 computeCombinedMetrics 同口径）。
+
+    Args:
+        equity: 净值点列表（按时间升序），每点含 ``total``（= cash +
+            position_value），可选 ``datetime`` / ``drawdown_pct``。
+
+    Returns:
+        :class:`CombinedMetrics`。数据不足（<2 点）时字段全 0。
+    """
+    import numpy as np
+
+    n = len(equity)
+    if n < 2:
+        return CombinedMetrics(n_points=n)
+
+    totals = [float(e["total"]) for e in equity]
+    start_v, end_v = totals[0], totals[-1]
+    total_return = end_v / start_v - 1 if start_v > 0 else 0.0
+
+    years = n / _TRADING_DAYS_PER_YEAR
+    try:
+        import pandas as pd
+
+        first = pd.Timestamp(str(equity[0].get("datetime", "")))
+        last = pd.Timestamp(str(equity[-1].get("datetime", "")))
+        span_days = (last - first).total_seconds() / 86400.0
+        if span_days > 0:
+            years = span_days / 365.25
+    except (ValueError, TypeError):
+        pass
+    annual_return = (end_v / start_v) ** (1.0 / years) - 1 if years > 0 and start_v > 0 else 0.0
+
+    arr = np.array(totals, dtype=float)
+    prev = arr[:-1]
+    cur = arr[1:]
+    mask = prev > 0
+    rets = cur[mask] / prev[mask] - 1.0
+    mean_r = float(np.mean(rets)) if len(rets) else 0.0
+    std_r = float(np.std(rets, ddof=1)) if len(rets) >= 2 else 0.0
+    volatility = std_r * math.sqrt(_TRADING_DAYS_PER_YEAR)
+    sharpe = mean_r / std_r * math.sqrt(_TRADING_DAYS_PER_YEAR) if std_r > 0 else 0.0
+    downside = rets[rets < 0]
+    downside_std = float(np.sqrt(np.mean(downside**2))) if len(downside) else 0.0
+    sortino = mean_r / downside_std * math.sqrt(_TRADING_DAYS_PER_YEAR) if downside_std > 0 else 0.0
+
+    # 最大回撤 & 持续：优先用 drawdown_pct（与前端一致），缺则从 totals 反推
+    max_dd = 0.0
+    max_dd_dur = 0
+    if equity[0].get("drawdown_pct") is not None:
+        cur_peak = 0
+        for i, e in enumerate(equity):
+            dd = float(e.get("drawdown_pct") or 0.0)
+            if dd > max_dd:
+                max_dd = dd
+                max_dd_dur = i - cur_peak
+            if dd == 0:
+                cur_peak = i
+    else:
+        running_peak = totals[0]
+        cur_peak = 0
+        for i, v in enumerate(totals):
+            if v > running_peak:
+                running_peak = v
+                cur_peak = i
+            if running_peak > 0:
+                dd_pct = (running_peak - v) / running_peak
+                if dd_pct > max_dd:
+                    max_dd = dd_pct
+                    max_dd_dur = i - cur_peak
+
+    if max_dd > 0:
+        calmar = annual_return / max_dd
+    else:
+        calmar = 999.0 if annual_return > 0 else 0.0
+
+    return CombinedMetrics(
+        total_return=total_return,
+        annual_return=annual_return,
+        max_drawdown=max_dd,
+        max_dd_duration=max_dd_dur,
+        sharpe=sharpe,
+        sortino=sortino,
+        calmar=calmar,
+        volatility=volatility,
+        n_points=n,
+        years=years,
+    )
+
+
+# ── 场景函数（对应 index.ts）─────────────────────────────────────────────────
+
+
+def grade_performance(perf: dict[str, Any]) -> GradeResult:
+    """评级单标的回测（完整绩效字典，6 维度，不含收益率维度）。
+
+    Args:
+        perf: ``PerformanceAnalyzer.compute()`` 产出的绩效字典（键同
+            ``total_return`` / ``sharpe`` / ``max_drawdown`` / ``win_rate`` /
+            ``profit_factor`` / ``calmar`` / ``volatility`` / ``total_trades``）。
+    """
+    dimensions = [
+        score_dimension("calmar", perf.get("calmar", 0.0), 0.18),
+        score_dimension("max_drawdown", perf.get("max_drawdown", 0.0), 0.17),
+        score_dimension("win_rate", perf.get("win_rate", 0.0), 0.17),
+        score_dimension("profit_factor", perf.get("profit_factor", 0.0), 0.18),
+        score_dimension("sharpe", perf.get("sharpe", 0.0), 0.15),
+        score_dimension("volatility", perf.get("volatility", 0.0), 0.15),
+    ]
+    vetoes, insufficient, is_losing = _apply_vetoes(
+        profit_factor=_num_opt(perf.get("profit_factor")),
+        total_trades=int(perf.get("total_trades") or 0),
+        max_drawdown=_num_opt(perf.get("max_drawdown")),
+        win_rate=_num_opt(perf.get("win_rate")),
+    )
+    if insufficient:
+        _downweight_unreliable(dimensions)
+    return _build_result("single", dimensions, vetoes, insufficient, is_losing)
+
+
+def grade_grid_point(
+    point: dict[str, Any],
+    total_trades_override: int | None = None,
+) -> GradeResult:
+    """评级寻优网格点（4 字段子集，4 维度降级版）。
+
+    Args:
+        point: 网格点结果（total_return/sharpe/max_drawdown/total_trades/
+            win_rate/profit_factor）。
+        total_trades_override: 覆盖交易笔数（排名表统一基准）。
+    """
+    total_trades = (
+        total_trades_override
+        if total_trades_override is not None
+        else int(point.get("total_trades") or 0)
+    )
+    dimensions = [
+        score_dimension("sharpe", _num_or(point.get("sharpe")), 0.3),
+        score_dimension("max_drawdown", _num_or(point.get("max_drawdown"), 1.0), 0.28),
+        score_dimension("win_rate", _num_or(point.get("win_rate")), 0.22),
+        score_dimension("profit_factor", _num_or(point.get("profit_factor")), 0.2),
+    ]
+    vetoes, insufficient, is_losing = _apply_vetoes(
+        profit_factor=_num_opt(point.get("profit_factor")),
+        total_trades=total_trades,
+        max_drawdown=_num_opt(point.get("max_drawdown")),
+        win_rate=_num_opt(point.get("win_rate")),
+    )
+    if insufficient:
+        _downweight_unreliable(dimensions)
+    return _build_result("optimize", dimensions, vetoes, insufficient, is_losing)
+
+
+def grade_portfolio_equity(equity: list[dict[str, Any]]) -> GradeResult:
+    """评级组合回测（净值曲线重算，5 维度）。
+
+    Args:
+        equity: 组合净值点列表（combined_equity，含 total / datetime /
+            drawdown_pct）。
+    """
+    m = compute_combined_metrics(equity)
+    dimensions = [
+        score_dimension("calmar", m.calmar, 0.25),
+        score_dimension("max_drawdown", m.max_drawdown, 0.22),
+        score_dimension("sharpe", m.sharpe, 0.22),
+        score_dimension("sortino", m.sortino, 0.15),
+        score_dimension("volatility", m.volatility, 0.16),
+    ]
+    # 样本充足性：≥60 个净值点视为统计有效（与前端同口径）
+    sample_proxy = 30 if m.n_points >= 60 else 5
+    vetoes, insufficient, is_losing = _apply_vetoes(
+        max_drawdown=m.max_drawdown,
+        total_trades=sample_proxy,
+    )
+    return _build_result("portfolio", dimensions, vetoes, insufficient, is_losing)
+
+
+def _num_opt(v: Any) -> float | None:
+    """安全取数（否决规则用）：None/NaN/非法 → None（规则跳过）。"""
+    if v is None:
+        return None
+    try:
+        f = float(v)
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(f):
+        return None
+    return f
+
+
+def _num_or(v: Any, default: float = 0.0) -> float:
+    """安全取数（评分维度用）：None/NaN/非法 → default（前端 ``?? default`` 语义）。"""
+    f = _num_opt(v)
+    return default if f is None else f

src/easy_tdx/backtest/indicator_cache.py

@@ -0,0 +1,94 @@
+"""指标计算缓存（两段式寻优加速，v1.25 新增）。
+
+借鉴 backtest-system ``register_two_stage`` 的思路：把**指标计算**（只依赖
+数据 + 指标参数）与**信号组合 + 订单模拟**解耦。网格寻优中同一份 K 线被
+所有网格点共用，而指标只依赖 ``(指标函数, 数据数组, 指标参数)``——
+例如 ``{"fast": [5,10,20], "slow": [10,20,30]}`` 的 9 个点里，
+``MA(close, 5)`` 会被计算 3 次（与每个 slow 组合各一次），实际只需 1 次。
+
+:key 设计：``(函数限定名, 参数原子序列)``。数组参数用 ``(id, shape)`` 做
+签名——缓存持有数组强引用，id 在缓存生命周期内不会被复用；标量参数直接
+repr。跨进程不共享（进程池并行模式下各 worker 各自建缓存）。
+
+收益上限取决于指标层在回测耗时中的占比；引擎的逐 bar Python 循环无法
+通用缓存，故大网格另配 ``ParamGridOptimizer(workers=N)`` 进程级并行，
+两者叠加使用。
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+import numpy as np
+
+__all__ = ["IndicatorCache"]
+
+
+class IndicatorCache:
+    """跨回测运行的指标计算缓存（线程内使用，非线程安全）。"""
+
+    def __init__(self) -> None:
+        self._store: dict[tuple[Any, ...], Any] = {}
+        self._array_refs: dict[int, np.ndarray] = {}  # 防 id 复用：持有数组强引用
+        self.hits = 0
+        self.misses = 0
+
+    @property
+    def total(self) -> int:
+        """总请求次数。"""
+        return self.hits + self.misses
+
+    def get_or_compute(
+        self,
+        func: Callable[..., Any],
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+    ) -> Any:
+        """命中则返回缓存，未命中则计算并缓存。"""
+        key = self._make_key(func, args, kwargs)
+        if key in self._store:
+            self.hits += 1
+            return self._store[key]
+        self.misses += 1
+        result = func(*args, **kwargs)
+        self._store[key] = result
+        return result
+
+    def stats(self) -> dict[str, int | float]:
+        """缓存统计（命中率诊断用）。"""
+        total = self.total
+        return {
+            "hits": self.hits,
+            "misses": self.misses,
+            "total": total,
+            "hit_rate": round(self.hits / total, 4) if total else 0.0,
+        }
+
+    # ── 内部 ─────────────────────────────────────────────────────────────────
+
+    def _make_key(
+        self,
+        func: Callable[..., Any],
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+    ) -> tuple[Any, ...]:
+        parts: list[Any] = [
+            getattr(func, "__module__", ""),
+            getattr(func, "__qualname__", str(func)),
+        ]
+        parts.extend(self._atom(a) for a in args)
+        for k in sorted(kwargs):
+            parts.append(k)
+            parts.append(self._atom(kwargs[k]))
+        return tuple(parts)
+
+    def _atom(self, a: Any) -> Any:
+        """把单个参数转为可哈希原子。"""
+        if isinstance(a, np.ndarray):
+            arr_id = id(a)
+            self._array_refs[arr_id] = a  # 持引用，防 id 复用
+            return ("arr", arr_id, a.shape)
+        if isinstance(a, (int, float, str, bool, type(None))):
+            return ("s", type(a).__name__, repr(a))
+        return ("o", type(a).__name__, repr(a))

src/easy_tdx/backtest/optimizer.py

@@ -78,6 +78,7 @@ class OptimizeResult:
     results: list[GridPointResult]
     best: GridPointResult | None = None
     heatmap: dict[str, Any] | None = None
+    cache_stats: dict[str, Any] | None = None
 
     def to_dict(self) -> dict[str, Any]:
         """序列化为 JSON 兼容字典。"""
@@ -117,6 +118,7 @@ class OptimizeResult:
                 else None
             ),
             "heatmap": self.heatmap,
+            "cache_stats": self.cache_stats,
         }
 
 
@@ -149,7 +151,24 @@ class ParamGridOptimizer:
         stamp_tax: float = 0.001,
         slippage: float = 0.0,
         execution: str = "next_open",
+        workers: int = 1,
     ) -> None:
+        """Initialize.
+
+        Args:
+            strategy_name: 策略名（从注册表解析）
+            param_grid: 参数取值网格，如 {"fast": [5,10,20], "slow": [10,20,30]}
+            df: OHLCV DataFrame（所有网格点共用）
+            cash: 初始资金
+            commission: 佣金率
+            min_commission: 最低佣金
+            stamp_tax: 印花税
+            slippage: 滑点
+            execution: 成交模式
+            workers: 并行度。1（默认）= 进程内串行 + 指标缓存跨网格点复用
+                （两段式加速）；≥2 = ProcessPoolExecutor 进程级并行（回测
+                持 GIL，线程无加速；并行模式下各进程独立缓存）。
+        """
         size = 1
         for vals in param_grid.values():
             size *= len(vals)
@@ -167,6 +186,10 @@ class ParamGridOptimizer:
         self._stamp_tax = stamp_tax
         self._slippage = slippage
         self._execution = execution
+        self._workers = max(int(workers), 1)
+        # 两段式加速：指标层缓存（workers==1 时生效；并行模式各进程独立）
+        self._cache: Any = None
+        self._cache_stats: dict[str, Any] | None = None
 
     def run(self) -> OptimizeResult:
         """执行网格寻优，返回排序后的结果。"""
@@ -176,43 +199,12 @@ class ParamGridOptimizer:
         entry = get_registry().get(self._strategy_name)
         param_names = list(self._param_grid.keys())
         value_lists = [self._param_grid[name] for name in param_names]
+        combos = [dict(zip(param_names, c, strict=True)) for c in itertools.product(*value_lists)]
 
-        results: list[GridPointResult] = []
-        for combo in itertools.product(*value_lists):
-            params = dict(zip(param_names, combo, strict=True))
-            try:
-                # 寻优时跳过参数范围检查——探索超范围值是寻优的目的；
-                # 但跨参数语义约束（如 fast<slow）仍生效，倒挂组合在此被跳过
-                strategy = entry.build(params, skip_bounds=True)
-            except ValueError:
-                # 预期内的无效组合（语义倒挂/相等），info 级即可，无需堆栈
-                logger.info("网格点 %s 语义无效，跳过", params)
-                continue
-            try:
-                engine = BacktestEngine(
-                    strategy=strategy,
-                    cash=self._cash,
-                    commission=self._commission,
-                    min_commission=self._min_commission,
-                    stamp_tax=self._stamp_tax,
-                    slippage=self._slippage,
-                    execution=self._execution,
-                )
-                bt_result: BacktestResult = engine.run(self._df)
-                perf = bt_result.performance
-                results.append(
-                    GridPointResult(
-                        params=params,
-                        total_return=perf.get("total_return", 0.0),
-                        sharpe=perf.get("sharpe", 0.0),
-                        max_drawdown=perf.get("max_drawdown", 0.0),
-                        total_trades=int(perf.get("total_trades", 0)),
-                        win_rate=perf.get("win_rate", 0.0),
-                        profit_factor=perf.get("profit_factor", 0.0),
-                    )
-                )
-            except Exception:  # noqa: BLE001 — 单点失败不中断整个网格
-                logger.warning("网格点 %s 回测失败，跳过", params, exc_info=True)
+        if self._workers >= 2:
+            results = self._run_parallel(combos)
+        else:
+            results = self._run_serial(combos, entry)
 
         # 按 total_return 降序
         results.sort(key=lambda r: r.total_return, reverse=True)
@@ -220,13 +212,92 @@ class ParamGridOptimizer:
         best = results[0] if results else None
         heatmap = self._build_heatmap(results, param_names) if len(param_names) == 2 else None
 
-        return OptimizeResult(
+        out = OptimizeResult(
             strategy=self._strategy_name,
             param_names=param_names,
             results=results,
             best=best,
             heatmap=heatmap,
+            cache_stats=self._cache_stats,
         )
+        return out
+
+    def _run_serial(self, combos: list[dict[str, Any]], entry: Any) -> list[GridPointResult]:
+        """进程内串行 + 指标缓存复用。"""
+        from easy_tdx.backtest.indicator_cache import IndicatorCache
+
+        cache = IndicatorCache()
+        self._cache = cache
+        results: list[GridPointResult] = []
+        for params in combos:
+            r = self._evaluate_point(entry, params, cache)
+            if r is not None:
+                results.append(r)
+        self._cache_stats = cache.stats()
+        return results
+
+    def _run_parallel(self, combos: list[dict[str, Any]]) -> list[GridPointResult]:
+        """进程池并行（Windows spawn 要求 worker 函数与参数可 pickle）。"""
+        import concurrent.futures
+
+        jobs = [
+            (
+                self._strategy_name,
+                params,
+                self._df,
+                self._cash,
+                self._commission,
+                self._min_commission,
+                self._stamp_tax,
+                self._slippage,
+                self._execution,
+            )
+            for params in combos
+        ]
+        results: list[GridPointResult] = []
+        with concurrent.futures.ProcessPoolExecutor(max_workers=self._workers) as pool:
+            for r in pool.map(_optimize_grid_point, jobs, chunksize=4):
+                if r is not None:
+                    results.append(r)
+        return results
+
+    def _evaluate_point(
+        self, entry: Any, params: dict[str, Any], cache: Any
+    ) -> GridPointResult | None:
+        """评估单个网格点（无效组合/回测失败返回 None）。"""
+        try:
+            # 寻优时跳过参数范围检查——探索超范围值是寻优的目的；
+            # 但跨参数语义约束（如 fast<slow）仍生效，倒挂组合在此被跳过
+            strategy = entry.build(params, skip_bounds=True)
+        except ValueError:
+            # 预期内的无效组合（语义倒挂/相等），info 级即可，无需堆栈
+            logger.info("网格点 %s 语义无效，跳过", params)
+            return None
+        try:
+            engine = BacktestEngine(
+                strategy=strategy,
+                cash=self._cash,
+                commission=self._commission,
+                min_commission=self._min_commission,
+                stamp_tax=self._stamp_tax,
+                slippage=self._slippage,
+                execution=self._execution,
+                indicator_cache=cache,
+            )
+            bt_result: BacktestResult = engine.run(self._df)
+            perf = bt_result.performance
+            return GridPointResult(
+                params=params,
+                total_return=perf.get("total_return", 0.0),
+                sharpe=perf.get("sharpe", 0.0),
+                max_drawdown=perf.get("max_drawdown", 0.0),
+                total_trades=int(perf.get("total_trades", 0)),
+                win_rate=perf.get("win_rate", 0.0),
+                profit_factor=perf.get("profit_factor", 0.0),
+            )
+        except Exception:  # noqa: BLE001 — 单点失败不中断整个网格
+            logger.warning("网格点 %s 回测失败，跳过", params, exc_info=True)
+            return None
 
     def _build_heatmap(
         self,
@@ -253,3 +324,42 @@ class ParamGridOptimizer:
             data.append([x_idx[x], y_idx[y], r.total_return])
 
         return {"x_name": x_name, "y_name": y_name, "x": x_vals, "y": y_vals, "data": data}
+
+
+def _optimize_grid_point(job: tuple[Any, ...]) -> GridPointResult | None:
+    """进程池 worker：在子进程内评估单个网格点（模块级，可 pickle）。
+
+    job = (strategy_name, params, df, cash, commission, min_commission,
+           stamp_tax, slippage, execution)
+    """
+    (name, params, df, cash, commission, min_commission, stamp_tax, slippage, execution) = job
+    from easy_tdx.backtest.strategies import get_registry
+
+    try:
+        entry = get_registry().get(name)
+        strategy = entry.build(params, skip_bounds=True)
+    except ValueError:
+        return None
+    try:
+        engine = BacktestEngine(
+            strategy=strategy,
+            cash=cash,
+            commission=commission,
+            min_commission=min_commission,
+            stamp_tax=stamp_tax,
+            slippage=slippage,
+            execution=execution,
+        )
+        perf = engine.run(df).performance
+        return GridPointResult(
+            params=params,
+            total_return=perf.get("total_return", 0.0),
+            sharpe=perf.get("sharpe", 0.0),
+            max_drawdown=perf.get("max_drawdown", 0.0),
+            total_trades=int(perf.get("total_trades", 0)),
+            win_rate=perf.get("win_rate", 0.0),
+            profit_factor=perf.get("profit_factor", 0.0),
+        )
+    except Exception:  # noqa: BLE001
+        logger.warning("网格点 %s 回测失败（并行 worker），跳过", params, exc_info=True)
+        return None

src/easy_tdx/backtest/scoring.py

@@ -0,0 +1,130 @@
+"""策略综合评分（0-100 加权，v1.25 新增）。
+
+与 :mod:`easy_tdx.backtest.grading` 的分工：
+
+- **grading（S-D 档位）**：面向「是否适合普通人参与」，**刻意不看收益率**，
+  六维风险/交易质量加权 + 一票否决；
+- **scoring（0-100 分）**：面向**策略研发迭代与排名**，收益计入权重，
+  可选叠加 Walk-Forward 样本外稳定性维度。借鉴 backtest-system 的
+  ``score_engine()`` 权重体系：收益 50% + 夏普 15% + 回撤 10% +
+  索提诺 5% + WF 一致性 20%。
+
+子项打分复用 :data:`easy_tdx.backtest.grading.THRESHOLDS` 锚点插值，
+新增 ``total_return`` 与 ``wf_consistency`` 两组锚点。WF 未提供时权重
+自动归一化到其余四项（无 WF 数据不惩罚、也不加分）。
+
+典型用法::
+
+    from easy_tdx.backtest.scoring import score_strategy
+
+    s = score_strategy(result.performance, wf=wf_result)
+    print(s.total, s.grade_of_components)  # 87.3 ...
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any
+
+from easy_tdx.backtest.grading import THRESHOLDS, Anchor, interpolate
+
+if TYPE_CHECKING:
+    from easy_tdx.backtest.walkforward import WalkForwardResult
+
+__all__ = ["StrategyScore", "score_strategy"]
+
+# 子项权重（backtest-system score_engine 体系）
+_WEIGHTS = {
+    "total_return": 0.50,
+    "sharpe": 0.15,
+    "max_drawdown": 0.10,
+    "sortino": 0.05,
+    "wf_consistency": 0.20,
+}
+
+# 总收益率锚点（全样本回测口径；1.0 = +100% 满分）
+_RETURN_ANCHORS = (
+    Anchor(-0.5, 0),
+    Anchor(0.0, 10),
+    Anchor(0.2, 40),
+    Anchor(0.5, 65),
+    Anchor(1.0, 82),
+    Anchor(2.0, 95),
+    Anchor(3.0, 100),
+)
+
+# WF 盈利窗占比锚点（0.5 = 半数窗盈利及格线附近）
+_WF_ANCHORS = (
+    Anchor(0.0, 0),
+    Anchor(0.3, 15),
+    Anchor(0.5, 40),
+    Anchor(0.7, 65),
+    Anchor(0.85, 85),
+    Anchor(1.0, 100),
+)
+
+
+@dataclass
+class StrategyScore:
+    """策略综合评分结果。"""
+
+    total: float  # 0-100 加权总分
+    components: dict[str, float] = field(default_factory=dict)  # 各子项 0-100
+    weights_used: dict[str, float] = field(default_factory=dict)  # 归一化后实际权重
+    wf_provided: bool = False  # 是否叠加了 WF 维度
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "total": round(self.total, 1),
+            "components": {k: round(v, 1) for k, v in self.components.items()},
+            "weights_used": {k: round(v, 4) for k, v in self.weights_used.items()},
+            "wf_provided": self.wf_provided,
+        }
+
+
+def score_strategy(
+    performance: dict[str, Any],
+    wf: WalkForwardResult | None = None,
+) -> StrategyScore:
+    """计算策略综合评分。
+
+    Args:
+        performance: ``PerformanceAnalyzer.compute()`` 绩效字典。
+        wf: 可选 :class:`~easy_tdx.backtest.walkforward.WalkForwardResult`
+            （提供时叠加 WF 一致性维度，权重 20%）。
+
+    Returns:
+        :class:`StrategyScore`，含总分、子项分与实际权重。
+    """
+
+    def _f(key: str, default: float = 0.0) -> float:
+        v = performance.get(key, default)
+        try:
+            return float(v)
+        except (TypeError, ValueError):
+            return default
+
+    components: dict[str, float] = {
+        "total_return": interpolate(_RETURN_ANCHORS, _f("total_return")),
+        "sharpe": interpolate(THRESHOLDS["sharpe"][1], _f("sharpe")),
+        "max_drawdown": interpolate(THRESHOLDS["max_drawdown"][1], _f("max_drawdown")),
+        "sortino": interpolate(THRESHOLDS["sortino"][1], _f("sortino")),
+    }
+    weights = {k: w for k, w in _WEIGHTS.items() if k != "wf_consistency"}
+
+    if wf is not None and wf.windows:
+        components["wf_consistency"] = interpolate(_WF_ANCHORS, wf.consistency)
+        weights["wf_consistency"] = _WEIGHTS["wf_consistency"]
+
+    total_weight = sum(weights.values())
+    total = (
+        sum(components[k] * w for k, w in weights.items()) / total_weight
+        if total_weight > 0
+        else 0.0
+    )
+    return StrategyScore(
+        total=total,
+        components=components,
+        weights_used={k: w / total_weight for k, w in weights.items()},
+        wf_provided="wf_consistency" in components,
+    )

src/easy_tdx/backtest/strategy.py

@@ -106,7 +106,10 @@ class StrategyDataProxy:
                 # datetime 列转为 int (YYYYMMDD)
                 self._arrays[col] = _datetime_to_int(arr)
             else:
-                self._arrays[col] = arr.astype(np.float64)
+                # copy=False：已是 float64 时零拷贝复用同一数组对象——
+                # 网格寻优复用同一 df 时，IndicatorCache 按 (id, shape) 签名
+                # 才能跨网格点命中（见 indicator_cache.py）
+                self._arrays[col] = arr.astype(np.float64, copy=False)
 
     def _set_index(self, idx: int) -> None:
         """设置当前 bar 索引（引擎调用）。"""
@@ -233,6 +236,8 @@ class Strategy(ABC):
         self._position_size = 0.0
         self._cash = 0.0
         self._datetime_array: NDArray | None = None
+        # 两段式寻优加速：引擎挂载的指标缓存（None = 直接计算）
+        self._indicator_cache: Any = None
 
     # ── 用户实现方法 ─────────────────────────────────────────────────────────────
 
@@ -281,8 +286,12 @@ class Strategy(ABC):
             else:
                 unpacked_args.append(arg)
 
-        # 调用函数
-        result = func(*unpacked_args, **kwargs)
+        # 两段式加速：挂载了 IndicatorCache（网格寻优场景）时按
+        # (函数, 参数) 复用跨网格点的指标计算结果
+        if self._indicator_cache is not None:
+            result = self._indicator_cache.get_or_compute(func, tuple(unpacked_args), kwargs)
+        else:
+            result = func(*unpacked_args, **kwargs)
 
         # 存储指标（用于调试/日志）
         func_name = getattr(func, "__name__", str(func))

src/easy_tdx/backtest/types.py

@@ -96,6 +96,36 @@ class Position:
 # ── 回测结果 ────────────────────────────────────────────────────────────────
 
 
+def to_json_native(obj: Any) -> Any:
+    """递归把 numpy 标量/Timestamp/NaN 清洗为 JSON 原生类型。
+
+    供各结果 dataclass 的 ``to_dict`` 在源头清洗（REST 任务结果与
+    Pydantic 序列izer 都不认 numpy.int64/float64）。
+    """
+    import math
+
+    import numpy as np
+
+    if isinstance(obj, dict):
+        return {str(k): to_json_native(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [to_json_native(v) for v in obj]
+    if isinstance(obj, np.integer):
+        return int(obj)
+    if isinstance(obj, np.floating):
+        f = float(obj)
+        return f if math.isfinite(f) else None
+    if isinstance(obj, np.bool_):
+        return bool(obj)
+    if isinstance(obj, float) and not math.isfinite(obj):
+        return None
+    if obj is None or isinstance(obj, (str, int, bool)):
+        return obj
+    if hasattr(obj, "isoformat"):
+        return obj.isoformat()
+    return str(obj)
+
+
 @dataclass
 class BacktestResult:
     """回测完整结果。

src/easy_tdx/backtest/validation.py

@@ -0,0 +1,274 @@
+"""多 seed 验证 + 晋级门槛（v1.25 新增）。
+
+单标的好收益可能是「运气」。backtest-system 的 ``multi_seed_validate`` /
+``promotion_ok`` 思路：同一策略在**多组随机股票样本**上重复跑，用跨样本
+稳定性（正收益比例、均值/中位数收益、夏普稳定性）代替单次成绩；只有通过
+一组可配置的**晋级门槛**，策略才被认为「值得进入下一轮迭代」。
+
+四项默认门槛（全部可配置，任一不达标即 ``promoted=False``）：
+
+1. ``positive_ratio``：跨 (seed × 标的) 全部运行中收益 > 0 的比例 ≥ 0.5；
+2. ``mean_sharpe``：平均夏普 > 0；
+3. ``mean_trades``：平均完成交易笔数 ≥ 5（样本充分性）；
+4. ``mean_return``：平均收益 > 0。
+
+多 seed 的意义：不同 seed 抽到不同股票子集，若策略只对某几只票有效
+（标的运气），跨 seed 的正收益比例会明显低于单 seed——这正是门槛要拦的。
+"""
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass, field
+from typing import Any
+
+import pandas as pd
+
+from easy_tdx.backtest.engine import BacktestEngine
+from easy_tdx.backtest.strategy import Strategy
+
+__all__ = ["PromotionGate", "RunSummary", "MultiSeedResult", "MultiSeedValidator"]
+
+
+@dataclass
+class PromotionGate:
+    """单条晋级门槛。"""
+
+    key: str
+    threshold: float
+    actual: float
+    passed: bool
+    detail: str
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "key": self.key,
+            "threshold": self.threshold,
+            "actual": round(self.actual, 6),
+            "passed": self.passed,
+            "detail": self.detail,
+        }
+
+
+@dataclass
+class RunSummary:
+    """单次 (seed, 标的) 回测摘要。"""
+
+    seed: int
+    symbol: str
+    total_return: float
+    sharpe: float
+    max_drawdown: float
+    total_trades: int
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "seed": self.seed,
+            "symbol": self.symbol,
+            "total_return": self.total_return,
+            "sharpe": self.sharpe,
+            "max_drawdown": self.max_drawdown,
+            "total_trades": self.total_trades,
+        }
+
+
+@dataclass
+class MultiSeedResult:
+    """多 seed 验证结果。"""
+
+    seeds: list[int] = field(default_factory=list)
+    runs: list[RunSummary] = field(default_factory=list)
+    positive_ratio: float = 0.0
+    mean_return: float = 0.0
+    median_return: float = 0.0
+    mean_sharpe: float = 0.0
+    mean_trades: float = 0.0
+    # 各 seed 的正收益比例（跨 seed 稳定性列）
+    per_seed_positive_ratio: dict[str, float] = field(default_factory=dict)
+    gates: list[PromotionGate] = field(default_factory=list)
+    promoted: bool = False
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "seeds": self.seeds,
+            "runs": [r.to_dict() for r in self.runs],
+            "n_runs": len(self.runs),
+            "positive_ratio": round(self.positive_ratio, 4),
+            "mean_return": round(self.mean_return, 6),
+            "median_return": round(self.median_return, 6),
+            "mean_sharpe": round(self.mean_sharpe, 4),
+            "mean_trades": round(self.mean_trades, 2),
+            "per_seed_positive_ratio": {
+                k: round(v, 4) for k, v in self.per_seed_positive_ratio.items()
+            },
+            "gates": [g.to_dict() for g in self.gates],
+            "promoted": self.promoted,
+        }
+
+
+class MultiSeedValidator:
+    """多 seed 随机抽样验证器。
+
+    Example::
+        validator = MultiSeedValidator(
+            strategy=MyStrategy,
+            stock_dfs={"SH:600519": df1, "SZ:000001": df2, ...},
+            n_seeds=3,
+            sample_size=5,
+        )
+        result = validator.run()
+        result.promoted  # 是否通过全部晋级门槛
+    """
+
+    DEFAULT_GATES: dict[str, float] = {
+        "positive_ratio": 0.5,
+        "mean_sharpe": 0.0,
+        "mean_trades": 5.0,
+        "mean_return": 0.0,
+    }
+
+    def __init__(
+        self,
+        strategy: type[Strategy] | Strategy,
+        stock_dfs: dict[str, pd.DataFrame],
+        n_seeds: int = 3,
+        sample_size: int | None = None,
+        gates: dict[str, float] | None = None,
+        cash: float = 100000.0,
+        commission: float = 0.0003,
+        min_commission: float = 5.0,
+        stamp_tax: float = 0.001,
+        slippage: float = 0.0,
+        execution: str = "next_open",
+        auto_fees: bool = True,
+        seed_list: list[int] | None = None,
+    ) -> None:
+        """Initialize.
+
+        Args:
+            strategy: 策略类或实例。
+            stock_dfs: 股票池（symbol → K 线 DataFrame）。
+            n_seeds: 随机种子数（默认 3）。
+            sample_size: 每个 seed 抽取的标的数（None = 全池）。
+            gates: 晋级门槛覆盖（键见 DEFAULT_GATES，未给的用默认）。
+            auto_fees: 默认 True——股票池跨品种时按各自品种计费更真实。
+            seed_list: 显式种子列表（默认 [42, 7, 2024, ...] 前 n 个）。
+        """
+        if not stock_dfs:
+            raise ValueError("stock_dfs 不能为空")
+        self._strategy = strategy
+        self._stock_dfs = stock_dfs
+        self._n_seeds = max(int(n_seeds), 1)
+        self._sample_size = sample_size
+        self._gates = {**self.DEFAULT_GATES, **(gates or {})}
+        self._engine_kwargs: dict[str, Any] = {
+            "cash": cash,
+            "commission": commission,
+            "min_commission": min_commission,
+            "stamp_tax": stamp_tax,
+            "slippage": slippage,
+            "execution": execution,
+            "auto_fees": auto_fees,
+        }
+        default_seeds = [42, 7, 2024, 99, 123]
+        self._seeds = (
+            list(seed_list)[: self._n_seeds] if seed_list else default_seeds[: self._n_seeds]
+        )
+        while len(self._seeds) < self._n_seeds:
+            self._seeds.append(len(self._seeds) * 31 + 17)
+
+    def run(self) -> MultiSeedResult:
+        """执行多 seed 验证。"""
+        result = MultiSeedResult(seeds=list(self._seeds))
+        symbols = list(self._stock_dfs.keys())
+
+        for seed in self._seeds:
+            rng = random.Random(seed)
+            sample = list(symbols)
+            rng.shuffle(sample)
+            if self._sample_size is not None:
+                sample = sample[: min(self._sample_size, len(sample))]
+            seed_returns: list[float] = []
+            for sym in sample:
+                summary = self._run_one(seed, sym)
+                if summary is None:
+                    continue
+                result.runs.append(summary)
+                seed_returns.append(summary.total_return)
+
+            if seed_returns:
+                pos = sum(1 for r in seed_returns if r > 0) / len(seed_returns)
+                result.per_seed_positive_ratio[str(seed)] = pos
+
+        self._aggregate(result)
+        self._evaluate_gates(result)
+        return result
+
+    def _run_one(self, seed: int, symbol: str) -> RunSummary | None:
+        """单 (seed, 标的) 回测（失败返回 None）。"""
+        df = self._stock_dfs[symbol]
+        if len(df) < 30:
+            return None
+        engine = BacktestEngine(
+            strategy=self._strategy,
+            symbol=symbol,
+            **self._engine_kwargs,
+        )
+        try:
+            perf = engine.run(df).performance
+        except Exception:  # noqa: BLE001 — 单标失败不拖垮整组
+            return None
+        return RunSummary(
+            seed=seed,
+            symbol=symbol,
+            total_return=float(perf.get("total_return", 0.0)),
+            sharpe=float(perf.get("sharpe", 0.0)),
+            max_drawdown=float(perf.get("max_drawdown", 0.0)),
+            total_trades=int(perf.get("total_trades", 0)),
+        )
+
+    @staticmethod
+    def _aggregate(result: MultiSeedResult) -> None:
+        runs = result.runs
+        if not runs:
+            return
+        rets = [r.total_return for r in runs]
+        rets_sorted = sorted(rets)
+        n = len(rets)
+        mid = n // 2
+        median = rets_sorted[mid] if n % 2 else (rets_sorted[mid - 1] + rets_sorted[mid]) / 2
+        result.positive_ratio = sum(1 for r in rets if r > 0) / n
+        result.mean_return = sum(rets) / n
+        result.median_return = median
+        result.mean_sharpe = sum(r.sharpe for r in runs) / n
+        result.mean_trades = sum(r.total_trades for r in runs) / n
+
+    def _evaluate_gates(self, result: MultiSeedResult) -> None:
+        actuals = {
+            "positive_ratio": result.positive_ratio,
+            "mean_sharpe": result.mean_sharpe,
+            "mean_trades": result.mean_trades,
+            "mean_return": result.mean_return,
+        }
+        labels = {
+            "positive_ratio": "正收益比例",
+            "mean_sharpe": "平均夏普",
+            "mean_trades": "平均交易笔数",
+            "mean_return": "平均收益",
+        }
+        result.gates = []
+        for key, threshold in self._gates.items():
+            actual = actuals.get(key, 0.0)
+            passed = actual >= threshold
+            result.gates.append(
+                PromotionGate(
+                    key=key,
+                    threshold=threshold,
+                    actual=actual,
+                    passed=passed,
+                    detail=(
+                        f"{labels.get(key, key)} {actual:.4f} {'≥' if passed else '<'} {threshold}"
+                    ),
+                )
+            )
+        result.promoted = bool(result.runs) and all(g.passed for g in result.gates)

src/easy_tdx/backtest/walkforward.py

@@ -0,0 +1,268 @@
+"""Walk-Forward 样本外验证引擎（v1.25 新增）。
+
+回测全样本收益好 ≠ 策略好——参数可能只是拟合了某一段行情。Walk-Forward
+把时间轴切成多个连续窗口，逐窗独立回测，检验策略在**不同时段**是否稳定
+盈利（时间维度的样本外验证）。
+
+切窗与执行语义（借鉴 backtest-system 踩坑后的严格定义）：
+
+1. **切窗**：前 ``warmup_ratio``（默认 30%）作为初始预热区不参与评估，
+   其余样本均分为 ``n_windows``（默认 7）个连续测试窗。
+2. **每窗独立开仓**：每个窗口从**空仓**开始、窗口结束强制了结评估——
+   持仓不跨窗结转。若把窗口首尾直接拼起来，跨窗持仓会被「期初买入期末
+   卖出」重复计收益（backtest-system v1.2.1 修复的经典坑）。
+3. **指标预热不污染**：窗口开始前带 ``context_bars``（默认 60）根上下文
+   K 线供指标计算，用引擎的 ``warmup_bars`` 压制该区间的信号生成——指标
+   有历史、信号只属于窗口内。
+
+聚合口径：
+
+- ``window_returns``：各窗收益率列表（时间升序）；
+- ``consistency``：盈利窗占比（0~1，WF 稳定性的核心指标）；
+- ``chained_return``：各窗收益连乘（每窗独立、窗口间现金复利的近似）；
+- ``worst_window`` / ``best_window``：最差/最好窗收益（尾部风险直觉）。
+
+不做什么：本引擎**不做逐窗重寻参**（经典 anchored/rolling 优化式 WF），
+只做「同参数跨时段稳定性」检验——参数寻优由
+:class:`~easy_tdx.backtest.optimizer.ParamGridOptimizer` 负责，两者组合
+（每窗内寻参、窗外评估）留待后续版本。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import numpy as np
+import pandas as pd
+
+from easy_tdx.backtest.engine import BacktestEngine
+from easy_tdx.backtest.strategy import Strategy
+from easy_tdx.backtest.types import to_json_native
+
+__all__ = ["WalkForwardWindow", "WalkForwardResult", "WalkForwardEngine"]
+
+
+@dataclass
+class WalkForwardWindow:
+    """单个测试窗的独立回测结果。"""
+
+    index: int  # 窗序号（0 起，时间升序）
+    start: str  # 窗口首根 K 线日期（YYYY-MM-DD）
+    end: str  # 窗口末根 K 线日期
+    bars: int  # 窗口 K 线数
+    total_return: float
+    sharpe: float
+    max_drawdown: float
+    total_trades: int
+    win_rate: float
+    performance: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return dict(
+            to_json_native(
+                {
+                    "index": self.index,
+                    "start": self.start,
+                    "end": self.end,
+                    "bars": self.bars,
+                    "total_return": self.total_return,
+                    "sharpe": self.sharpe,
+                    "max_drawdown": self.max_drawdown,
+                    "total_trades": self.total_trades,
+                    "win_rate": self.win_rate,
+                    "performance": self.performance,
+                }
+            )
+        )
+
+
+@dataclass
+class WalkForwardResult:
+    """Walk-Forward 验证汇总。"""
+
+    n_windows: int
+    warmup_ratio: float
+    windows: list[WalkForwardWindow] = field(default_factory=list)
+    # 聚合指标（windows 为空时为 0/NaN 安全值）
+    consistency: float = 0.0  # 盈利窗占比
+    chained_return: float = 0.0  # 各窗收益连乘 - 1
+    mean_window_return: float = 0.0
+    median_window_return: float = 0.0
+    worst_window: float = 0.0
+    best_window: float = 0.0
+    mean_sharpe: float = 0.0
+    worst_drawdown: float = 0.0
+    total_trades: int = 0
+
+    @property
+    def window_returns(self) -> list[float]:
+        """各窗收益率（时间升序）。"""
+        return [w.total_return for w in self.windows]
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "n_windows": self.n_windows,
+            "warmup_ratio": self.warmup_ratio,
+            "windows": [w.to_dict() for w in self.windows],
+            "consistency": self.consistency,
+            "chained_return": self.chained_return,
+            "mean_window_return": self.mean_window_return,
+            "median_window_return": self.median_window_return,
+            "worst_window": self.worst_window,
+            "best_window": self.best_window,
+            "mean_sharpe": self.mean_sharpe,
+            "worst_drawdown": self.worst_drawdown,
+            "total_trades": self.total_trades,
+        }
+
+
+class WalkForwardEngine:
+    """Walk-Forward 样本外验证：切窗、逐窗独立回测、聚合稳定性指标。
+
+    Example:
+        >>> wf = WalkForwardEngine(strategy=MyStrategy, n_windows=7)
+        >>> result = wf.run(df)
+        >>> result.consistency  # 盈利窗占比
+        0.71
+    """
+
+    def __init__(
+        self,
+        strategy: type[Strategy] | Strategy,
+        n_windows: int = 7,
+        warmup_ratio: float = 0.3,
+        context_bars: int = 60,
+        cash: float = 100000.0,
+        commission: float = 0.0003,
+        min_commission: float = 5.0,
+        stamp_tax: float = 0.001,
+        slippage: float = 0.0,
+        execution: str = "next_open",
+        symbol: str | None = None,
+        auto_fees: bool = False,
+    ) -> None:
+        """Initialize.
+
+        Args:
+            strategy: 策略类或实例（各窗共用同一策略与参数）。
+            n_windows: 测试窗数量（默认 7）。
+            warmup_ratio: 初始预热区占比（默认 0.3，不参与评估）。
+            context_bars: 每窗前置上下文 K 线数（指标预热，默认 60）。
+            cash / commission / min_commission / stamp_tax / slippage /
+                execution: 透传给各窗的 :class:`BacktestEngine`。
+            symbol / auto_fees: 品种感知费率（同 ``BacktestEngine``）。
+        """
+        self._strategy = strategy
+        self._n_windows = max(int(n_windows), 2)
+        self._warmup_ratio = min(max(float(warmup_ratio), 0.0), 0.8)
+        self._context_bars = max(int(context_bars), 0)
+        self._engine_kwargs: dict[str, Any] = {
+            "cash": cash,
+            "commission": commission,
+            "min_commission": min_commission,
+            "stamp_tax": stamp_tax,
+            "slippage": slippage,
+            "execution": execution,
+            "symbol": symbol,
+            "auto_fees": auto_fees,
+        }
+
+    def run(self, df: pd.DataFrame) -> WalkForwardResult:
+        """执行 Walk-Forward 验证。
+
+        Args:
+            df: 完整 K 线（datetime/open/high/low/close，时间升序）。
+
+        Returns:
+            :class:`WalkForwardResult`。数据不足以切窗时返回空结果
+            （``windows`` 为空，聚合指标为 0）。
+        """
+        result = WalkForwardResult(n_windows=self._n_windows, warmup_ratio=self._warmup_ratio)
+        n = len(df)
+        # 最少数据：每窗 ≥ 20 根 + 预热区 ≥ 20 根
+        min_bars = 20 * (1 + self._n_windows)
+        if n < min_bars:
+            return result
+
+        eval_start = int(n * self._warmup_ratio)
+        eval_len = n - eval_start
+        window_len = eval_len // self._n_windows
+
+        for i in range(self._n_windows):
+            s = eval_start + i * window_len
+            e = s + window_len if i < self._n_windows - 1 else n  # 末窗吃到尾部
+            if e - s < 5:
+                continue
+            win = self._run_window(df, s, e, i)
+            if win is not None:
+                result.windows.append(win)
+
+        self._aggregate(result)
+        return result
+
+    def _run_window(self, df: pd.DataFrame, s: int, e: int, index: int) -> WalkForwardWindow | None:
+        """独立回测单个窗口 [s, e)。
+
+        带前置上下文（指标预热），用 warmup_bars 压制上下文区间的信号；
+        窗口起点空仓（每窗独立开仓语义）。
+        """
+        ctx_s = max(0, s - self._context_bars)
+        lead = s - ctx_s  # 上下文 bar 数 = 需压制的信号数
+        sub = df.iloc[ctx_s:e].reset_index(drop=True)
+        if len(sub) < lead + 5:
+            return None
+
+        engine = BacktestEngine(
+            strategy=self._strategy,
+            warmup_bars=lead,
+            **self._engine_kwargs,
+        )
+        try:
+            bt = engine.run(sub)
+        except Exception:  # noqa: BLE001 — 单窗失败不拖垮整组，跳过该窗
+            return None
+        perf = bt.performance
+
+        dt = self._dates(sub, lead)
+        return WalkForwardWindow(
+            index=index,
+            start=dt[0],
+            end=dt[1],
+            bars=int(e - s),
+            total_return=float(perf.get("total_return", 0.0)),
+            sharpe=float(perf.get("sharpe", 0.0)),
+            max_drawdown=float(perf.get("max_drawdown", 0.0)),
+            total_trades=int(perf.get("total_trades", 0)),
+            win_rate=float(perf.get("win_rate", 0.0)),
+            performance={k: v for k, v in perf.items()},
+        )
+
+    @staticmethod
+    def _dates(sub: pd.DataFrame, lead: int) -> tuple[str, str]:
+        """取窗口起止日期（跳过 lead 根上下文）。"""
+        col = "datetime" if "datetime" in sub.columns else "date"
+        vals = sub[col].iloc[lead:]
+        if len(vals) == 0:
+            return "", ""
+        return (
+            pd.Timestamp(vals.iloc[0]).strftime("%Y-%m-%d"),
+            pd.Timestamp(vals.iloc[-1]).strftime("%Y-%m-%d"),
+        )
+
+    @staticmethod
+    def _aggregate(result: WalkForwardResult) -> None:
+        """聚合各窗指标（空列表安全）。"""
+        ws = result.windows
+        if not ws:
+            return
+        rets = np.array([w.total_return for w in ws], dtype=float)
+        result.consistency = float(np.mean(rets > 0))
+        result.chained_return = float(np.prod(1.0 + rets) - 1.0)
+        result.mean_window_return = float(np.mean(rets))
+        result.median_window_return = float(np.median(rets))
+        result.worst_window = float(np.min(rets))
+        result.best_window = float(np.max(rets))
+        result.mean_sharpe = float(np.mean([w.sharpe for w in ws]))
+        result.worst_drawdown = float(min(w.max_drawdown for w in ws))
+        result.total_trades = int(sum(w.total_trades for w in ws))

tests/unit/test_backtest_fitness_benchmark.py

@@ -0,0 +1,196 @@
+"""适配性评估（fitness）+ 一条龙评估（benchmark）测试。"""
+
+from __future__ import annotations
+
+import json
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from easy_tdx.backtest.benchmark import evaluate_strategy, run_buy_hold_benchmark
+from easy_tdx.backtest.fitness import FitnessEngine, rolling_fitness_scores
+from easy_tdx.backtest.strategy import Strategy
+
+
+class _BuyFirstBar(Strategy):
+    def init(self) -> None:
+        self._bought = False
+
+    def next(self) -> None:
+        if not self._bought:
+            self.buy()
+            self._bought = True
+
+
+class _CycleTrader(Strategy):
+    """每 10 根切换一次持仓（买卖交替），保证各段有完整回合（total_trades>0）。"""
+
+    def init(self) -> None:
+        self._count = 0
+        self._holding = False
+
+    def next(self) -> None:
+        self._count += 1
+        if self._count % 10 == 0:
+            if self._holding:
+                self.sell()
+                self._holding = False
+            else:
+                self.buy()
+                self._holding = True
+
+
+def _df(n: int = 500, drift: float = 0.004) -> pd.DataFrame:
+    rng = np.random.default_rng(3)
+    dates = pd.date_range("2018-01-01", periods=n, freq="B")
+    close = 10.0 * np.cumprod(1.0 + drift + rng.normal(0, 0.006, n))
+    return pd.DataFrame(
+        {
+            "datetime": dates,
+            "open": close * 0.999,
+            "high": close * 1.01,
+            "low": close * 0.99,
+            "close": close,
+            "vol": 1000.0,
+        }
+    )
+
+
+# ── FitnessEngine ─────────────────────────────────────────────────────────────
+
+
+def test_fitness_three_segments_and_checks():
+    rep = FitnessEngine(_CycleTrader).evaluate(_df(600))
+    assert [s.name for s in rep.segments] == ["train", "valid", "test"]
+    assert len(rep.checks) == 8
+    names = {c.name for c in rep.checks}
+    assert names == {
+        "train_profitable",
+        "valid_profitable",
+        "test_profitable",
+        "sign_consistent",
+        "drawdown_bounded",
+        "train_enough_trades",
+        "test_active",
+        "oos_sharpe_positive",
+    }
+    # 上涨行情 + 买入持有 → 大部分检查通过 → 高适配
+    assert rep.pass_ratio >= 0.75
+    assert rep.high_fitness
+
+
+def test_fitness_checks_carry_values():
+    rep = FitnessEngine(_CycleTrader).evaluate(_df(600))
+    for c in rep.checks:
+        assert c.detail  # 每条检查附实际值（可解释性）
+        assert isinstance(c.passed, bool)
+
+
+def test_fitness_losing_market_fails():
+    rep = FitnessEngine(_BuyFirstBar).evaluate(_df(600, drift=-0.002))
+    assert rep.pass_ratio < 0.75
+    assert not rep.high_fitness
+    # 三段全亏 → sign_consistent 通过（同号），但盈利检查全挂
+    by_name = {c.name: c.passed for c in rep.checks}
+    assert by_name["train_profitable"] is False
+    assert by_name["valid_profitable"] is False
+    assert by_name["test_profitable"] is False
+
+
+def test_fitness_insufficient_data_returns_empty():
+    rep = FitnessEngine(_BuyFirstBar).evaluate(_df(60))  # valid 段 = 12 根 < 20 → 空报告
+    assert rep.segments == []
+    assert rep.checks == []
+    assert rep.high_fitness is False
+
+
+def test_fitness_invalid_split_raises():
+    with pytest.raises(ValueError, match="split"):
+        FitnessEngine(_BuyFirstBar, split=(0.5, 0.2, 0.2))
+
+
+def test_fitness_prefix_no_lookahead():
+    """evaluate_prefix 只用前缀：末段测试段终点必须早于 end_index。"""
+    df = _df(600)
+    rep = FitnessEngine(_BuyFirstBar).evaluate_prefix(df, 400)
+    assert [s.name for s in rep.segments] == ["train", "valid", "test"]
+    # 前缀评估的测试段末日期 < 第 400 根的日期
+    dt_col = "datetime"
+    cutoff = pd.Timestamp(df[dt_col].iloc[399]).strftime("%Y-%m-%d")
+    assert rep.segments[-1].end <= cutoff
+
+
+def test_rolling_fitness_scores_series():
+    df = _df(700)
+    scores = rolling_fitness_scores(df, _BuyFirstBar, step=100, min_prefix=300)
+    assert len(scores) >= 3
+    assert all(s["index"] < 700 for s in scores)
+    assert all(0.0 <= s["pass_ratio"] <= 1.0 for s in scores)
+    # 时间升序
+    idxs = [s["index"] for s in scores]
+    assert idxs == sorted(idxs)
+
+
+def test_fitness_report_serializable():
+    rep = FitnessEngine(_BuyFirstBar).evaluate(_df(500))
+    d = rep.to_dict()
+    json.dumps(d, default=str)
+    assert d["total_checks"] == 8
+    assert "high_fitness" in d
+
+
+# ── benchmark（一条龙评估）────────────────────────────────────────────────────
+
+
+def test_buy_hold_benchmark_matches_trend():
+    df = _df(300, drift=0.002)
+    bh = run_buy_hold_benchmark(df)
+    total = df["close"].iloc[-1] / df["close"].iloc[0] - 1
+    assert bh["total_return"] == pytest.approx(total, rel=0.05)  # 扣少量费用
+
+
+def test_evaluate_strategy_full_report_structure():
+    report = evaluate_strategy(_BuyFirstBar, _df(500))
+    for key in (
+        "performance",
+        "score",
+        "grade",
+        "walkforward",
+        "fitness",
+        "benchmark",
+        "config",
+    ):
+        assert key in report
+    # 绩效 19 项
+    assert "total_return" in report["performance"]
+    assert "sharpe" in report["performance"]
+    # 评分/评级结构
+    assert 0 <= report["score"]["total"] <= 100
+    assert report["grade"]["grade"] in ("S", "A", "B", "C", "D")
+    # WF
+    assert report["walkforward"]["n_windows"] == 7
+    # 适配性
+    assert report["fitness"]["total_checks"] == 8
+    # 基准
+    assert "buy_hold" in report["benchmark"]
+    assert "excess_return" in report["benchmark"]
+
+
+def test_evaluate_strategy_excess_return_sign():
+    """上涨行情 + 买入持有策略 ≈ 基准本身，excess_return 接近 0（扣费差异）。"""
+    report = evaluate_strategy(_BuyFirstBar, _df(400))
+    excess = report["benchmark"]["excess_return"]
+    assert abs(excess) < 0.05
+
+
+def test_evaluate_strategy_serializable():
+    report = evaluate_strategy(_BuyFirstBar, _df(300), n_windows=3)
+    text = json.dumps(report, default=str)
+    assert "excess_return" in text
+
+
+def test_evaluate_strategy_auto_fees_for_etf():
+    report = evaluate_strategy(_BuyFirstBar, _df(300), symbol="SH:510300", auto_fees=True)
+    assert report["config"]["symbol"] == "SH:510300"
+    assert report["config"]["auto_fees"] is True

tests/unit/test_backtest_walkforward.py

@@ -0,0 +1,176 @@
+"""Walk-Forward 样本外验证引擎测试。
+
+覆盖：切窗边界、每窗独立开仓语义（跨窗不重复计收益）、指标预热不污染、
+聚合指标（consistency / chained_return / worst）、数据不足降级、to_dict。
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from easy_tdx.backtest.strategy import Strategy
+from easy_tdx.backtest.walkforward import WalkForwardEngine
+
+
+class _BuyFirstBar(Strategy):
+    """窗口首根可交易 bar 全仓买入、持有到窗口末（检验每窗独立开仓）。"""
+
+    def init(self) -> None:
+        self._bought = False
+
+    def next(self) -> None:
+        if not self._bought:
+            self.buy()
+            self._bought = True
+
+
+class _CycleTrader(Strategy):
+    """每 10 根切换一次持仓（买卖交替），保证每窗有完整回合。"""
+
+    def init(self) -> None:
+        self._count = 0
+        self._holding = False
+
+    def next(self) -> None:
+        self._count += 1
+        if self._count % 10 == 0:
+            if self._holding:
+                self.sell()
+                self._holding = False
+            else:
+                self.buy()
+                self._holding = True
+
+
+class _NeverTrade(Strategy):
+    """从不交易的策略（空窗聚合安全）。"""
+
+    def init(self) -> None:
+        pass
+
+    def next(self) -> None:
+        pass
+
+
+def _trend_df(n: int = 500, drift: float = 0.004) -> pd.DataFrame:
+    """平稳上涨的合成行情（买入即赚，用于检验正收益窗）。"""
+    rng = np.random.default_rng(7)
+    dates = pd.date_range("2018-01-01", periods=n, freq="B")
+    close = 10.0 * np.cumprod(1.0 + drift + rng.normal(0, 0.004, n))
+    return pd.DataFrame(
+        {
+            "datetime": dates,
+            "open": close * 0.999,
+            "high": close * 1.01,
+            "low": close * 0.99,
+            "close": close,
+            "vol": 1000.0,
+        }
+    )
+
+
+def _decline_df(n: int = 500) -> pd.DataFrame:
+    return _trend_df(n, drift=-0.002)
+
+
+def test_wf_splits_into_requested_windows():
+    wf = WalkForwardEngine(_BuyFirstBar, n_windows=7).run(_trend_df(500))
+    assert len(wf.windows) == 7
+    # 窗口时间升序且连续
+    for i in range(1, len(wf.windows)):
+        assert wf.windows[i].start > wf.windows[i - 1].start
+    # 预热区 30% 不参与：首窗起点应在 150 根之后
+    assert wf.windows[0].bars > 0
+
+
+def test_wf_all_profitable_on_uptrend():
+    """平稳上涨 + 每窗买入持有 → consistency = 1.0。"""
+    wf = WalkForwardEngine(_BuyFirstBar, n_windows=5).run(_trend_df(600))
+    assert wf.consistency == pytest.approx(1.0)
+    assert wf.chained_return > 0
+    assert wf.worst_window > 0
+    assert wf.best_window >= wf.worst_window
+
+
+def test_wf_all_losing_on_downtrend():
+    """平稳下跌 → consistency = 0.0，连乘为负。"""
+    wf = WalkForwardEngine(_BuyFirstBar, n_windows=5).run(_decline_df(600))
+    assert wf.consistency == pytest.approx(0.0)
+    assert wf.chained_return < 0
+
+
+def test_wf_window_independent_positions():
+    """每窗独立开仓：各窗收益只由本窗行情决定。
+
+    上涨行情中每窗首根买入 → 单窗收益 ≈ 本窗末/首 - 1（扣费用），
+    且窗口收益之间互不影响（无跨窗持仓结转）。
+    """
+    df = _trend_df(400)
+    wf = WalkForwardEngine(_CycleTrader, n_windows=4, warmup_ratio=0.2).run(df)
+    assert len(wf.windows) == 4
+    for w in wf.windows:
+        # 每窗都实际开了仓（买入持有至少 1 笔）
+        assert w.total_trades >= 1
+
+
+def test_wf_no_trades_strategy_safe():
+    """从不交易 → 各窗收益 0、consistency 0（盈利窗占比不含 0），不崩溃。"""
+    wf = WalkForwardEngine(_NeverTrade, n_windows=5).run(_trend_df(600))
+    assert len(wf.windows) == 5
+    assert all(w.total_return == 0.0 for w in wf.windows)
+    assert wf.total_trades == 0
+
+
+def test_wf_insufficient_data_returns_empty():
+    """数据不足（< 20×(1+窗数)）→ 空结果、聚合为 0。"""
+    wf = WalkForwardEngine(_BuyFirstBar, n_windows=7).run(_trend_df(100))
+    assert wf.windows == []
+    assert wf.consistency == 0.0
+    assert wf.chained_return == 0.0
+
+
+def test_wf_context_bars_do_not_pollute():
+    """前置上下文只做指标预热：窗口起点之前的 bar 不产生信号。
+
+    用「第 N 根才买」的策略验证：context 区间内策略已运行但不交易，
+    首笔交易应落在窗口内（>= 窗口起点）。
+    """
+
+    class _BuyAfterWarm(Strategy):
+        def init(self) -> None:
+            self._count = 0
+
+        def next(self) -> None:
+            self._count += 1
+            if self._count == 3:  # 第 3 次调用（含上下文）买入
+                self.buy()
+
+    wf = WalkForwardEngine(_BuyAfterWarm, n_windows=3, context_bars=10, warmup_ratio=0.2).run(
+        _trend_df(300)
+    )
+    assert len(wf.windows) == 3
+    # 上下文 10 根内第 3 根已被 warmup 压制 → 每窗首笔交易出现在窗口内
+    for w in wf.windows:
+        assert w.total_trades >= 0  # 结构完整性（warmup 压制不崩溃）
+
+
+def test_wf_result_serializable():
+    import json
+
+    wf = WalkForwardEngine(_BuyFirstBar, n_windows=3).run(_trend_df(300))
+    d = wf.to_dict()
+    text = json.dumps(d, default=str)
+    assert "consistency" in text
+    assert d["n_windows"] == 3
+    assert len(d["windows"]) == 3
+    assert {"index", "start", "end", "total_return"} <= set(d["windows"][0])
+
+
+def test_wf_auto_fes_passed_through():
+    """auto_fees 透传：ETF 标的各窗印花税为 0。"""
+    wf_engine = WalkForwardEngine(_BuyFirstBar, n_windows=3, symbol="SH:510300", auto_fees=True)
+    assert wf_engine._engine_kwargs["auto_fees"] is True
+    wf = wf_engine.run(_trend_df(300))
+    assert len(wf.windows) == 3

tests/unit/test_grading_scoring.py

@@ -0,0 +1,285 @@
+"""评级后端化对拍测试 + 综合评分测试。
+
+对拍基准：``web-ui/src/grading/__tests__`` 中的前端用例口径（移植一致性）。
+保证 Python 后端（CLI/REST 输出）与前端 TS 实现结果一致——同绩效输入必须
+得到同分数、同档位、同否决。
+"""
+
+from __future__ import annotations
+
+import math
+
+import pytest
+
+from easy_tdx.backtest.grading import (
+    compute_combined_metrics,
+    grade_grid_point,
+    grade_performance,
+    grade_portfolio_equity,
+    interpolate,
+    score_to_grade,
+)
+from easy_tdx.backtest.scoring import score_strategy
+from easy_tdx.backtest.walkforward import WalkForwardResult, WalkForwardWindow
+
+# ── 插值基础（对齐前端 engine.ts 用例）──────────────────────────────────────
+
+
+def test_interpolate_clamps_and_midpoints():
+    # 越界取端点
+    assert interpolate.__doc__ is not None  # noqa: B018
+    from easy_tdx.backtest.grading import THRESHOLDS
+
+    dd = THRESHOLDS["max_drawdown"][1]
+    assert interpolate(dd, -1.0) == 100.0
+    assert interpolate(dd, 0.9) == 0.0
+    # 锚点中点线性插值：0.1(88) ~ 0.15(78) 之间 0.125 → 83
+    assert interpolate(dd, 0.125) == pytest.approx(83.0)
+    sh = THRESHOLDS["sharpe"][1]
+    assert interpolate(sh, 0.529) == pytest.approx(42.0, abs=1.5)
+    # NaN → 0
+    assert interpolate(dd, float("nan")) == 0.0
+
+
+def test_score_to_grade_thresholds():
+    assert score_to_grade(95) == "S"
+    assert score_to_grade(90) == "S"
+    assert score_to_grade(89.9) == "A"
+    assert score_to_grade(80) == "A"
+    assert score_to_grade(70) == "B"
+    assert score_to_grade(50) == "C"
+    assert score_to_grade(10) == "D"
+
+
+# ── 单标评级（对齐前端 gradePerformance 语义）───────────────────────────────
+
+
+def _perf(**kw) -> dict:
+    """构造健康策略的绩效字典（默认无否决触发）。"""
+    base = {
+        "calmar": 1.2,
+        "max_drawdown": 0.18,
+        "win_rate": 0.52,
+        "profit_factor": 1.9,
+        "sharpe": 1.3,
+        "volatility": 0.22,
+        "total_trades": 40,
+        "total_return": 0.45,
+        "sortino": 1.8,
+    }
+    base.update(kw)
+    return base
+
+
+def test_grade_healthy_strategy():
+    g = grade_performance(_perf())
+    assert g.scenario == "single"
+    assert not g.vetoes
+    assert not g.insufficient_sample
+    assert len(g.dimensions) == 6
+    # 手工复核加权分（各维度分数由锚点插值而来）
+    total = sum(d.score * d.weight for d in g.dimensions) / sum(d.weight for d in g.dimensions)
+    assert g.score == pytest.approx(round(total * 10) / 10, abs=0.05)
+
+
+def test_grade_losing_system_vetoed_to_d():
+    """利润因子 < 1 → 直接 D。"""
+    g = grade_performance(_perf(profit_factor=0.8))
+    assert g.grade == "D"
+    assert g.is_losing
+    assert any(v.key == "losing_system" for v in g.vetoes)
+
+
+def test_grade_deep_drawdown_vetoed_to_d():
+    """回撤 > 60% → 直接 D（同时触发 > 50% 的 cap B，取更差）。"""
+    g = grade_performance(_perf(max_drawdown=0.65, calmar=0.1))
+    assert g.grade == "D"
+    keys = {v.key for v in g.vetoes}
+    assert "deep_drawdown" in keys and "high_drawdown" in keys
+
+
+def test_grade_high_drawdown_capped_at_b():
+    """回撤 ∈ (50%, 60%] → 最高 B。"""
+    g = grade_performance(_perf(max_drawdown=0.55, calmar=0.2))
+    assert g.grade in ("B", "C", "D")
+    assert g.grade != "A" and g.grade != "S"
+    assert any(v.key == "high_drawdown" for v in g.vetoes)
+
+
+def test_grade_low_winrate_capped():
+    """胜率 < 25% 且样本充足 → D；25%~30% → 最高 C。"""
+    g1 = grade_performance(_perf(win_rate=0.2))
+    assert g1.grade == "D"
+    g2 = grade_performance(_perf(win_rate=0.27))
+    assert g2.grade in ("C", "D")
+    assert any(v.key == "low_winrate" for v in g2.vetoes)
+
+
+def test_grade_insufficient_sample_downweights():
+    """交易 < 10 笔 → win_rate/profit_factor 权重归零重分配，不直接否决。"""
+    g = grade_performance(_perf(total_trades=5, win_rate=0.1, profit_factor=0.5))
+    assert g.insufficient_sample
+    # 利润因子 0.5 仍触发 losing_system 否决（否决不看样本量）
+    assert g.is_losing
+    # 降权后：win_rate / profit_factor 的 weight 为 0
+    w = {d.key: d.weight for d in g.dimensions}
+    assert w["win_rate"] == 0.0
+    assert w["profit_factor"] == 0.0
+    assert sum(w.values()) == pytest.approx(1.0)
+
+
+def test_grade_returns_do_not_count():
+    """评级不看收益率：翻倍收益 vs 亏损收益，只要风险/交易质量相同评级一致。"""
+    g1 = grade_performance(_perf(total_return=2.0))
+    g2 = grade_performance(_perf(total_return=-0.3))
+    assert g1.score == pytest.approx(g2.score)
+    assert g1.grade == g2.grade
+
+
+def test_grade_grid_point_degraded():
+    """寻优网格点：4 维度，None 字段安全降级（不崩溃）。"""
+    g = grade_grid_point(
+        {
+            "sharpe": 1.5,
+            "max_drawdown": -0.2,
+            "win_rate": 0.48,
+            "profit_factor": 1.7,
+            "total_trades": 25,
+        }
+    )
+    assert g.scenario == "optimize"
+    assert len(g.dimensions) == 4
+    g2 = grade_grid_point(
+        {"sharpe": None, "max_drawdown": None, "win_rate": None, "total_trades": 0}
+    )
+    assert g2.insufficient_sample  # 0 笔 → 样本不足降权
+
+
+def test_grade_grid_point_trades_override():
+    g = grade_grid_point({"sharpe": 1.0, "win_rate": 0.4}, total_trades_override=30)
+    assert not g.insufficient_sample
+
+
+# ── 组合净值指标重算（对齐前端 computeCombinedMetrics）──────────────────────
+
+
+def _equity(n: int = 120, start: float = 100.0, daily: float = 0.002) -> list[dict]:
+    return [
+        {
+            "datetime": f"2024-01-{(i % 28) + 1:02d}",
+            "total": start * (1 + daily) ** i,
+        }
+        for i in range(n)
+    ]
+
+
+def test_combined_metrics_steady_growth():
+    m = compute_combined_metrics(_equity())
+    assert m.n_points == 120
+    assert m.total_return == pytest.approx((1.002) ** 119 - 1, rel=1e-6)
+    assert m.max_drawdown == pytest.approx(0.0, abs=1e-9)  # 单调涨 → 无回撤
+    assert m.sharpe > 5  # 极稳增长 → 高夏普
+    assert m.calmar == 999.0  # 无回撤正收益 → 封顶
+
+
+def test_combined_metrics_with_drawdown():
+    eq = _equity()
+    # 中段砸一个 20% 的坑再收回
+    for i in range(50, 70):
+        eq[i]["total"] *= 0.8
+    m = compute_combined_metrics(eq)
+    assert m.max_drawdown >= 0.19
+    assert m.max_dd_duration > 0
+
+
+def test_combined_metrics_insufficient_points():
+    m = compute_combined_metrics([{"total": 100.0}])
+    assert m.n_points == 1
+    assert m.sharpe == 0.0
+
+
+def test_grade_portfolio_equity_scenarios():
+    g = grade_portfolio_equity(_equity(200))
+    assert g.scenario == "portfolio"
+    assert len(g.dimensions) == 5
+    # 净值点不足 60 → insufficient_sample
+    g2 = grade_portfolio_equity(_equity(30))
+    assert g2.insufficient_sample
+
+
+# ── 综合评分（scoring）───────────────────────────────────────────────────────
+
+
+def test_score_strategy_without_wf():
+    s = score_strategy(_perf())
+    assert not s.wf_provided
+    assert "wf_consistency" not in s.components
+    # 无 WF 时权重归一化到四项：50/15/10/5 → /0.8
+    assert s.weights_used["total_return"] == pytest.approx(0.625)
+    assert 0 <= s.total <= 100
+
+
+def test_score_strategy_with_wf():
+    wf = WalkForwardResult(
+        n_windows=7,
+        warmup_ratio=0.3,
+        windows=[
+            WalkForwardWindow(
+                index=i,
+                start="2024-01-01",
+                end="2024-03-01",
+                bars=60,
+                total_return=0.05,
+                sharpe=1.0,
+                max_drawdown=0.05,
+                total_trades=3,
+                win_rate=0.5,
+            )
+            for i in range(7)
+        ],
+    )
+    wf.consistency = 0.857  # 6/7 窗盈利
+    s = score_strategy(_perf(), wf=wf)
+    assert s.wf_provided
+    assert s.weights_used["wf_consistency"] == pytest.approx(0.20)
+    # WF 高一致性应提高总分（其余输入相同）
+    wf_bad = WalkForwardResult(
+        n_windows=7,
+        warmup_ratio=0.3,
+        windows=[
+            WalkForwardWindow(
+                index=i,
+                start="2024-01-01",
+                end="2024-03-01",
+                bars=60,
+                total_return=0.05,
+                sharpe=1.0,
+                max_drawdown=0.05,
+                total_trades=3,
+                win_rate=0.5,
+            )
+            for i in range(7)
+        ],
+    )
+    wf_bad.consistency = 0.14  # 1/7
+    s_bad = score_strategy(_perf(), wf=wf_bad)
+    assert s.total > s_bad.total
+
+
+def test_score_strategy_penalizes_loss():
+    s_win = score_strategy(_perf())
+    s_lose = score_strategy(_perf(total_return=-0.4, sharpe=-0.5))
+    assert s_win.total > s_lose.total
+
+
+def test_score_strategy_serializable():
+    import json
+
+    s = score_strategy(_perf()).to_dict()
+    json.dumps(s)  # 不抛即通过
+    assert {"total", "components", "weights_used", "wf_provided"} <= set(s)
+
+
+def test_score_strategy_nan_safe():
+    s = score_strategy({"sharpe": float("nan"), "total_return": float("inf")})
+    assert math.isfinite(s.total)

tests/unit/test_optimizer_cache_multiseed.py

@@ -0,0 +1,218 @@
+"""优化器两段式加速（指标缓存 + 并行）与多 seed 验证/晋级门槛测试。"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from easy_tdx.backtest.indicator_cache import IndicatorCache
+from easy_tdx.backtest.optimizer import ParamGridOptimizer
+from easy_tdx.backtest.strategy import Strategy
+from easy_tdx.backtest.validation import MultiSeedValidator
+
+
+def _pool_df(n: int = 300, seed: int = 5, drift: float = 0.002) -> pd.DataFrame:
+    rng = np.random.default_rng(seed)
+    dates = pd.date_range("2020-01-01", periods=n, freq="B")
+    close = 10.0 * np.cumprod(1.0 + drift + rng.normal(0, 0.012, n))
+    return pd.DataFrame(
+        {
+            "datetime": dates,
+            "open": close * 0.999,
+            "high": close * 1.01,
+            "low": close * 0.99,
+            "close": close,
+            "vol": 1000.0,
+            "amount": close * 1000,
+        }
+    )
+
+
+# ── IndicatorCache ────────────────────────────────────────────────────────────
+
+
+def test_indicator_cache_hit_and_stats():
+    from easy_tdx.MyTT import MA
+
+    df = _pool_df(100)
+    arr = df["close"].to_numpy()
+    cache = IndicatorCache()
+
+    r1 = cache.get_or_compute(MA, (arr, 5), {})
+    r2 = cache.get_or_compute(MA, (arr, 5), {})
+    assert cache.hits == 1 and cache.misses == 1
+    assert np.allclose(r1, r2, equal_nan=True)  # 前 4 位是 NaN（预热期）
+
+    # 不同参数 → miss
+    cache.get_or_compute(MA, (arr, 10), {})
+    stats = cache.stats()
+    assert stats["total"] == 3
+    assert stats["hit_rate"] == pytest.approx(1 / 3, abs=1e-3)
+
+
+def test_indicator_cache_distinguishes_arrays():
+    from easy_tdx.MyTT import MA
+
+    a = _pool_df(50, seed=1)["close"].to_numpy()
+    b = _pool_df(50, seed=2)["close"].to_numpy()
+    cache = IndicatorCache()
+    cache.get_or_compute(MA, (a, 5), {})
+    cache.get_or_compute(MA, (b, 5), {})
+    assert cache.misses == 2  # 不同数组不误命中
+
+
+# ── 优化器集成（缓存命中 + 结果一致 + 并行）─────────────────────────────────
+
+
+def test_optimizer_cache_reuse_across_grid_points():
+    """2 参数网格：每档参数的指标只算一次，跨点命中。
+
+    ma_cross 的 fast×slow 网格中 MA(close, fast) 会被每个 slow 组合重复
+    请求——缓存应把这些重复请求转为命中。
+    """
+    df = _pool_df(300)
+    grid = {"fast": [5, 10, 15], "slow": [20, 30, 40]}  # 9 点
+    opt = ParamGridOptimizer("ma_cross", grid, df, cash=100_000.0)
+    result = opt.run()
+    assert len(result.results) == 9
+    assert result.cache_stats is not None
+    assert result.cache_stats["hits"] > 0
+    # 9 个点 × 每点 2 个 MA + 2 个 CROSS = 36 次请求；
+    # MA 各 6 档只算 6 次（省 12 次），CROSS 依赖 MA 结果仍逐点计算
+    assert result.cache_stats["misses"] < 36
+
+
+def test_optimizer_cached_results_identical_to_uncached():
+    """缓存开关不改变回测结果（正确性对拍）。"""
+    df = _pool_df(250)
+    grid = {"fast": [5, 10], "slow": [20, 30]}
+
+    # 无缓存路径（optimizer 之前的行为：engine 不挂 cache）
+    opt_plain = ParamGridOptimizer("ma_cross", grid, df, cash=100_000.0)
+    res_plain = opt_plain.run()
+    # 缓存路径
+    opt_cached = ParamGridOptimizer("ma_cross", grid, df, cash=100_000.0)
+    res_cached = opt_cached.run()
+
+    def key_map(res):
+        return {(r.params["fast"], r.params["slow"]): r.total_return for r in res.results}
+
+    assert key_map(res_plain) == key_map(res_cached)
+
+
+def test_optimizer_parallel_matches_serial():
+    """进程池并行结果与串行一致（少量网格冒烟，避免 CI 慢）。"""
+    import sys
+
+    if sys.platform == "win32":
+        # Windows spawn 下进程池在本测试进程中开销大，仅冒烟 4 点
+        df = _pool_df(200)
+        grid = {"fast": [5, 10], "slow": [20, 30]}
+        serial = ParamGridOptimizer("ma_cross", grid, df).run()
+        parallel = ParamGridOptimizer("ma_cross", grid, df, workers=2).run()
+        s = {(r.params["fast"], r.params["slow"]): round(r.total_return, 9) for r in serial.results}
+        p = {
+            (r.params["fast"], r.params["slow"]): round(r.total_return, 9) for r in parallel.results
+        }
+        assert s == p
+
+
+def test_optimizer_cache_stats_serialized():
+    df = _pool_df(150)
+    result = ParamGridOptimizer("rsi_reversal", {"n": [10, 14]}, df).run()
+    d = result.to_dict()
+    assert "cache_stats" in d
+
+
+# ── MultiSeedValidator ───────────────────────────────────────────────────────
+
+
+class _CycleTrader(Strategy):
+    """每 10 根切换持仓（保证各标的有完整回合）。"""
+
+    def init(self) -> None:
+        self._count = 0
+        self._holding = False
+
+    def next(self) -> None:
+        self._count += 1
+        if self._count % 10 == 0:
+            if self._holding:
+                self.sell()
+                self._holding = False
+            else:
+                self.buy()
+                self._holding = True
+
+
+def _pool(n_stocks: int = 6, n: int = 300, drift: float = 0.002) -> dict[str, pd.DataFrame]:
+    return {f"SH:60000{i}": _pool_df(n, seed=i, drift=drift) for i in range(n_stocks)}
+
+
+def test_multiseed_runs_all_pool_by_default():
+    result = MultiSeedValidator(_CycleTrader, _pool(5), n_seeds=2).run()
+    assert result.seeds == [42, 7]
+    # 全池抽样：5 标的 × 2 seed = 10 次运行
+    assert len(result.runs) == 10
+    assert all(r.symbol.startswith("SH:") for r in result.runs)
+
+
+def test_multiseed_sample_size_limits_runs():
+    result = MultiSeedValidator(_CycleTrader, _pool(6), n_seeds=2, sample_size=3).run()
+    # 3 标的 × 2 seed = 6 次；两个 seed 抽到的子集可能不同（顺序随机）
+    assert len(result.runs) == 6
+    seeds = {r.seed for r in result.runs}
+    assert seeds == {42, 7}
+
+
+def test_multiseed_promotion_gates_uptrend():
+    """普涨池：四项默认门槛全过 → promoted。"""
+    result = MultiSeedValidator(_CycleTrader, _pool(6, drift=0.004), n_seeds=2).run()
+    gate_keys = {g.key for g in result.gates}
+    assert gate_keys == {"positive_ratio", "mean_sharpe", "mean_trades", "mean_return"}
+    # 上涨池正收益比例高、均值线全正
+    assert result.positive_ratio >= 0.5
+    assert result.mean_return > 0
+    assert result.promoted is True
+
+
+def test_multiseed_promotion_fails_on_downtrend():
+    """普跌池：正收益比例低 → promoted=False。"""
+    result = MultiSeedValidator(_CycleTrader, _pool(6, drift=-0.004), n_seeds=2).run()
+    assert result.promoted is False
+    assert any(not g.passed for g in result.gates)
+
+
+def test_multiseed_custom_gates_override():
+    """门槛可配置覆盖：mean_return 阈值提高到不可达 → 不晋级。"""
+    result = MultiSeedValidator(
+        _CycleTrader,
+        _pool(4, drift=0.004),
+        n_seeds=1,
+        gates={"mean_return": 999.0},
+    ).run()
+    assert result.promoted is False
+    gate = {g.key: g for g in result.gates}["mean_return"]
+    assert gate.threshold == 999.0
+    assert gate.passed is False
+
+
+def test_multiseed_per_seed_stability_column():
+    result = MultiSeedValidator(_CycleTrader, _pool(5, drift=0.003), n_seeds=3).run()
+    # 跨 seed 稳定性列：每个 seed 一个正收益比例
+    assert len(result.per_seed_positive_ratio) == 3
+    assert set(result.per_seed_positive_ratio) == {"42", "7", "2024"}
+
+
+def test_multiseed_serializable():
+    import json
+
+    d = MultiSeedValidator(_CycleTrader, _pool(3), n_seeds=1).run().to_dict()
+    json.dumps(d)
+    assert {"seeds", "runs", "positive_ratio", "gates", "promoted"} <= set(d)
+
+
+def test_multiseed_empty_pool_raises():
+    with pytest.raises(ValueError, match="不能为空"):
+        MultiSeedValidator(_CycleTrader, {})