Files
sanguo_vnpy_v2/tests/portfolio/test_promotion_gate.py
T
claude_dev d73bbe6841 feat(portfolio): 晋级闸门纯函数——默认阈值+回测/影子两套判定+DB覆盖汇合点 [vps] [no-doc]
spec §4.5 决议 L 六件之 2:一处定义三处调用;frozen dataclass;压线=过;
thresholds_from_overrides 支撑「代码默认+每机运行值」两级阈值。

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-24 20:53:10 +08:00

126 lines
4.7 KiB
Python

# tests/portfolio/test_promotion_gate.py
"""gate 纯函数 TDD——spec §4.5 决议 L 六件之 2(默认值逐字抄 spec).
回测 gate=夏普>=1.0+最大回撤<=30%;影子毕业 gate=4 周+TE 年化<=5%+fillRate>=95%
+周均收益偏差<=3%。一处定义三处调用(转移闸门/梯子页/回测报告)。
"""
import pytest
from sanguo_portfolio.promotion_gate import (
DEFAULT_THRESHOLDS,
THRESHOLD_KEYS,
GateThresholds,
evaluate_backtest_gate,
evaluate_shadow_gate,
thresholds_from_overrides,
)
def test_default_thresholds_match_spec():
t = DEFAULT_THRESHOLDS
assert t.backtest_sharpe_min == 1.0
assert t.backtest_max_dd == 0.30
assert t.shadow_weeks_min == 4
assert t.shadow_te_annual_max == 0.05
assert t.shadow_fill_rate_min == 0.95
assert t.shadow_weekly_dev_max == 0.03
assert THRESHOLD_KEYS == ("backtest_sharpe_min", "backtest_max_dd",
"shadow_weeks_min", "shadow_te_annual_max",
"shadow_fill_rate_min", "shadow_weekly_dev_max")
def test_backtest_gate_pass_and_fail():
ok = evaluate_backtest_gate(sharpe=1.4, max_dd=0.22)
assert ok.passed and len(ok.checks) == 2
bad_sharpe = evaluate_backtest_gate(sharpe=0.9, max_dd=0.22)
assert not bad_sharpe.passed
bad_dd = evaluate_backtest_gate(sharpe=1.4, max_dd=0.35)
assert not bad_dd.passed
def test_shadow_gate_all_four_checks():
ok = evaluate_shadow_gate(weeks=4, te_annual=0.04, fill_rate=0.97,
weekly_dev=0.01)
assert ok.passed and len(ok.checks) == 4
short = evaluate_shadow_gate(weeks=3, te_annual=0.04, fill_rate=0.97,
weekly_dev=0.01)
assert not short.passed
te = evaluate_shadow_gate(weeks=4, te_annual=0.06, fill_rate=0.97,
weekly_dev=0.01)
assert not te.passed
fill = evaluate_shadow_gate(weeks=4, te_annual=0.04, fill_rate=0.90,
weekly_dev=0.01)
assert not fill.passed
dev = evaluate_shadow_gate(weeks=4, te_annual=0.04, fill_rate=0.97,
weekly_dev=0.05)
assert not dev.passed
def test_boundary_is_inclusive():
# 恰好压线=过(>=1.0 / <=0.30 / >=4 / <=0.05 / >=0.95 / <=0.03)
assert evaluate_backtest_gate(sharpe=1.0, max_dd=0.30).passed
assert evaluate_shadow_gate(weeks=4, te_annual=0.05, fill_rate=0.95,
weekly_dev=0.03).passed
def test_to_dict_roundtrip_for_snapshot():
r = evaluate_shadow_gate(weeks=4, te_annual=0.04, fill_rate=0.97,
weekly_dev=0.01)
d = r.to_dict()
assert d["passed"] is True and d["checks"][0]["key"] == "shadow_weeks_min"
def test_thresholds_from_overrides_and_unknown_key():
t = thresholds_from_overrides({"backtest_sharpe_min": 1.2})
assert t.backtest_sharpe_min == 1.2 and t.shadow_weeks_min == 4
with pytest.raises(ValueError, match="未知阈值键"):
thresholds_from_overrides({"nope": 1.0})
def test_frozen():
with pytest.raises(Exception):
DEFAULT_THRESHOLDS.backtest_sharpe_min = 2.0 # type: ignore[misc]
def test_cli_backtest_explicit_flags(capsys):
from sanguo_portfolio.promotion_gate import main
rc = main(["backtest", "--sharpe", "1.4", "--max-dd", "0.22"])
assert rc == 0
out = capsys.readouterr().out
assert '"passed": true' in out
def test_cli_backtest_task_id_reads_stats(tmp_path, capsys):
import json
import sqlite3
p = str(tmp_path / "bt.db")
conn = sqlite3.connect(p)
conn.execute("CREATE TABLE backtest_stats (id INTEGER PRIMARY KEY "
"AUTOINCREMENT, task_id TEXT, statistics TEXT, equity_path TEXT)")
conn.execute("INSERT INTO backtest_stats(task_id, statistics) VALUES(?,?)",
("bt1", json.dumps({"sharpe_ratio": 1.05, "max_drawdown": 0.28})))
conn.commit()
conn.close()
from sanguo_portfolio.promotion_gate import main
rc = main(["backtest", "--task-id", "bt1", "--db", p])
assert rc == 0
out = capsys.readouterr().out
assert '"passed": true' in out and '"value": 1.05' in out
def test_cli_task_id_missing_keys_errors(tmp_path, capsys):
import json
import sqlite3
p = str(tmp_path / "bt.db")
conn = sqlite3.connect(p)
conn.execute("CREATE TABLE backtest_stats (id INTEGER PRIMARY KEY "
"AUTOINCREMENT, task_id TEXT, statistics TEXT, equity_path TEXT)")
conn.execute("INSERT INTO backtest_stats(task_id, statistics) VALUES(?,?)",
("bt2", json.dumps({"total_return": 0.2})))
conn.commit()
conn.close()
from sanguo_portfolio.promotion_gate import main
rc = main(["backtest", "--task-id", "bt2", "--db", p])
assert rc == 1
assert "显式" in capsys.readouterr().err