simon
fce725e13c
从 Point-in-Time 股票筛选到统一 Web 前端的完整链路： 筛选 → 画像 → 策略 → 回测 → Walk-forward → 绩效分析 → 报告/前端。 架构 - 数据层与策略层分离；策略代码不写 SQL，只经 data/repo.py 取数 - 所有业务阈值集中在 config/*.yml，代码零硬编码（字段写错直接报错） - 报告只做「run_id → SQL → 渲染」，不做任何计算，数字可追溯 - 前后端分离：output/ 静态站点 + hdiv web 提供的 REST API 数据安全 - 只增不删：SQL 钩子拦截 DELETE/DROP/TRUNCATE，并有源码扫描测试守护 - qlib 原有表只读，本项目数据写入 hd_ 前缀表 - 回补使用 INSERT IGNORE，保证既有行零改动 - .env 存密钥且已 gitignore；output/、logs/、.venv/ 不入库 交付物 - 30 张 hd_* 表、7 个 YAML 配置、283 项自动化测试 - 统一 Web 前端（hash 路由 SPA）+ nginx 部署配置与 launchd 托管脚本 如实声明的限制 - 策略缺少稳定的样本外超额收益（Walk-forward 7 窗口均值 -0.95%， 基准 +2.29%）；其价值体现在回撤控制，而非超额收益 - 涨跌停/停牌约束仅覆盖 2019 年起；index_weight 尚未填充 - AI Agent 层（plan.md 第四版 P8）未实现 详见 docs/user-guide.md 与 docs/implementation-status.md。
304 lines
11 KiB
Python
304 lines
11 KiB
Python
"""CLI 接口契约测试。
|
||
|
||
**为什么需要**：使用手册（``docs/user-guide.md``）里的命令就是用户看到的接口。
|
||
文档与实现脱节是极易发生且很难察觉的问题 —— 实测发现手册记载的
|
||
``hdiv sync financial --interleaved`` 在 CLI 中根本没有暴露，
|
||
而 ``hdiv ddl verify`` 的输出格式也与手册不符。
|
||
|
||
本测试把「手册承诺的接口」变成可执行断言：
|
||
只要有人删掉一个参数或改掉命令名，测试立刻失败。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from hdiv.cli import build_parser
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def parser():
|
||
return build_parser()
|
||
|
||
|
||
def _subparsers(p):
|
||
"""取子命令表。同时支持两种声明方式：
|
||
|
||
- ``add_subparsers()`` → choices 是 dict（sync/universe/...）
|
||
- ``add_argument("action", choices=[...])`` → choices 是 list（ddl/report/strategy）
|
||
"""
|
||
for action in p._actions:
|
||
ch = getattr(action, "choices", None)
|
||
if not ch:
|
||
continue
|
||
if isinstance(ch, dict):
|
||
return ch
|
||
raise AssertionError("未找到子命令")
|
||
|
||
|
||
def _action_choices(p) -> set[str]:
|
||
"""取位置参数（如 ddl 的 plan/apply/verify）的取值集合。"""
|
||
for action in p._actions:
|
||
ch = getattr(action, "choices", None)
|
||
if isinstance(ch, (list, tuple, set)) and not action.option_strings:
|
||
return set(ch)
|
||
return set()
|
||
|
||
|
||
def _opts(p) -> set[str]:
|
||
out: set[str] = set()
|
||
for a in p._actions:
|
||
out.update(a.option_strings)
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 手册记载的顶层命令必须全部存在
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DOCUMENTED_COMMANDS = [
|
||
"ddl", "sync", "audit", "report", "universe",
|
||
"profile", "strategy", "backtest", "sensitivity",
|
||
"web", "site",
|
||
]
|
||
|
||
|
||
def test_all_documented_commands_exist(parser) -> None:
|
||
subs = _subparsers(parser)
|
||
missing = [c for c in DOCUMENTED_COMMANDS if c not in subs]
|
||
assert not missing, f"手册记载但 CLI 不存在的命令：{missing}"
|
||
|
||
|
||
def test_no_undocumented_commands(parser) -> None:
|
||
"""反向检查：CLI 不应有手册未提及的命令。"""
|
||
subs = set(_subparsers(parser))
|
||
extra = subs - set(DOCUMENTED_COMMANDS)
|
||
assert not extra, f"CLI 存在手册未记载的命令：{sorted(extra)}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 手册记载的动作/参数必须存在
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"cmd,expected",
|
||
[
|
||
("ddl", {"plan", "apply", "verify"}),
|
||
("report", {"index", "validate"}),
|
||
("strategy", {"validate", "register", "list", "diff"}),
|
||
(
|
||
"sync",
|
||
{"dividend", "financial", "index", "price", "trading", "backfill"},
|
||
),
|
||
("site", {"normalize", "archive", "build", "status"}),
|
||
],
|
||
)
|
||
def test_documented_actions_exist(parser, cmd: str, expected: set[str]) -> None:
|
||
sub = _subparsers(parser)[cmd]
|
||
actual = _action_choices(sub)
|
||
if not actual: # 用 add_subparsers 声明的命令
|
||
actual = set(_subparsers(sub))
|
||
missing = expected - actual
|
||
assert not missing, f"hdiv {cmd} 缺少手册记载的动作：{sorted(missing)}"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"cmd,flags",
|
||
[
|
||
("sync", {"--only-missing", "--limit", "--symbols", "--apis",
|
||
"--interleaved", "--start", "--end", "--no-resume", "--no-weight"}),
|
||
("universe", {"-c", "--config", "--asof", "--no-persist", "--no-html"}),
|
||
("profile", {"--universe-run", "--symbols", "--asof", "--html-limit"}),
|
||
("backtest", {"-s", "--strategy", "--mode", "--start", "--end", "--universe-run"}),
|
||
("sensitivity", {"-s", "--strategy", "--sweep"}),
|
||
("strategy", {"-f", "--file", "--other"}),
|
||
("audit", {"--no-persist", "--no-html"}),
|
||
("report", {"--dir"}),
|
||
("web", {"--host", "--port", "--api-only", "--out-dir"}),
|
||
],
|
||
)
|
||
def test_documented_flags_exist(parser, cmd: str, flags: set[str]) -> None:
|
||
sub = _subparsers(parser)[cmd]
|
||
have = _opts(sub)
|
||
missing = flags - have
|
||
assert not missing, f"hdiv {cmd} 缺少手册记载的参数：{sorted(missing)}"
|
||
|
||
|
||
def test_sync_interleaved_is_a_real_flag(parser) -> None:
|
||
"""回归：手册要求 ``hdiv sync financial --interleaved``，CLI 必须真的支持。"""
|
||
sub = _subparsers(parser)["sync"]
|
||
assert "--interleaved" in _opts(sub)
|
||
# 且必须真的能解析（不只是声明）
|
||
args = parser.parse_args(["sync", "financial", "--interleaved"])
|
||
assert args.interleaved is True
|
||
args2 = parser.parse_args(["sync", "financial"])
|
||
assert args2.interleaved is False
|
||
|
||
|
||
def test_backtest_universe_run_flag(parser) -> None:
|
||
"""股票池 ↔ 回测 的关联入口：``--universe-run`` 必须存在且可解析。"""
|
||
args = parser.parse_args(["backtest", "--universe-run", "abc123"])
|
||
assert args.universe_run == "abc123"
|
||
args2 = parser.parse_args(["backtest"])
|
||
assert args2.universe_run is None
|
||
|
||
|
||
@pytest.mark.db
|
||
def test_frozen_universe_is_used_when_run_id_given() -> None:
|
||
"""指定 --universe-run 时引擎必须使用该股票池，而不是重新筛选。"""
|
||
from datetime import date
|
||
|
||
from hdiv.backtest.engine import BacktestEngine
|
||
from hdiv.core.config import load_config
|
||
from hdiv.data import db
|
||
|
||
db.load_dotenv_once()
|
||
try:
|
||
cfg = load_config("datasource")
|
||
df = db.read_sql(
|
||
"SELECT run_id FROM hd_universe_run WHERE deleted_at IS NULL "
|
||
"ORDER BY created_at DESC LIMIT 1", cfg=cfg,
|
||
)
|
||
if df.empty:
|
||
pytest.skip("没有筛选记录")
|
||
rid = df["run_id"].iloc[0]
|
||
except Exception as exc:
|
||
pytest.skip(f"数据库不可用：{exc}")
|
||
|
||
eng = BacktestEngine.from_strategy(
|
||
"config/strategy/high_dividend_v1.yml", universe_run_id=rid
|
||
)
|
||
assert eng.universe_run_id == rid
|
||
ctx = eng._prepare(eng.repo.trading_days(date(2024, 1, 1), date(2024, 6, 28)),
|
||
verbose=False)
|
||
sets = list(ctx["universe_by_refresh"].values())
|
||
assert sets, "应产生股票池"
|
||
assert all(x == sets[0] for x in sets), "冻结股票池在各调仓日必须完全相同"
|
||
|
||
|
||
def test_backtest_mode_choices(parser) -> None:
|
||
"""手册只承诺 single / walkforward 两种模式。"""
|
||
sub = _subparsers(parser)["backtest"]
|
||
for a in sub._actions:
|
||
if "--mode" in a.option_strings:
|
||
assert set(a.choices) == {"single", "walkforward"}
|
||
return
|
||
raise AssertionError("backtest 缺少 --mode 参数")
|
||
|
||
|
||
def test_price_which_choices(parser) -> None:
|
||
sub = _subparsers(parser)["sync"]
|
||
for a in sub._actions:
|
||
if "--which" in a.option_strings:
|
||
assert set(a.choices) == {"daily", "adj_factor", "daily_basic"}
|
||
return
|
||
raise AssertionError("sync 缺少 --which 参数")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 手册记载的输出格式
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_ddl_verify_output_mentions_table_count() -> None:
|
||
"""手册写的是「OK：30 张 hd_* 表结构全部符合 schema 定义」，输出须含数量。"""
|
||
import inspect
|
||
|
||
from hdiv import cli
|
||
from hdiv.data.schema import ALL_TABLES
|
||
|
||
src = inspect.getsource(cli.cmd_ddl)
|
||
assert "len(ddl.ALL_TABLES)" in src, "ddl verify 输出应包含表的数量"
|
||
assert len(ALL_TABLES) == 30
|
||
|
||
|
||
def test_help_text_is_chinese(parser) -> None:
|
||
"""手册面向中文用户，所有命令都应有中文说明。"""
|
||
subs = _subparsers(parser)
|
||
for name in DOCUMENTED_COMMANDS:
|
||
assert subs[name].description, f"hdiv {name} 缺少说明文字"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 手册文件本身
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_user_guide_exists_and_covers_all_commands() -> None:
|
||
from hdiv.core.paths import project_root
|
||
|
||
guide = project_root() / "docs" / "user-guide.md"
|
||
assert guide.is_file(), "缺少使用手册 docs/user-guide.md"
|
||
text = guide.read_text(encoding="utf-8")
|
||
for cmd in DOCUMENTED_COMMANDS:
|
||
assert f"hdiv {cmd}" in text, f"手册未记载命令：hdiv {cmd}"
|
||
# 手册必须如实记录已知限制与部署要求（这两点最容易在文档里被淡化）
|
||
assert "已知限制" in text
|
||
assert "echarts.min.js" in text, "手册必须说明图表库的部署依赖"
|
||
assert "asset_mode" in text, "手册必须说明自包含模式的开关"
|
||
# 前后端分离的部署要点必须在手册里
|
||
assert "/api" in text, "手册必须说明 API 反代"
|
||
assert "nginx" in text, "手册必须给出 nginx 配置指引"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 参数组合与错误呈现
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
# 只有这三个命令同时具备 --no-persist 与 --html（audit 传内存结果，两者可共存）
|
||
@pytest.mark.parametrize("cmd", ["universe", "backtest"])
|
||
def test_no_persist_with_html_is_rejected_clearly(cmd: str) -> None:
|
||
"""回归：--no-persist 与 --html 互相矛盾，必须给出可操作提示。
|
||
|
||
静态报告只从数据库读取（保证数字可追溯到 SQL），未落库的运行渲染不出来。
|
||
早期实现会崩在报告生成器的 ValueError 里，用户看不出是参数冲突。
|
||
"""
|
||
import subprocess
|
||
import sys
|
||
|
||
r = subprocess.run(
|
||
[sys.executable, "-m", "hdiv", cmd, "--no-persist", "--html"],
|
||
capture_output=True, text=True, env={**__import__("os").environ, "PYTHONPATH": "src"},
|
||
)
|
||
assert r.returncode == 1
|
||
out = r.stdout + r.stderr
|
||
assert "--no-persist 与 --html 不能同时使用" in out, out[:400]
|
||
# 必须是友好提示，而不是 traceback
|
||
assert "Traceback" not in out, "用户可理解的错误不应打印调用栈"
|
||
|
||
|
||
def test_audit_can_combine_no_persist_and_html() -> None:
|
||
"""audit 传的是内存结果而非按 run_id 读库，因此两者可以共存。"""
|
||
import inspect
|
||
|
||
from hdiv import cli
|
||
|
||
src = inspect.getsource(cli.cmd_audit)
|
||
assert "_reject_no_persist_with_html" not in src, \
|
||
"audit 不需拦截该组合（报告用内存 summary 渲染）"
|
||
|
||
|
||
def test_hdiv_error_is_caught_without_traceback() -> None:
|
||
"""main() 应优雅处理 HdivError（数据缺失/配置冲突等），不打印调用栈。"""
|
||
import inspect
|
||
|
||
from hdiv import cli
|
||
|
||
src = inspect.getsource(cli.main)
|
||
assert "except HdivError" in src, "main() 未捕获 HdivError"
|
||
branch = src.split("except HdivError")[1]
|
||
assert "print_exc" not in branch, "HdivError 分支不应打印调用栈"
|
||
assert "return 1" in branch, "应返回非零退出码"
|
||
|
||
|
||
def test_html_flag_exists_on_all_report_producing_commands() -> None:
|
||
"""所有会产出报告的命令都应有 --html 开关且默认关闭。"""
|
||
from hdiv.cli import build_parser
|
||
|
||
p = build_parser()
|
||
for cmd in ("universe", "profile", "backtest", "audit", "sensitivity"):
|
||
args = p.parse_args([cmd])
|
||
assert getattr(args, "html", None) is False, f"hdiv {cmd} --html 默认应为 False"
|