"""CLI 接口契约测试。 **为什么需要**：使用手册（``docs/user-guide.md``）里的命令就是用户看到的接口。 文档与实现脱节是极易发生且很难察觉的问题 —— 实测发现手册记载的 ``hdiv sync financial --interleaved`` 在 CLI 中根本没有暴露， 而 ``hdiv ddl verify`` 的输出格式也与手册不符。 本测试把「手册承诺的接口」变成可执行断言： 只要有人删掉一个参数或改掉命令名，测试立刻失败。 """ from __future__ import annotations import pandas as pd 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", "--basic-start", "--basic-end"}), ("universe", {"-c", "--config", "--asof", "--no-persist", "--no-html"}), ("profile", {"--universe-run", "--symbols", "--asof", "--html-limit"}), ("backtest", {"-s", "--strategy", "--mode", "--start", "--end", "--universe-run", "--allow-lookahead-universe"}), ("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" assert args.allow_lookahead_universe is False, "默认不得放行未来函数" args2 = parser.parse_args(["backtest"]) assert args2.universe_run is None def test_backtest_lookahead_override_flag(parser) -> None: """放行未来函数必须是一个显式开关，不能是默认行为。""" a = parser.parse_args( ["backtest", "--universe-run", "abc123", "--allow-lookahead-universe"] ) assert a.allow_lookahead_universe is True @pytest.mark.db def test_frozen_universe_is_used_when_run_id_given() -> None: """显式放行时，冻结股票池在各调仓日必须完全相同。""" 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, allow_lookahead_universe=True, ) 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), "冻结股票池在各调仓日必须完全相同" @pytest.mark.db def test_future_universe_is_rejected_by_default() -> None: """回归：股票池 asof 晚于回测起点 = 未来函数，必须默认拒绝。 早期实现允许用 2025 年选出的股票池跑 2015 起的单次回测，2015-01-06 就按那份事后名单成交 —— 与 walk-forward 已明确拒绝的做法自相矛盾。 """ from datetime import date from hdiv.backtest.engine import BacktestEngine from hdiv.core.config import load_config from hdiv.core.errors import HdivError from hdiv.data import db db.load_dotenv_once() try: cfg = load_config("datasource") df = db.read_sql( "SELECT run_id, asof_date FROM hd_universe_run WHERE deleted_at IS NULL " "ORDER BY asof_date DESC LIMIT 1", cfg=cfg, ) if df.empty: pytest.skip("没有筛选记录") rid = df["run_id"].iloc[0] uasof = pd.to_datetime(df["asof_date"].iloc[0]).date() except Exception as exc: pytest.skip(f"数据库不可用：{exc}") eng = BacktestEngine.from_strategy( "config/strategy/high_dividend_v1.yml", universe_run_id=rid ) early = date(uasof.year - 5, 1, 1) with pytest.raises(HdivError) as ei: eng._prepare(eng.repo.trading_days(early, date(uasof.year - 5, 3, 31)), verbose=False) msg = str(ei.value) assert "晚于回测起点" in msg assert "--allow-lookahead-universe" in msg, "错误信息必须给出放行开关" # 起点晚于股票池 asof 时是合法的：此时股票池属于「事前信息」 later = date(uasof.year + 1, 1, 2) eng2 = BacktestEngine.from_strategy( "config/strategy/high_dividend_v1.yml", universe_run_id=rid ) days = eng2.repo.trading_days(later, date(later.year, 3, 31)) if len(days) >= 2: ctx = eng2._prepare(days, verbose=False) assert ctx["universe_by_refresh"], "asof 不晚于起点时应正常使用该股票池" assert eng2.lookahead_universe_note is None 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" def test_walkforward_rejects_universe_run() -> None: """回归：--universe-run 与 walk-forward 时序不兼容，必须明确拒绝。 股票池自带 asof（如 2025-01-21），而 walk-forward 窗口从 2015 年就开始训练； 把未来时点选出的股票池套到更早的窗口上等于用未来信息选股。 早期实现没有该参数，于是 --universe-run 被**静默忽略** —— 用户以为按自己的股票池跑了，实际跑的是逐窗口自筛选。 """ import os import subprocess import sys r = subprocess.run( [sys.executable, "-m", "hdiv", "backtest", "--universe-run", "whatever", "--mode", "walkforward"], capture_output=True, text=True, env={**os.environ, "PYTHONPATH": "src"}, ) assert r.returncode == 1 out = r.stdout + r.stderr assert "不支持 --universe-run" in out, out[:400] assert "未来" in out, "应说明原因（未来函数）" assert "Traceback" not in out def test_walkforward_runner_has_no_universe_param() -> None: """再确认一层：runner 本身不接受冻结股票池，避免日后被误加回去。""" import inspect from hdiv.backtest.walk_forward import WalkForwardRunner sig = inspect.signature(WalkForwardRunner.__init__) assert "universe_run_id" not in sig.parameters, ( "WalkForwardRunner 不应接受 universe_run_id —— " "冻结股票池与 walk-forward 的时序纪律冲突" ) def test_no_future_warning_on_selector() -> None: """回归：object 列的 fillna 曾触发 pandas Downcasting FutureWarning。 用 -W error::FutureWarning 跑一遍筛选路径，确保不再产生该警告 （pandas 未来版本会改变行为，届时结果可能静默变化）。 """ import os import subprocess import sys r = subprocess.run( [sys.executable, "-W", "error::FutureWarning", "-m", "hdiv", "universe", "--asof", "2025-03-21", "--no-persist", "--no-html"], capture_output=True, text=True, env={**os.environ, "PYTHONPATH": "src"}, timeout=600, ) assert "FutureWarning" not in (r.stdout + r.stderr), \ f"仍存在 FutureWarning：{(r.stdout + r.stderr)[-400:]}"