mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 21:34:21 +08:00
Justin Gu
c8535e2596
CLI 此前缺失的三块补齐，引擎层复用现成实现（ParamGridOptimizer/ STRATEGY_PRESETS/evaluate_portfolio/PortfolioWalkForwardEngine）， CLI、Web API 与 Python SDK 三条通路能力对等： - 新增 easy-tdx optimize 参数网格寻优命令：单策略网格搜索 （--strategy 用预设网格或 --param 自定义），--all 一键寻优所有 内置策略并按总收益率全局排名（对齐 WebUI /optimize 页与 /backtest/optimize-all/run/async）；--workers 进程级并行； 策略名/参数名联网前前置校验快速失败 - optimizer 新增 optimize_all_strategies 规范实现（模块级 worker 可 pickle、主进程解析 label、跨策略进程池并行、presets 可注入 子集网格），CLI 与后续 Web 端共用 - 新增 easy-tdx strategies 内置策略列表命令（名称/参数默认值/ 预设网格/说明；--output json 与 GET /backtest/strategies 同构） - portfolio 补 --evaluate（组合级一条龙）/ --wf（组合级 WF）/ --auto-fees，输出与 WebUI /portfolio 页同构 - 测试新增 12 例（optimize 互斥/未知策略/未知参数校验、strategies 表格与 JSON、portfolio 新旗标、optimize_all_strategies 排名序/ skipped/JSON 原生类型）；pytest 1611 通过、ruff/mypy 全绿 - 文档同步：README、docs/backtest_usage.md CLI 章节+目录、 CHANGELOG 未发布小节、examples/20_cli/cli_examples.sh 补 回测系列 §39-48（输出样例均为真实行情实测）
144 lines
4.9 KiB
Python
144 lines
4.9 KiB
Python
"""回测 CLI 命令测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from click.testing import CliRunner
|
|
|
|
|
|
class TestBacktestCLI:
|
|
"""测试回测 CLI 命令。"""
|
|
|
|
def test_help(self):
|
|
"""测试 --help 显示帮助。"""
|
|
from easy_tdx.backtest.cli import backtest
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(backtest, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "回测引擎" in result.output
|
|
assert "--strategy-file" in result.output
|
|
assert "--cash" in result.output
|
|
assert "--commission" in result.output
|
|
|
|
def test_missing_strategy_fails(self):
|
|
"""测试不指定策略则失败。"""
|
|
from easy_tdx.backtest.cli import backtest
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(backtest, ["SZ", "000001"])
|
|
assert result.exit_code == 1
|
|
assert "必须指定" in result.output or "错误" in result.output
|
|
|
|
|
|
class TestStrategiesCLI:
|
|
"""测试内置策略列表命令。"""
|
|
|
|
def test_table_output(self):
|
|
"""表格输出应包含注册表策略与预设网格信息。"""
|
|
from easy_tdx.backtest.cli import strategies
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(strategies, [])
|
|
assert result.exit_code == 0
|
|
assert "内置策略" in result.output
|
|
assert "ma_cross" in result.output
|
|
assert "预设网格" in result.output
|
|
|
|
def test_json_output(self):
|
|
"""JSON 输出应为策略 schema 列表（含 preset_grid）。"""
|
|
import json
|
|
|
|
from easy_tdx.backtest.cli import strategies
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(strategies, ["--output", "json"])
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.output)
|
|
names = {entry["name"] for entry in data}
|
|
assert "ma_cross" in names
|
|
ma_cross = next(e for e in data if e["name"] == "ma_cross")
|
|
assert ma_cross["preset_grid"]["fast"]
|
|
assert all("default" in p for p in ma_cross["params"])
|
|
|
|
|
|
class TestOptimizeCLI:
|
|
"""测试参数寻优命令（校验逻辑，不联网）。"""
|
|
|
|
def test_help(self):
|
|
"""测试 --help 显示帮助。"""
|
|
from easy_tdx.backtest.cli import optimize
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(optimize, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "--strategy" in result.output
|
|
assert "--all" in result.output
|
|
assert "--param" in result.output
|
|
|
|
def test_strategy_and_all_mutually_exclusive(self):
|
|
"""--all 与 --strategy 同时指定应报错（联网之前快速失败）。"""
|
|
from easy_tdx.backtest.cli import optimize
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(optimize, ["SZ", "000001", "--strategy", "ma_cross", "--all"])
|
|
assert result.exit_code == 1
|
|
assert "二选一" in result.output
|
|
|
|
def test_missing_mode_fails(self):
|
|
"""--all 与 --strategy 都不指定应报错。"""
|
|
from easy_tdx.backtest.cli import optimize
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(optimize, ["SZ", "000001"])
|
|
assert result.exit_code == 1
|
|
assert "二选一" in result.output
|
|
|
|
def test_unknown_strategy_fails(self):
|
|
"""未知策略名应报错并列出可选值（联网之前快速失败）。"""
|
|
from easy_tdx.backtest.cli import optimize
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(optimize, ["SZ", "000001", "--strategy", "no_such_strat"])
|
|
assert result.exit_code == 1
|
|
assert "未知策略" in result.output
|
|
assert "ma_cross" in result.output
|
|
|
|
def test_unknown_param_fails(self):
|
|
"""--param 传不存在的参数名应报错（避免网格点被静默清空）。"""
|
|
from easy_tdx.backtest.cli import optimize
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
optimize,
|
|
["SZ", "000001", "--strategy", "ma_cross", "--param", "no_such=5,10"],
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "未知参数" in result.output
|
|
|
|
def test_malformed_param_fails(self):
|
|
"""--param 缺少等号应报错。"""
|
|
from easy_tdx.backtest.cli import optimize
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
optimize,
|
|
["SZ", "000001", "--strategy", "ma_cross", "--param", "fast"],
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "错误" in result.output
|
|
|
|
|
|
class TestPortfolioCLIFlags:
|
|
"""测试组合回测命令的新增分析旗标。"""
|
|
|
|
def test_help_includes_evaluate_and_wf(self):
|
|
"""--help 应列出 --evaluate / --wf / --auto-fees。"""
|
|
from easy_tdx.backtest.cli import portfolio
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(portfolio, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "--evaluate" in result.output
|
|
assert "--wf" in result.output
|
|
assert "--auto-fees" in result.output
|