mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 20:24:19 +08:00
218 lines
8.1 KiB
Python
218 lines
8.1 KiB
Python
"""单元测试：参数网格寻优器."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from easy_tdx.backtest.optimizer import (
|
||
GridPointResult,
|
||
OptimizeResult,
|
||
ParamGridOptimizer,
|
||
)
|
||
|
||
|
||
def _make_df(n: int = 150, seed: int = 42) -> pd.DataFrame:
|
||
"""生成带趋势的合成 OHLCV（确保均线策略能产生交易）。"""
|
||
rng = np.random.default_rng(seed)
|
||
close = 10.0 + np.cumsum(rng.normal(0, 0.3, n) + 0.05)
|
||
return pd.DataFrame(
|
||
{
|
||
"datetime": pd.date_range("2024-01-01", periods=n, freq="B"),
|
||
"open": close - 0.1,
|
||
"high": close + 0.2,
|
||
"low": close - 0.2,
|
||
"close": close,
|
||
"vol": rng.integers(1000, 10000, n).astype(float),
|
||
"amount": close * 5000,
|
||
}
|
||
)
|
||
|
||
|
||
class TestParamGridOptimizer:
|
||
"""寻优器核心逻辑."""
|
||
|
||
def test_grid_enumeration(self) -> None:
|
||
"""网格点数应等于笛卡尔积大小。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid={"fast": [5, 10], "slow": [15, 20, 30]},
|
||
df=_make_df(),
|
||
)
|
||
result = opt.run()
|
||
assert len(result.results) == 6 # 2 × 3
|
||
|
||
def test_results_sorted_by_return_descending(self) -> None:
|
||
"""结果应按 total_return 降序排列。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid={"fast": [5, 10, 20], "slow": [15, 20, 30]},
|
||
df=_make_df(),
|
||
)
|
||
result = opt.run()
|
||
returns = [r.total_return for r in result.results]
|
||
assert returns == sorted(returns, reverse=True)
|
||
|
||
def test_best_is_first_result(self) -> None:
|
||
"""best 应是 results[0]。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid={"fast": [5, 10], "slow": [20, 30]},
|
||
df=_make_df(),
|
||
)
|
||
result = opt.run()
|
||
assert result.best is not None
|
||
assert result.best.params == result.results[0].params
|
||
assert result.best.total_return == result.results[0].total_return
|
||
|
||
def test_heatmap_2_params(self) -> None:
|
||
"""2 参数应生成热力图矩阵。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid={"fast": [5, 10, 20], "slow": [15, 20, 30]},
|
||
df=_make_df(),
|
||
)
|
||
result = opt.run()
|
||
assert result.heatmap is not None
|
||
assert result.heatmap["x_name"] == "fast"
|
||
assert result.heatmap["y_name"] == "slow"
|
||
assert len(result.heatmap["x"]) == 3
|
||
assert len(result.heatmap["y"]) == 3
|
||
# 3×3=9 中 fast=20×slow=15（倒挂）与 20×20（相等）被语义约束跳过（issue #39）
|
||
assert len(result.heatmap["data"]) == 7
|
||
|
||
def test_no_heatmap_for_single_param(self) -> None:
|
||
"""1 参数时 heatmap 应为 None。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="rsi_reversal",
|
||
param_grid={"n": [7, 14, 21]},
|
||
df=_make_df(),
|
||
)
|
||
result = opt.run()
|
||
assert result.heatmap is None
|
||
assert len(result.results) == 3
|
||
|
||
def test_inverted_combos_skipped(self) -> None:
|
||
"""网格含倒挂组合（fast≥slow）时应全部跳过，best 不可能是倒挂（issue #39）。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid={"fast": [5, 20, 30], "slow": [10, 20]},
|
||
df=_make_df(),
|
||
)
|
||
result = opt.run()
|
||
# 6 组合中仅 (5,10) 与 (5,20) 语义有效
|
||
assert len(result.results) == 2
|
||
assert all(r.params["fast"] < r.params["slow"] for r in result.results)
|
||
assert result.best is not None
|
||
assert result.best.params["fast"] < result.best.params["slow"]
|
||
|
||
def test_grid_size_limit_exceeded(self) -> None:
|
||
"""超过 MAX_GRID_POINTS 应抛 ValueError。"""
|
||
big_grid = {f"p{i}": list(range(5)) for i in range(6)} # 5^6 = 15625
|
||
with pytest.raises(ValueError, match="超过上限"):
|
||
ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid=big_grid,
|
||
df=_make_df(),
|
||
)
|
||
|
||
def test_empty_value_list_rejected(self) -> None:
|
||
"""空取值列表应抛 ValueError。"""
|
||
with pytest.raises(ValueError, match="空取值列表"):
|
||
ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid={"fast": [], "slow": [20]},
|
||
df=_make_df(),
|
||
)
|
||
|
||
def test_to_dict_serializable(self) -> None:
|
||
"""to_dict 应返回 JSON 兼容结构。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="ma_cross",
|
||
param_grid={"fast": [5, 10], "slow": [20, 30]},
|
||
df=_make_df(),
|
||
)
|
||
result = opt.run()
|
||
d = result.to_dict()
|
||
|
||
assert d["strategy"] == "ma_cross"
|
||
assert d["param_names"] == ["fast", "slow"]
|
||
assert len(d["results"]) == 4
|
||
assert d["best"] is not None
|
||
assert "params" in d["best"]
|
||
assert "total_return" in d["best"]
|
||
|
||
def test_to_dict_cleans_nan(self) -> None:
|
||
"""NaN 指标应被清洗为 None（JSON 兼容）。"""
|
||
# 构造含 NaN 的结果（无交易的参数组合 sharpe 可能 NaN）
|
||
result = OptimizeResult(
|
||
strategy="test",
|
||
param_names=["n"],
|
||
results=[
|
||
GridPointResult(params={"n": 1}, total_return=float("nan"), sharpe=float("inf")),
|
||
],
|
||
)
|
||
d = result.to_dict()
|
||
assert d["results"][0]["total_return"] is None
|
||
assert d["results"][0]["sharpe"] is None
|
||
|
||
def test_unknown_strategy_raises(self) -> None:
|
||
"""未知策略应在 run() 时抛 KeyError。"""
|
||
opt = ParamGridOptimizer(
|
||
strategy_name="nope",
|
||
param_grid={"x": [1]},
|
||
df=_make_df(),
|
||
)
|
||
with pytest.raises(KeyError):
|
||
opt.run()
|
||
|
||
|
||
class TestStrategyPresets:
|
||
"""预设网格与策略注册表的一致性。
|
||
|
||
「一键寻优所有策略」只遍历 STRATEGY_PRESETS（backtest.py optimize-all），
|
||
未登记的策略会被静默跳过——v1.30.2 曾因此只寻优 19/54 个策略。
|
||
这里双向锁定：每个已注册策略必须有预设，且网格合法。
|
||
"""
|
||
|
||
def test_every_registered_strategy_has_preset(self) -> None:
|
||
"""注册表与 STRATEGY_PRESETS 键集一一对应（双向：不多不少）。"""
|
||
from easy_tdx.backtest.strategies import (
|
||
builtin, # noqa: F401 # 触发注册
|
||
get_registry,
|
||
)
|
||
from easy_tdx.backtest.strategies.presets import STRATEGY_PRESETS
|
||
|
||
names = set(get_registry().names())
|
||
missing = names - set(STRATEGY_PRESETS)
|
||
extra = set(STRATEGY_PRESETS) - names
|
||
assert not missing, f"这些策略无预设网格，会被一键寻优静默跳过: {sorted(missing)}"
|
||
assert not extra, f"这些预设指向未注册的策略: {sorted(extra)}"
|
||
|
||
def test_preset_values_within_param_bounds(self) -> None:
|
||
"""预设取值必须在参数 schema 边界内（否则寻优端点 422）。"""
|
||
from easy_tdx.backtest.strategies import (
|
||
builtin, # noqa: F401 # 触发注册
|
||
get_registry,
|
||
)
|
||
from easy_tdx.backtest.strategies.presets import STRATEGY_PRESETS
|
||
|
||
registry = get_registry()
|
||
for name, grid in STRATEGY_PRESETS.items():
|
||
params = {p.name: p for p in registry.get(name).params}
|
||
for pname, values in grid.items():
|
||
assert pname in params, f"{name}: 预设参数 {pname} 不在 schema 中"
|
||
for v in values:
|
||
params[pname].validate(v) # 越界抛 ValueError
|
||
|
||
def test_preset_grid_size_within_limit(self) -> None:
|
||
"""单策略笛卡尔积 ≤ 200（ParamGridOptimizer.MAX_GRID_POINTS）。"""
|
||
import math
|
||
|
||
from easy_tdx.backtest.strategies.presets import STRATEGY_PRESETS
|
||
|
||
for name, grid in STRATEGY_PRESETS.items():
|
||
size = math.prod(len(v) for v in grid.values()) if grid else 1
|
||
assert size <= 200, f"{name}: 预设网格 {size} 点超上限"
|