diff --git a/CHANGELOG.md b/CHANGELOG.md index 058ae8d..16f8d0e 100644 --- a/CHANGELOG.md +++ b/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`。 diff --git a/pyproject.toml b/pyproject.toml index 79b2874..3c32557 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/easy_tdx/backtest/benchmark.py b/src/easy_tdx/backtest/benchmark.py new file mode 100644 index 0000000..0f3268c --- /dev/null +++ b/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), + }, + } diff --git a/src/easy_tdx/backtest/fitness.py b/src/easy_tdx/backtest/fitness.py new file mode 100644 index 0000000..6c36cbe --- /dev/null +++ b/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 diff --git a/src/easy_tdx/backtest/grading.py b/src/easy_tdx/backtest/grading.py new file mode 100644 index 0000000..6ce3a57 --- /dev/null +++ b/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 diff --git a/src/easy_tdx/backtest/indicator_cache.py b/src/easy_tdx/backtest/indicator_cache.py new file mode 100644 index 0000000..10591ce --- /dev/null +++ b/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)) diff --git a/src/easy_tdx/backtest/optimizer.py b/src/easy_tdx/backtest/optimizer.py index 202c064..f40c48d 100644 --- a/src/easy_tdx/backtest/optimizer.py +++ b/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= 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 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 diff --git a/src/easy_tdx/backtest/scoring.py b/src/easy_tdx/backtest/scoring.py new file mode 100644 index 0000000..9010e1e --- /dev/null +++ b/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, + ) diff --git a/src/easy_tdx/backtest/strategy.py b/src/easy_tdx/backtest/strategy.py index 77c7dad..dbb36d0 100644 --- a/src/easy_tdx/backtest/strategy.py +++ b/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)) diff --git a/src/easy_tdx/backtest/types.py b/src/easy_tdx/backtest/types.py index f5d235c..dc47b69 100644 --- a/src/easy_tdx/backtest/types.py +++ b/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: """回测完整结果。 diff --git a/src/easy_tdx/backtest/validation.py b/src/easy_tdx/backtest/validation.py new file mode 100644 index 0000000..f6e93c1 --- /dev/null +++ b/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) diff --git a/src/easy_tdx/backtest/walkforward.py b/src/easy_tdx/backtest/walkforward.py new file mode 100644 index 0000000..3b3fc2c --- /dev/null +++ b/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)) diff --git a/tests/unit/test_backtest_fitness_benchmark.py b/tests/unit/test_backtest_fitness_benchmark.py new file mode 100644 index 0000000..52e5e34 --- /dev/null +++ b/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 diff --git a/tests/unit/test_backtest_walkforward.py b/tests/unit/test_backtest_walkforward.py new file mode 100644 index 0000000..7ddbec0 --- /dev/null +++ b/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 diff --git a/tests/unit/test_grading_scoring.py b/tests/unit/test_grading_scoring.py new file mode 100644 index 0000000..035ecbe --- /dev/null +++ b/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) diff --git a/tests/unit/test_optimizer_cache_multiseed.py b/tests/unit/test_optimizer_cache_multiseed.py new file mode 100644 index 0000000..999b334 --- /dev/null +++ b/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, {})