Skip to content

fix(risk): enforce paired scenario envelopes #295

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Aug 27, 2026
Merged

fix(risk): enforce paired scenario envelopes #295

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 30 additions & 4 deletions docs/qsl_long_horizon_risk_composer_v1.zh-CN.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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`。缺少偏好即失败，不会静默选择“均衡”或任何默认档位。

Expand All @@ -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` 或覆盖已签风险政策。后续接线必须为每个候选建立新的、可回放的风险政策版本，并遵守：
Expand Down
72 changes: 62 additions & 10 deletions python/scripts/long_horizon_risk_composer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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"]),
Expand All @@ -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": "",
}
Expand Down
57 changes: 57 additions & 0 deletions python/tests/test_long_horizon_risk_composer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down