fix(risk): enforce paired scenario envelopes

Validated by the complete repository CI suite. This remains advisory-only and does not change runtime targets, strategy parameters, policies, or execution.

Author: Pigbibi

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,18 +43,42 @@
 - 各情景对数几何增长的下中位数；
 - 最坏绝对与相对基准最大回撤；
 - 最坏基准最大回撤与最长水下期；
-- 至少三分之二情景为正增长，且策略最坏回撤不超过所选偏好的基准回撤倍数时，才标记该尺度合格。
+- 每条策略路径只与**同一条、同一窗口的配对基准路径**比较回撤天花板；某个极端压力基准的回撤不得放宽 walk-forward 或 bootstrap 的限制；
+- 先在 `WALK_FORWARD`、`BOOTSTRAP`、`STRESS` 各自内部计算下中位增长与正增长覆盖率，再把三个证据家族等权比较。至少两个家族的正增长覆盖率达到三分之二，且每条路径都满足其配对回撤天花板，才标记该尺度合格。
 
 每个情景明确携带 `session_count`，且必须恰好比收益率数组多一个起点：例如 252 个 XNYS 观测日对应 251 个相邻日收益率。长度门检查的是已签名的观测日数量，避免把完整的一年前瞻窗口误判为不足，也避免把收益率数组伪装成更多交易日。
 
-在所有合格尺度中，选择下中位数对数几何增长最高者；同分时选择更低风险尺度。输出的 `recommended_max_drawdown_bps` 是由冻结的基准路径和偏好模板计算出的候选上限，不是订单阈值或已启用政策。
+在所有合格尺度中，选择三个等权证据家族的下中位数对数几何增长最高者；同分时选择更低风险尺度。输出的 `recommended_max_drawdown_bps` 是该选择在冻结 P3 配对路径中实际观察到的最坏策略回撤，不是订单阈值、单一情景上限或已启用政策。合格性由每条路径各自的上限决定，不能把它们错误压缩为一个较宽松的全局数字。
 
 ## SOXL 与 TQQQ 的使用方式
 
 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` 或覆盖已签风险政策。后续接线必须为每个候选建立新的、可回放的风险政策版本，并遵守：

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": "",
     }

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]