mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 14:34:18 +08:00
315 lines
13 KiB
Python
315 lines
13 KiB
Python
"""向量化快速路径（v1.28）对拍与约束检测单测。
|
||
|
||
核心保证：**同一 df + 同参数下，向量化路径与逐 bar 路径的输出逐位一致**
|
||
（performance / trades / equity_curve / positions 全比对）。
|
||
|
||
- 对拍覆盖：全部内置策略（默认参数，当前 54 个）+ ma_cross/macd/boll/rsi 的非默认
|
||
参数组合 + warmup / 极低资金（买不足 1 手的退化路径）/ 非默认费率与成交价模式；
|
||
- 约束检测：``_vectorize_eligibility`` 的显式约束（无掩码 / 缠论注入）与
|
||
``signal_path`` 的 auto/vector/loop 语义。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from easy_tdx.backtest.engine import BacktestEngine
|
||
from easy_tdx.backtest.strategies import get_registry
|
||
from easy_tdx.backtest.strategy import Strategy
|
||
from easy_tdx.MyTT import MA
|
||
|
||
# ── 合成行情（确定性） ───────────────────────────────────────────────────────
|
||
|
||
|
||
def _synthetic_ohlcv(n: int = 800, seed: int = 42, base: float = 20.0) -> pd.DataFrame:
|
||
"""确定性随机游走 OHLCV（对拍两路径用同一份 df）。"""
|
||
rng = np.random.default_rng(seed)
|
||
rets = rng.normal(0.0005, 0.02, n)
|
||
close = base * np.cumprod(1.0 + rets)
|
||
open_ = np.concatenate([[base], close[:-1]])
|
||
high = np.maximum(open_, close) * (1.0 + np.abs(rng.normal(0.0, 0.008, n)))
|
||
low = np.minimum(open_, close) * (1.0 - np.abs(rng.normal(0.0, 0.008, n)))
|
||
vol = rng.integers(50_000, 5_000_000, n).astype(float)
|
||
dates = pd.bdate_range("2022-01-04", periods=n)
|
||
return pd.DataFrame(
|
||
{
|
||
"datetime": dates,
|
||
"open": open_,
|
||
"high": high,
|
||
"low": low,
|
||
"close": close,
|
||
"vol": vol,
|
||
"amount": vol * close,
|
||
}
|
||
)
|
||
|
||
|
||
def _oscillating_ohlcv(n: int = 800, seed: int = 8, base: float = 20.0) -> pd.DataFrame:
|
||
"""周期振荡 OHLCV（正弦 + 噪声，high/low 恰为 max/min(open, close)）。
|
||
|
||
donchian / wr_reversal 这类「突破 N 日高进 / 跌破 N 日低出」的策略，
|
||
其通道窗口包含当根（upper ≥ 当根 high），``close >= upper`` 只有在
|
||
close == high == 窗口最大（即收盘即创新高）时才成立——high/low 不放大，
|
||
正弦行情每个周期顶/底都会双向触发，覆盖完整的开平仓循环。
|
||
"""
|
||
rng = np.random.default_rng(seed)
|
||
t = np.arange(n)
|
||
close = base * (1.0 + 0.15 * np.sin(2 * np.pi * t / 40) + rng.normal(0.0, 0.002, n))
|
||
open_ = np.concatenate([[base], close[:-1]])
|
||
high = np.maximum(open_, close)
|
||
low = np.minimum(open_, close)
|
||
vol = rng.integers(50_000, 5_000_000, n).astype(float)
|
||
dates = pd.bdate_range("2022-01-04", periods=n)
|
||
return pd.DataFrame(
|
||
{
|
||
"datetime": dates,
|
||
"open": open_,
|
||
"high": high,
|
||
"low": low,
|
||
"close": close,
|
||
"vol": vol,
|
||
"amount": vol * close,
|
||
}
|
||
)
|
||
|
||
|
||
def _assert_perf_equal(a: dict[str, Any], b: dict[str, Any]) -> None:
|
||
"""performance 字典逐位比对（NaN 视为相等——两条路径应产生完全相同的浮点数）。"""
|
||
assert set(a.keys()) == set(b.keys()), f"键集不一致: {set(a) ^ set(b)}"
|
||
for key in a:
|
||
va, vb = a[key], b[key]
|
||
if isinstance(va, float) and isinstance(vb, float) and math.isnan(va) and math.isnan(vb):
|
||
continue
|
||
assert va == vb, f"performance[{key}] 不一致: loop={va!r} vector={vb!r}"
|
||
|
||
|
||
def _assert_results_identical(loop: Any, vec: Any) -> None:
|
||
_assert_perf_equal(loop.performance, vec.performance)
|
||
pd.testing.assert_frame_equal(loop.trades, vec.trades)
|
||
pd.testing.assert_frame_equal(loop.equity_curve, vec.equity_curve)
|
||
pd.testing.assert_frame_equal(loop.positions, vec.positions)
|
||
assert loop.config == vec.config
|
||
|
||
|
||
def _run_both(
|
||
name: str,
|
||
df: pd.DataFrame,
|
||
params: dict[str, Any] | None = None,
|
||
*,
|
||
skip_bounds: bool = False,
|
||
**engine_kw: Any,
|
||
):
|
||
"""同一配置下分别跑逐 bar / 自动画两条路径。"""
|
||
entry = get_registry().get(name)
|
||
loop = BacktestEngine(
|
||
entry.build(params, skip_bounds=skip_bounds), signal_path="loop", **engine_kw
|
||
).run(df)
|
||
vec = BacktestEngine(
|
||
entry.build(params, skip_bounds=skip_bounds), signal_path="auto", **engine_kw
|
||
).run(df)
|
||
return loop, vec
|
||
|
||
|
||
# ── 对拍：全部内置策略 × 默认参数 ───────────────────────────────────────────
|
||
|
||
|
||
#: 已知「默认参数下不会交易」的策略：MyTT 的 WR 是 0~100 刻度（100=超卖），
|
||
#: 而 wr_reversal 默认阈值为 -80/-20（通达信 -100~0 惯例），entry 恒 False。
|
||
#: 这是策略的既有行为（两路径一致地不交易），语义修正不属于向量化改动范围；
|
||
#: 对拍改用非默认参数覆盖（见 test_key_strategies_alternate_params）。
|
||
_DEAD_DEFAULT_STRATEGIES = {"wr_reversal"}
|
||
|
||
|
||
@pytest.mark.parametrize("name", get_registry().names())
|
||
def test_all_builtin_strategies_default_params(name: str) -> None:
|
||
"""全部内置策略：向量化与逐 bar 输出逐位一致（对拍核心保证）。"""
|
||
df = _synthetic_ohlcv()
|
||
loop, vec = _run_both(name, df)
|
||
# 确认确实产生了交易（空交易的对拍没有意义）：随机游走无交易换种子，
|
||
# 再无交易换振荡行情（donchian/wr 等带状策略只在振荡行情双向触发）
|
||
if len(loop.trades) == 0:
|
||
loop, vec = _run_both(name, _synthetic_ohlcv(seed=7, base=50.0))
|
||
if len(loop.trades) == 0:
|
||
loop, vec = _run_both(name, _oscillating_ohlcv())
|
||
if name not in _DEAD_DEFAULT_STRATEGIES:
|
||
assert len(loop.trades) > 0, f"{name} 在三组行情下均无交易，对拍无效"
|
||
_assert_results_identical(loop, vec)
|
||
|
||
|
||
# ── 对拍：关键策略的非默认参数 + 引擎配置变化 ───────────────────────────────
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("name", "params"),
|
||
[
|
||
("ma_cross", {"fast": 10, "slow": 60}),
|
||
("ma_cross", {"fast": 3, "slow": 8}),
|
||
("macd", {"short": 6, "long": 13, "signal": 5}),
|
||
("boll_breakout", {"n": 10, "p": 2.5}),
|
||
("rsi_reversal", {"n": 7, "oversold": 25, "overbought": 78}),
|
||
("donchian", {"n": 20}),
|
||
],
|
||
)
|
||
def test_key_strategies_alternate_params(name: str, params: dict[str, Any]) -> None:
|
||
"""任务点名的四个策略（含非默认参数）对拍一致。"""
|
||
df = _synthetic_ohlcv(seed=99)
|
||
loop, vec = _run_both(name, df, params)
|
||
if len(loop.trades) == 0:
|
||
loop, vec = _run_both(name, _oscillating_ohlcv(), params)
|
||
assert len(loop.trades) > 0, f"{name}{params} 两组行情下均无交易，对拍无效"
|
||
_assert_results_identical(loop, vec)
|
||
|
||
|
||
def test_wr_reversal_parity_with_skip_bounds() -> None:
|
||
"""wr_reversal 的向量化机制对拍（需 skip_bounds 越过死的默认边界）。
|
||
|
||
其阈值参数边界为负数区间（-100~-40 / -60~0），而 MyTT 的 WR 是 0~100
|
||
刻度——任何合法参数都无法触发交易（策略现状如此，两路径行为一致）。
|
||
为了让它的掩码/状态机路径也被对拍覆盖，用 skip_bounds 传 0~100 刻度内
|
||
的阈值绕过边界（寻优器同款机制）。
|
||
"""
|
||
df = _oscillating_ohlcv()
|
||
loop, vec = _run_both(
|
||
"wr_reversal", df, {"n": 14, "oversold": 40, "overbought": 60}, skip_bounds=True
|
||
)
|
||
assert len(loop.trades) > 0
|
||
_assert_results_identical(loop, vec)
|
||
|
||
|
||
@pytest.mark.parametrize("warmup", [0, 20, 100])
|
||
def test_warmup_bars_consistency(warmup: int) -> None:
|
||
"""warmup 期不产生信号：两路径一致（向量化按候选 bar 过滤）。"""
|
||
df = _synthetic_ohlcv(seed=11)
|
||
loop, vec = _run_both("ma_cross", df, warmup_bars=warmup)
|
||
_assert_results_identical(loop, vec)
|
||
|
||
|
||
def test_degenerate_low_cash_consistency() -> None:
|
||
"""极低资金：BUY 信号买不足 1 手（策略仓位状态不变），两路径仍一致。"""
|
||
df = _synthetic_ohlcv(seed=5, base=200.0) # 高价股 + 小资金 → 整手买入失败
|
||
loop, vec = _run_both("ma_cross", df, cash=1500.0)
|
||
_assert_results_identical(loop, vec)
|
||
|
||
|
||
def test_nondefault_fees_and_execution_consistency() -> None:
|
||
"""非默认费率/滑点/成交价模式：信号路径无关下游，但全流程仍应一致。"""
|
||
df = _synthetic_ohlcv(seed=13)
|
||
kw: dict[str, Any] = {
|
||
"commission": 0.0005,
|
||
"stamp_tax": 0.0005,
|
||
"slippage": 0.01,
|
||
"execution": "next_close",
|
||
}
|
||
loop, vec = _run_both("macd", df, **kw)
|
||
_assert_results_identical(loop, vec)
|
||
|
||
|
||
def test_indicator_cache_consistency() -> None:
|
||
"""挂载指标缓存（寻优场景）时向量化路径照常工作且一致。"""
|
||
from easy_tdx.backtest.indicator_cache import IndicatorCache
|
||
|
||
df = _synthetic_ohlcv(seed=21)
|
||
loop = BacktestEngine(
|
||
get_registry().get("ma_cross").build(), signal_path="loop", indicator_cache=IndicatorCache()
|
||
).run(df)
|
||
cache = IndicatorCache()
|
||
vec = BacktestEngine(
|
||
get_registry().get("ma_cross").build(), signal_path="auto", indicator_cache=cache
|
||
).run(df)
|
||
_assert_results_identical(loop, vec)
|
||
|
||
|
||
# ── 约束检测（显式、可测试） ─────────────────────────────────────────────────
|
||
|
||
|
||
class _PlainStrategy(Strategy):
|
||
"""未实现 entry_exit_masks 的普通策略（应走逐 bar）。"""
|
||
|
||
def init(self) -> None:
|
||
self.ma = self.I(MA, self.data.close, 5)
|
||
|
||
def next(self) -> None:
|
||
if self.ma[self._bar_index] > 0 and self.position["size"] == 0:
|
||
self.buy()
|
||
|
||
|
||
def test_eligibility_requires_masks_hook() -> None:
|
||
"""未覆写 entry_exit_masks → 不具备资格（原因可读）。"""
|
||
engine = BacktestEngine(_PlainStrategy)
|
||
eligible, reason = engine._vectorize_eligibility(_PlainStrategy(), None)
|
||
assert eligible is False
|
||
assert "entry_exit_masks" in reason
|
||
|
||
|
||
def test_eligibility_rejects_chanlun() -> None:
|
||
"""缠论注入（result 或 level）→ 不具备资格。"""
|
||
strat = get_registry().get("ma_cross").build()
|
||
engine = BacktestEngine(strat, chanlun_level="DAILY")
|
||
eligible, reason = engine._vectorize_eligibility(strat, None)
|
||
assert eligible is False
|
||
assert "缠论" in reason
|
||
|
||
engine2 = BacktestEngine(strat)
|
||
eligible2, reason2 = engine2._vectorize_eligibility(strat, {"fake": "chanlun"})
|
||
assert eligible2 is False
|
||
assert "缠论" in reason2
|
||
|
||
|
||
def test_eligibility_accepts_builtin() -> None:
|
||
"""内置策略（无缠论）→ 具备资格。"""
|
||
strat = get_registry().get("ma_cross").build()
|
||
engine = BacktestEngine(strat)
|
||
eligible, _ = engine._vectorize_eligibility(strat, None)
|
||
assert eligible is True
|
||
|
||
|
||
def test_signal_path_vector_forces_or_raises() -> None:
|
||
"""signal_path='vector'：满足约束时正常，不满足时显式抛错。"""
|
||
df = _synthetic_ohlcv(200)
|
||
# 满足约束：正常运行且与 loop 一致
|
||
loop, vec = _run_both("ma_cross", df)
|
||
forced = BacktestEngine(get_registry().get("ma_cross").build(), signal_path="vector").run(df)
|
||
_assert_results_identical(loop, forced)
|
||
|
||
# 不满足约束：显式 ValueError（而非静默回退）
|
||
with pytest.raises(ValueError, match="向量化约束"):
|
||
BacktestEngine(_PlainStrategy, signal_path="vector").run(df)
|
||
|
||
|
||
def test_signal_path_invalid_rejected() -> None:
|
||
with pytest.raises(ValueError, match="signal_path"):
|
||
BacktestEngine(_PlainStrategy, signal_path="fast")
|
||
|
||
|
||
def test_auto_falls_back_on_mask_shape_mismatch() -> None:
|
||
"""掩码形状错误（策略实现 bug）：auto 静默回退逐 bar，结果仍一致。"""
|
||
|
||
class _BadMaskStrategy(_PlainStrategy):
|
||
def entry_exit_masks(self) -> tuple[np.ndarray, np.ndarray]:
|
||
return np.zeros(3, dtype=bool), np.zeros(3, dtype=bool)
|
||
|
||
df = _synthetic_ohlcv(200, seed=3)
|
||
loop = BacktestEngine(_PlainStrategy, signal_path="loop").run(df)
|
||
fallback = BacktestEngine(_BadMaskStrategy, signal_path="auto").run(df)
|
||
_assert_results_identical(loop, fallback)
|
||
|
||
|
||
def test_vector_path_actually_used_for_builtins() -> None:
|
||
"""默认 signal_path='auto' 下内置策略确实走了向量化（防止回退被掩盖）。
|
||
|
||
若某策略信号依赖路径状态（无法用静态掩码等价表达），应在此说明并
|
||
考虑引擎走逐 bar 回放的白名单机制（当前无此类策略）。
|
||
"""
|
||
from easy_tdx.backtest.strategy import Strategy as Base
|
||
|
||
for name in get_registry().names():
|
||
strat_cls = get_registry().get(name).strategy_cls
|
||
assert strat_cls.entry_exit_masks is not Base.entry_exit_masks, (
|
||
f"{name} 未实现 entry_exit_masks，auto 将永远走逐 bar"
|
||
)
|