"""配置加载与严格校验测试（对应 development-plan.md P3 验收：非法配置 100% 被拒）。""" from __future__ import annotations import copy import pytest import yaml from hdiv.core.config import ( config_hash, load_all, load_config, resolve_strategy_path, ) from hdiv.core.errors import ConfigError, ConfigNotFound from hdiv.core.paths import config_dir # --------------------------------------------------------------------------- # 正向：全部配置文件可加载 # --------------------------------------------------------------------------- @pytest.mark.parametrize( "name", ["datasource", "universe", "profile", "cost", "backtest", "report"] ) def test_all_configs_load(name: str) -> None: cfg = load_config(name) assert cfg is not None assert config_hash(cfg) def test_load_all() -> None: cfgs = load_all() assert set(cfgs) == {"datasource", "universe", "profile", "cost", "backtest", "report"} def test_strategy_config_loads() -> None: s = load_config("strategy:high_dividend_v1.yml") assert s.strategy.id == "HD_MR_V1" assert s.entry.yield_percentile == 75 assert s.exit.yield_percentile == 25 # 参数扁平化用于 hd_strategy_param pm = s.param_map() assert pm["strategy.id"] == "HD_MR_V1" assert pm["entry.yield_percentile"] == 75 assert len(pm) > 20 def test_strategy_path_resolution_variants() -> None: names = [ "high_dividend_v1.yml", "strategy/high_dividend_v1.yml", "config/strategy/high_dividend_v1.yml", ] paths = {resolve_strategy_path(n) for n in names} assert len(paths) == 1 assert paths.pop().is_file() def test_resolve_strategy_path_missing() -> None: with pytest.raises(ConfigNotFound): resolve_strategy_path("不存在.yml") # --------------------------------------------------------------------------- # 反向：拼写错误必须报错（不静默取默认值） # --------------------------------------------------------------------------- def _load_modified(name: str, mutate) -> None: """把配置改坏后写入临时文件并加载，断言抛错。""" raw = yaml.safe_load((config_dir() / f"{name}.yml").read_text(encoding="utf-8")) mutate(raw) import tempfile from pathlib import Path with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as fh: yaml.safe_dump(raw, fh, allow_unicode=True) tmp = Path(fh.name) try: with pytest.raises(ConfigError): load_config(name, path=tmp) finally: tmp.unlink(missing_ok=True) def test_unknown_field_rejected() -> None: _load_modified("universe", lambda r: r["market"].update({"min_market_capp": 1})) def test_wrong_type_rejected() -> None: _load_modified("universe", lambda r: r["market"].update({"min_listing_years": "十年"})) def test_unknown_top_level_section_rejected() -> None: _load_modified("universe", lambda r: r.update({"bogus_section": {"a": 1}})) def test_bad_enum_rejected() -> None: _load_modified("cost", lambda r: r["slippage"].update({"mode": "percentage"})) def test_dividend_years_consistency_rejected() -> None: def mutate(r: dict) -> None: r["dividend"]["min_continuous_years"] = 10 r["dividend"]["window_years"] = 6 _load_modified("universe", mutate) def test_negative_capital_rejected() -> None: _load_modified("backtest", lambda r: r["capital"].update({"initial": -1})) def test_period_order_rejected() -> None: _load_modified("backtest", lambda r: r["period"].update({"end": "2010-01-01"})) def test_freeze_params_disabled_rejected() -> None: """plan.md §25：测试阶段禁止重新调参 —— 关闭该开关必须被拒绝。""" _load_modified( "backtest", lambda r: r["walk_forward"].update({"freeze_params_in_test": False}) ) def test_duplicate_benchmark_rejected() -> None: def mutate(r: dict) -> None: r["benchmark"] = [ {"code": "000300.SH", "name": "沪深300"}, {"code": "000300.SH", "name": "重复"}, ] _load_modified("backtest", mutate) def test_profile_percentiles_must_be_sorted() -> None: _load_modified("profile", lambda r: r.update({"percentiles": [90, 10, 50]})) def test_profile_duplicate_metric_rejected() -> None: _load_modified( "profile", lambda r: r["metrics"].update({"valuation": ["pb", "pb"]}) ) def test_composite_weights_must_sum_to_one() -> None: def mutate(r: dict) -> None: r["safety_margin"]["mode"] = "composite" r["safety_margin"]["weights"] = {"dividend_yield": 0.5, "valuation": 0.2} _load_modified("profile", mutate) def test_datasource_readonly_writeoverlap_rejected() -> None: def mutate(r: dict) -> None: r["database"]["allow_write_tables"] = ["stock"] _load_modified("datasource", mutate) # --------------------------------------------------------------------------- # 策略语义校验 # --------------------------------------------------------------------------- def _load_strategy_mutated(mutate) -> None: raw = yaml.safe_load( (config_dir() / "strategy" / "high_dividend_v1.yml").read_text(encoding="utf-8") ) mutate(raw) import tempfile from pathlib import Path with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as fh: yaml.safe_dump(raw, fh, allow_unicode=True) tmp = Path(fh.name) try: with pytest.raises(ConfigError): load_config(f"strategy:{tmp}") finally: tmp.unlink(missing_ok=True) def test_strategy_entry_must_exceed_exit() -> None: _load_strategy_mutated(lambda r: r["entry"].update({"yield_percentile": 20})) def test_strategy_entry_percentile_range() -> None: _load_strategy_mutated(lambda r: r["entry"].update({"yield_percentile": 150})) def test_strategy_scale_in_must_be_ascending() -> None: def mutate(r: dict) -> None: r["entry"]["scale_in"] = [ {"percentile": 80, "weight": 0.25}, {"percentile": 75, "weight": 0.50}, ] _load_strategy_mutated(mutate) def test_strategy_scale_out_must_descend() -> None: def mutate(r: dict) -> None: r["exit"]["scale_out"] = [ {"percentile": 25, "weight": 0.50}, {"percentile": 50, "weight": 0.00}, ] _load_strategy_mutated(mutate) def test_strategy_scale_out_last_must_be_zero() -> None: def mutate(r: dict) -> None: r["exit"]["scale_out"] = [ {"percentile": 50, "weight": 0.50}, {"percentile": 25, "weight": 0.30}, ] _load_strategy_mutated(mutate) def test_strategy_position_bounds() -> None: _load_strategy_mutated(lambda r: r["position"].update({"max_position": 1.5})) def test_strategy_bad_status() -> None: _load_strategy_mutated(lambda r: r["strategy"].update({"status": "RUNNING"})) # --------------------------------------------------------------------------- # 实时画像闸门（profile_gate）配置校验 # # 这些必须挡在配置期：写错指标名若拖到运行时，只会表现为 # 「无法验证 → 保守不买」，即策略悄悄再也不交易，极难定位。 # --------------------------------------------------------------------------- def test_profile_gate_unknown_metric_rejected() -> None: def mutate(r: dict) -> None: r["entry"]["profile_gate"]["rules"] = [ {"metric": "not_a_metric", "op": ">=", "value": 1.0} ] _load_strategy_mutated(mutate) def test_profile_gate_percentile_on_scalar_metric_rejected() -> None: """标量指标没有历史分位，不能用 current_percentile。""" def mutate(r: dict) -> None: r["entry"]["profile_gate"]["rules"] = [ {"metric": "payout_ratio", "stat": "current_percentile", "op": "<=", "value": 1.0} ] _load_strategy_mutated(mutate) def test_profile_gate_enabled_without_rules_rejected() -> None: def mutate(r: dict) -> None: r["entry"]["profile_gate"] = {"enabled": True, "rules": []} _load_strategy_mutated(mutate) def test_profile_gate_bad_operator_rejected() -> None: def mutate(r: dict) -> None: r["entry"]["profile_gate"]["rules"] = [ {"metric": "dv_yield", "op": "~=", "value": 1.0} ] _load_strategy_mutated(mutate) def test_profile_gate_coverage_bounds() -> None: """min_window_coverage 必须落在 [0,1]：1.0 = 必须完整覆盖名义窗口。""" def mutate(r: dict) -> None: r["entry"]["profile_gate"]["min_window_coverage"] = 1.5 _load_strategy_mutated(mutate) def test_profile_gate_disabled_without_rules_is_allowed() -> None: """默认（未启用、无规则）必须能正常加载 —— 否则所有历史配置都会失效。""" raw = yaml.safe_load( (config_dir() / "strategy" / "high_dividend_v1.yml").read_text(encoding="utf-8") ) raw["entry"]["profile_gate"] = {"enabled": False, "rules": []} import tempfile from pathlib import Path with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as fh: yaml.safe_dump(raw, fh, allow_unicode=True) tmp = Path(fh.name) try: cfg = load_config(f"strategy:{tmp}") assert cfg.entry.profile_gate.enabled is False finally: tmp.unlink(missing_ok=True) # --------------------------------------------------------------------------- # 可复现性：config_hash 稳定性 # --------------------------------------------------------------------------- def test_config_hash_is_stable_and_sensitive() -> None: a = load_config("universe") b = load_config("universe") assert config_hash(a) == config_hash(b) raw = yaml.safe_load((config_dir() / "universe.yml").read_text(encoding="utf-8")) changed = copy.deepcopy(raw) changed["dividend"]["min_dividend_yield"] = 0.035 assert config_hash(raw) != config_hash(changed) def test_config_hash_ignores_key_order() -> None: raw = yaml.safe_load((config_dir() / "universe.yml").read_text(encoding="utf-8")) reordered = {k: raw[k] for k in reversed(list(raw))} assert config_hash(raw) == config_hash(reordered) def test_missing_config_raises() -> None: with pytest.raises(ConfigNotFound): load_config("universe", path="/nonexistent/nope.yml") # --------------------------------------------------------------------------- # 回测层面的行业排除清单（backtest.yml: universe_exclusions） # --------------------------------------------------------------------------- def test_backtest_config_excludes_real_estate_industries() -> None: """当前生效配置必须真的在排除房地产，且四个口径齐全。 数据库里没有「房地产业」这个取值，它被拆成四个行业名； 少写一条就少排一类（例如只写「全国地产」会漏掉全部区域地产公司）。 """ bt = load_config("backtest") excl = set(bt.universe_exclusions.industries) assert {"全国地产", "区域地产", "房产服务", "园区开发"} <= excl def test_resolved_universe_applies_backtest_exclusions() -> None: from hdiv.core.config import BacktestConfig, UniverseConfig bt = BacktestConfig.model_validate( { "period": {"start": "2015-01-01", "end": "latest"}, "universe_exclusions": {"industries": ["全国地产", "区域地产"]}, } ) uni = UniverseConfig.model_validate({"name": "t"}) merged = bt.resolved_universe(uni) assert merged.industry_exclusions == ["全国地产", "区域地产"] assert uni.industry_exclusions == [], "不得就地修改传入的筛选配置" def test_resolved_universe_is_a_union_not_an_override() -> None: """universe.yml 自带的排除项不能被 backtest.yml 顶掉（只做减法）。""" from hdiv.core.config import BacktestConfig, UniverseConfig bt = BacktestConfig.model_validate( { "period": {"start": "2015-01-01", "end": "latest"}, "universe_exclusions": {"industries": ["房产服务"]}, } ) uni = UniverseConfig.model_validate( {"name": "t", "industry_exclusions": ["园区开发"]} ) merged = bt.resolved_universe(uni) assert merged.industry_exclusions == ["园区开发", "房产服务"] def test_resolved_universe_without_exclusions_returns_same_object() -> None: """不配排除清单 → 原样返回，保证「不配 = 行为与改动前一致」。""" from hdiv.core.config import BacktestConfig, UniverseConfig bt = BacktestConfig.model_validate({"period": {"start": "2015-01-01"}}) uni = UniverseConfig.model_validate({"name": "t"}) assert bt.resolved_universe(uni) is uni def test_duplicate_exclusion_industry_rejected() -> None: raw = yaml.safe_load((config_dir() / "backtest.yml").read_text(encoding="utf-8")) raw["universe_exclusions"] = {"industries": ["全国地产", "全国地产"]} with pytest.raises(ConfigError): _validate_tmp("backtest", raw) def test_unknown_exclusion_field_rejected() -> None: """字段名写错必须报错，不能静默忽略（本项目的一贯纪律）。""" raw = yaml.safe_load((config_dir() / "backtest.yml").read_text(encoding="utf-8")) raw["universe_exclusions"] = {"industry": ["全国地产"]} with pytest.raises(ConfigError): _validate_tmp("backtest", raw) def _validate_tmp(name: str, raw: dict) -> object: import tempfile from pathlib import Path with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as fh: yaml.safe_dump(raw, fh, allow_unicode=True) tmp = Path(fh.name) try: return load_config(name, path=tmp) finally: tmp.unlink(missing_ok=True) def test_unknown_config_name_raises() -> None: with pytest.raises(ConfigError): load_config("no_such_config")