simon/ggx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始提交：高股息策略研究与回测系统

Browse Source 
从 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。
This commit is contained in:
simon
2026-10-03 13:54:56 +08:00
commit fce725e13c
127 changed files with 28001 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_config.py
+261
View File
@@ -0,0 +1,261 @@
"""配置加载与严格校验测试（对应 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"}))
# ---------------------------------------------------------------------------
# 可复现性：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")
def test_unknown_config_name_raises() -> None:
with pytest.raises(ConfigError):
load_config("no_such_config")