From af5bcf18f6ee736e8500c32febf9e3835fb3a884 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 27 Aug 2026 18:17:25 +0800 Subject: [PATCH] fix(risk): enforce paired scenario envelopes Co-Authored-By: Codex --- ...qsl_long_horizon_risk_composer_v1.zh-CN.md | 34 +++++++-- python/scripts/long_horizon_risk_composer.py | 72 ++++++++++++++++--- .../tests/test_long_horizon_risk_composer.py | 57 +++++++++++++++ 3 files changed, 149 insertions(+), 14 deletions(-) diff --git a/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md b/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md index cc45515..6860c5e 100644 --- a/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md +++ b/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md @@ -12,13 +12,15 @@ 人工只选择三个简单、重要的偏好之一： -| 偏好 | 系统使用的无杠杆基准最大回撤倍数 | 含义 | +| 偏好 | v1 兼容期的无杠杆基准最大回撤天花板 | 含义 | | --- | ---: | --- | | `CAPITAL_PRESERVATION` | 1.00 倍 | 风险尺度在所有合格情景的最大回撤不得超过基准的最大回撤。 | | `BALANCED_COMPOUNDING` | 1.25 倍 | 允许有限的额外回撤，换取经多情景验证的更高几何增长。 | | `GROWTH_COMPOUNDING` | 1.50 倍 | 仅在仍通过多情景长期复利门槛时允许更高风险尺度。 | -这三个倍数是透明、版本化的偏好模板，不是模型从历史数据“发现”的真理。系统计算的内容是每个候选在每个尺度下的实际净成本路径、最大回撤、相对基准回撤、水下持续期和每 session 对数几何增长；它不会把一次历史最优结果伪装成未来保证。 +这三个倍数是透明、版本化的偏好模板，不是模型从历史数据“发现”的真理，也不是长期复利的普适最优比例。它们只是在 v1 过渡期给出不可放宽的相对回撤天花板。系统计算的内容是每个候选在每个尺度下的实际净成本路径、最大回撤、相对基准回撤、水下持续期和每 session 对数几何增长；它不会把一次历史最优结果伪装成未来保证。 + +因此，三个名称是跨策略、跨平台的账户/组合层选项，不能由策略代码、平台适配器或插件自行选择或提高。当前所有者只选一个名称；系统只可据新证据自动降尺度或 `PARKED`，不得静默切换到更激进档位、增加尺度或授予 live 权限。 命令行也遵守这条分工：已有 owner-bound 输入可用 `--input`；私有观察件必须同时给出 `--observation` 与 `--risk-preference`。缺少偏好即失败，不会静默选择“均衡”或任何默认档位。 @@ -41,11 +43,12 @@ - 各情景对数几何增长的下中位数； - 最坏绝对与相对基准最大回撤； - 最坏基准最大回撤与最长水下期； -- 至少三分之二情景为正增长，且策略最坏回撤不超过所选偏好的基准回撤倍数时，才标记该尺度合格。 +- 每条策略路径只与**同一条、同一窗口的配对基准路径**比较回撤天花板；某个极端压力基准的回撤不得放宽 walk-forward 或 bootstrap 的限制； +- 先在 `WALK_FORWARD`、`BOOTSTRAP`、`STRESS` 各自内部计算下中位增长与正增长覆盖率，再把三个证据家族等权比较。至少两个家族的正增长覆盖率达到三分之二，且每条路径都满足其配对回撤天花板，才标记该尺度合格。 每个情景明确携带 `session_count`，且必须恰好比收益率数组多一个起点：例如 252 个 XNYS 观测日对应 251 个相邻日收益率。长度门检查的是已签名的观测日数量，避免把完整的一年前瞻窗口误判为不足，也避免把收益率数组伪装成更多交易日。 -在所有合格尺度中，选择下中位数对数几何增长最高者；同分时选择更低风险尺度。输出的 `recommended_max_drawdown_bps` 是由冻结的基准路径和偏好模板计算出的候选上限，不是订单阈值或已启用政策。 +在所有合格尺度中，选择三个等权证据家族的下中位数对数几何增长最高者；同分时选择更低风险尺度。输出的 `recommended_max_drawdown_bps` 是该选择在冻结 P3 配对路径中实际观察到的最坏策略回撤，不是订单阈值、单一情景上限或已启用政策。合格性由每条路径各自的上限决定，不能把它们错误压缩为一个较宽松的全局数字。 ## SOXL 与 TQQQ 的使用方式 @@ -53,6 +56,29 @@ SOXL 必须传入 SOXX 的无杠杆路径，TQQQ 必须传入 QQQ 的无杠杆 这样系统衡量的是“杠杆策略相对基准额外承担了多少回撤和修复期”，而不是只看 SOXL/TQQQ 自身的历史收益。 组合或插件也必须各自提供同一截止日、同一净成本口径的冻结基准路径；不能复用别的候选的推荐结果。 +## 向所有策略、组合和插件扩展 + +现有 v1 内核已接受 `individual`、`combo`、`plugin` 三类候选；通用的是风险偏好、证据边界和计算器，而不是某个 SOXL 参数。每个新的 P3 生产器必须依照下列规则提供自己的私有观察件，缺少任何必要证据就保持 `PARKED`，不能借用 SOXL、TQQQ 或另一平台的结论： + +| 候选类型 | 必须提供的配对基准与证据 | 不能做的事 | +| --- | --- | --- | +| 杠杆 ETF / 定向现货 | 对应无杠杆总收益基准、日复位/融资/跳空压力路径 | 用标称杠杆或另一只 ETF 的历史代替配对路径 | +| 轮动或多资产组合 | 预先登记的政策混合基准、组合级净收益、再平衡与相关性证据 | 把单策略回撤或收益简单相加 | +| DCA / 有外部现金流的策略 | 现金流匹配基准、时间加权收益和净成本路径 | 让充值、提款或账户余额改变风险尺度判断 | +| 期权、收益或保证金策略 | 基准路径以及跳空、波动率、指派、保证金和流动性压力证据 | 因历史最大回撤低就忽略未观测尾部风险 | +| 市场中性 / 绝对收益 | 现金或短债加预登记目标收益基准、杠杆和流动性压力证据 | 机械套用权益无杠杆指数的回撤倍数 | +| 插件 | 只有改变收益路径的“策略 + 插件 bundle”才可形成新候选，并重新完成 P1/P2/P3 | 信号/观察插件自行提交风险建议、改变档位或绕过 P3 | + +后续 `risk_observation.v2` 的扩展必须以新 schema 并行引入，不能向 v1 偷加字段或破坏既有证据哈希。v2 必须增加以下冻结声明后，才允许把更丰富的风险能力接入统一 Composer： + +1. `risk_capability`：候选是否为独立收益候选、收益尺度能否线性缩放，还是必须在每个尺度重放； +2. `benchmark_policy`：基准类别、交易日历、币种、总收益与现金流口径，组合还要声明政策权重； +3. `risk_factor_coverage`：杠杆、融资、保证金、集中度、流动性、跳空及相关性中哪些已被 P3 覆盖； +4. `scenario_family`：保持 walk-forward、重采样、压力证据可审计，并允许每个家族有自己的最小覆盖要求； +5. `portfolio_scope`：单候选或组合。组合必须在组合层重新计算边际风险贡献与总预算。 + +v2 的选择语义也应由版本化、与候选 P3 隔离的 profile 实现：资本保护选择基准风险内的稳健增长解；均衡复利选择有效前沿折点；增长复利选择满足生存、流动性与恢复期硬门的稳健对数增长解。不得拿同一候选的历史结果反向拟合这三个 profile。 + ## 不能自动做的事 Composer 只产生建议，不能直接调用 `deterministic_risk_gate` 或覆盖已签风险政策。后续接线必须为每个候选建立新的、可回放的风险政策版本，并遵守： diff --git a/python/scripts/long_horizon_risk_composer.py b/python/scripts/long_horizon_risk_composer.py index c164e36..8415f44 100644 --- a/python/scripts/long_horizon_risk_composer.py +++ b/python/scripts/long_horizon_risk_composer.py @@ -435,6 +435,49 @@ def _path_metrics( } +def _profile_drawdown_cap_bps(*, benchmark_drawdown_bps: int, benchmark_multiple_bps: int) -> int: + """Return the profile envelope for one *paired* benchmark path. + + A stress path must never relax the envelope applied to an unrelated + walk-forward or bootstrap path. The calculation is therefore deliberately + per-scenario; callers must not aggregate benchmark drawdowns before they + call this helper. + """ + return min(10_000, (benchmark_drawdown_bps * benchmark_multiple_bps + 9_999) // 10_000) + + +def _family_growth_summary( + paths: Sequence[Mapping[str, Any]], metrics: Sequence[Mapping[str, int]] +) -> tuple[int, bool]: + """Summarize growth with equal evidence-family influence. + + A producer may emit more bootstrap paths than historical paths. Counting + every path in one global pool would let that implementation detail outweigh + walk-forward or stress evidence. We take the lower median inside each + required family, then the lower median of the three family summaries. + + A family is growth-positive only when at least two thirds of *its* paths + are positive. At least two of the three independent evidence families + must meet that test. This retains the prior two-thirds policy while making + the policy invariant to how many valid bootstrap replicas were supplied. + """ + family_log_growth: dict[str, list[int]] = {kind: [] for kind in _SCENARIO_KINDS} + for path, metric in zip(paths, metrics, strict=True): + family_log_growth[path["scenario_kind"]].append(metric["log_growth_ppm"]) + + # Coverage is checked before this helper is called, so every list is + # non-empty. Keeping the guard makes the pure helper fail closed if reused. + if any(not values for values in family_log_growth.values()): + _fail("scenario evidence family coverage is incomplete") + + family_medians = [median_low(values) for values in family_log_growth.values()] + positive_families = sum( + sum(value > 0 for value in values) * 3 >= len(values) * 2 + for values in family_log_growth.values() + ) + return median_low(family_medians), positive_families * 3 >= len(_SCENARIO_KINDS) * 2 + + def _parked_recommendation(value: Mapping[str, Any], reasons: list[str]) -> dict[str, Any]: recommendation: dict[str, Any] = { "schema": RISK_COMPOSER_RECOMMENDATION_SCHEMA_ID, @@ -482,19 +525,28 @@ def compose_long_horizon_risk_recommendation(value: Any) -> dict[str, Any]: for path in paths ] worst_benchmark_drawdown = max(item["benchmark_drawdown_bps"] for item in metrics) - profile_drawdown_cap = (worst_benchmark_drawdown * benchmark_multiple_bps + 9_999) // 10_000 + profile_drawdown_caps = [ + _profile_drawdown_cap_bps( + benchmark_drawdown_bps=item["benchmark_drawdown_bps"], + benchmark_multiple_bps=benchmark_multiple_bps, + ) + for item in metrics + ] positive_growth = sum(item["log_growth_ppm"] > 0 for item in metrics) - scenario_count = len(metrics) + family_median_log_growth, families_meet_growth_requirement = _family_growth_summary(paths, metrics) eligible = ( - positive_growth * 3 >= scenario_count * 2 - and max(item["max_drawdown_bps"] for item in metrics) <= profile_drawdown_cap + families_meet_growth_requirement + and all( + item["max_drawdown_bps"] <= profile_drawdown_cap + for item, profile_drawdown_cap in zip(metrics, profile_drawdown_caps, strict=True) + ) ) frontier.append( { "scale_bps": scale_bps, - "median_log_growth_ppm": median_low(item["log_growth_ppm"] for item in metrics), + "median_log_growth_ppm": family_median_log_growth, "positive_growth_scenarios": positive_growth, - "scenario_count": scenario_count, + "scenario_count": len(metrics), "worst_max_drawdown_bps": max(item["max_drawdown_bps"] for item in metrics), "worst_relative_drawdown_bps": max(item["relative_drawdown_bps"] for item in metrics), "worst_benchmark_drawdown_bps": worst_benchmark_drawdown, @@ -507,9 +559,6 @@ def compose_long_horizon_risk_recommendation(value: Any) -> dict[str, Any]: if not eligible_frontier: return _parked_recommendation(validated, ["NO_SCALE_MEETS_COMPOUNDING_AND_DRAWDOWN_CONSTRAINTS"]) chosen = max(eligible_frontier, key=lambda item: (item["median_log_growth_ppm"], -item["scale_bps"])) - maximum_drawdown = ( - chosen["worst_benchmark_drawdown_bps"] * benchmark_multiple_bps + 9_999 - ) // 10_000 recommendation: dict[str, Any] = { "schema": RISK_COMPOSER_RECOMMENDATION_SCHEMA_ID, "candidate": dict(validated["candidate"]), @@ -519,7 +568,10 @@ def compose_long_horizon_risk_recommendation(value: Any) -> dict[str, Any]: "status": "ADVISORY_RECOMMENDATION_READY", "reason_codes": [], "recommended_scale_bps": chosen["scale_bps"], - "recommended_max_drawdown_bps": maximum_drawdown, + # This is the maximum *observed* drawdown in the selected paired P3 + # paths. Eligibility has already enforced a separate profile envelope + # for every path; do not collapse those envelopes into one policy limit. + "recommended_max_drawdown_bps": chosen["worst_max_drawdown_bps"], "frontier": frontier, "recommendation_sha256": "", } diff --git a/python/tests/test_long_horizon_risk_composer.py b/python/tests/test_long_horizon_risk_composer.py index 3a17324..843c440 100644 --- a/python/tests/test_long_horizon_risk_composer.py +++ b/python/tests/test_long_horizon_risk_composer.py @@ -122,6 +122,63 @@ def test_capital_preservation_reduces_scale_without_changing_the_frozen_evidence self.assertLess(capital["recommended_scale_bps"], balanced["recommended_scale_bps"]) self.assertLess(capital["recommended_max_drawdown_bps"], balanced["recommended_max_drawdown_bps"]) + def test_a_severe_stress_benchmark_cannot_relax_a_paired_walk_forward_drawdown_limit(self): + risk_input = self._input(preference="CAPITAL_PRESERVATION") + for path in risk_input["scenario_paths"]: + path["strategy_returns_bps"] = self._returns(25, 300) + path["benchmark_returns_bps"] = self._returns( + 25, + 5_000 if path["scenario_kind"] == "STRESS" else 100, + ) + risk_input["input_sha256"] = composer.calculate_risk_composer_input_sha256(risk_input) + + recommendation = composer.compose_long_horizon_risk_recommendation(risk_input) + + # The old global-worst-benchmark calculation would have accepted this + # row because STRESS has an almost total benchmark drawdown. The + # walk-forward and bootstrap paths must each enforce their own paired + # unlevered benchmark envelope instead. + self.assertFalse(recommendation["frontier"][-1]["eligible"]) + self.assertLess(recommendation["recommended_scale_bps"], 10_000) + + def test_bootstrap_replica_count_cannot_outvote_two_negative_evidence_families(self): + risk_input = self._input() + paths: list[dict[str, object]] = [] + for kind in ("WALK_FORWARD", "STRESS"): + returns = [0] * 240 + [-10] * 12 + paths.append( + { + "scenario_id": f"family_growth_{kind.lower()}", + "scenario_kind": kind, + "session_count": 253, + "strategy_returns_bps": returns, + "benchmark_returns_bps": list(returns), + } + ) + for index in range(8): + returns = [20] * 240 + [-10] * 12 + paths.append( + { + "scenario_id": f"family_growth_bootstrap_{index + 1}", + "scenario_kind": "BOOTSTRAP", + "session_count": 253, + "strategy_returns_bps": returns, + "benchmark_returns_bps": list(returns), + } + ) + risk_input["scenario_paths"] = paths + risk_input["input_sha256"] = composer.calculate_risk_composer_input_sha256(risk_input) + + recommendation = composer.compose_long_horizon_risk_recommendation(risk_input) + + # Eight positive bootstrap replicas are one evidence family, not eight + # votes that can hide negative walk-forward and stress results. + self.assertEqual(recommendation["status"], "PARKED") + self.assertEqual( + recommendation["reason_codes"], + ["NO_SCALE_MEETS_COMPOUNDING_AND_DRAWDOWN_CONSTRAINTS"], + ) + def test_missing_long_horizon_scenario_kind_parks_instead_of_extrapolating_a_limit(self): risk_input = self._input() risk_input["scenario_paths"] = risk_input["scenario_paths"][:2]