From 6202818990acbdf480593c5c048c3c8ed041e389 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 27 Aug 2026 18:30:43 +0800 Subject: [PATCH] feat(risk): add generic observation v2 contract Co-Authored-By: Codex --- ...6_CURRENT_STATE_AND_DRIVER_POLICY.zh-CN.md | 2 +- ...qsl_long_horizon_risk_composer_v1.zh-CN.md | 2 + ..._long_horizon_risk_observation_v2.zh-CN.md | 93 ++++ .../scripts/long_horizon_risk_composer_v2.py | 467 ++++++++++++++++++ ...ong_horizon_risk_observation_ingress_v2.py | 109 ++++ ...ong_horizon_risk_observation_ingress_v2.py | 151 ++++++ .../test_long_horizon_risk_observation_v2.py | 173 +++++++ 7 files changed, 996 insertions(+), 1 deletion(-) create mode 100644 docs/qsl_long_horizon_risk_observation_v2.zh-CN.md create mode 100644 python/scripts/long_horizon_risk_composer_v2.py create mode 100644 python/scripts/long_horizon_risk_observation_ingress_v2.py create mode 100644 python/tests/test_long_horizon_risk_observation_ingress_v2.py create mode 100644 python/tests/test_long_horizon_risk_observation_v2.py diff --git a/docs/QSL_P0_P6_CURRENT_STATE_AND_DRIVER_POLICY.zh-CN.md b/docs/QSL_P0_P6_CURRENT_STATE_AND_DRIVER_POLICY.zh-CN.md index 790aca6..0316d9d 100644 --- a/docs/QSL_P0_P6_CURRENT_STATE_AND_DRIVER_POLICY.zh-CN.md +++ b/docs/QSL_P0_P6_CURRENT_STATE_AND_DRIVER_POLICY.zh-CN.md @@ -111,7 +111,7 @@ AI 只做监测、研究候选生成、证据验证、受限的文本诊断和 | `qsl.research_task.v1` 与控制台队列 | 已接线（只读），待首份合格真实来源快照 | AIAudit Watcher 以专用 token 向控制台发布来源摘要；来源和控制台会各自复核 SHA、revision、摘要和 no-order authority。空队列不是故障，也不能由 Issue 推断任务。 | | P5 forward observation、risk-bound admission 与 shadow receipt | 已接线，未激活 | UESP 的 forward observation、AlpacaPlatform v2 input adapter、shadow ledger、pure controller、risk-bound receipt admission 和默认 `PARKED` 单周期编排都已存在。admission 仅接受闭合风险 decision envelope；禁止/缺失/不一致时不读写收据。已合入的 GCS adapter 只接受调用方注入的 bucket client：它按不可变 cycle 精确读取输入、用 generation-match create-only 写回执，绝不列举、覆盖、删除、读取凭据或接触 broker；存储异常闭合为 `PARKED`。没有真实 bucket、运行身份、已签 active policy、已部署 runner/scheduler、broker、账户、订单或资金。 | | P4 / P5 风险控制与 policy-gate receipt 契约 | 已实现，未接线到运行 | 可离线校验受限自动运行边界，并把一次成功的 KMS 验签投影为无敏感字段的短期 receipt；没有网络、账户、订单或资金能力。 | -| 长期复利风险政策 Composer | 私有 P3 ingress 已实现；政策写入未接线 | 通用离线内核现先校验 `qsl.long_horizon_risk_observation.v1` 私有观察件，再由所有者显式选择保守/均衡/增长偏好；它不会静默选择档位。观察件绑定 candidate 与 P1/P2/P3/plugin 摘要、净成本策略路径和同周期无杠杆基准路径；它要求 walk-forward、bootstrap、stress 各至少一个完整 252-session 情景，计算尺度—几何增长—回撤—水下期前沿，缺证据即 `PARKED`。SOXL v7 P3 已有一个纯离线生产器：它使用既有滚动 OOS、15 bps 成本压力及成对 21-session 移动块 bootstrap；只有调用方给出新的受保护本地路径时才 create-only 写入，默认工作流不上传或公开路径。尚无受限共享 ingress 存储/运行身份、已写入风险政策、策略参数变更、P4/P5/P6 或 live 权限。Composer 输出不含路径、账户或订单。 | +| 长期复利风险政策 Composer | 私有 P3 ingress 已实现；v2 通用合约与 exact ingress port 已实现；政策写入未接线 | v1 内核先校验 `qsl.long_horizon_risk_observation.v1` 私有观察件，再由所有者显式选择保守/均衡/增长偏好；它不会静默选择档位。观察件绑定 candidate 与 P1/P2/P3/plugin 摘要、净成本策略路径和同周期无杠杆基准路径；它要求 walk-forward、bootstrap、stress 各至少一个完整 252-session 情景，按配对路径回撤门与等权证据家族计算尺度—几何增长—回撤—水下期前沿，缺证据即 `PARKED`。并行的 v2 将可移植风险偏好、候选风险能力、基准政策分开绑定，并通过只读、exact-object ingress 接收私有工件：当前只对单候选、线性净收益、无现金流、无杠杆基准口径出建议；组合、现金流、非线性与混合基准均闭合为有原因码的 `PARKED`，等待专用 Composer。SOXL v7 P3 已有一个纯离线生产器：它使用既有滚动 OOS、15 bps 成本压力及成对 21-session 移动块 bootstrap；只有调用方给出新的受保护本地路径时才 create-only 写入，默认工作流不上传或公开路径。尚无受限共享 ingress 存储/运行身份、已写入风险政策、策略参数变更、P4/P5/P6 或 live 权限。Composer 输出不含路径、账户或订单。 | | P4 执行与 P5 实际调度/回执持久化 | 未实现 | 无 paper adapter、已签 active policy、已签发的运行 receipt、真实受限存储、运行身份、已部署 runner/scheduler 或真实日更 shadow receipt。P5 的代码级 GCS adapter 不创建 bucket、配置、调度或任何运行权限。 | | P6 | 未实现 | 无 live、账户、订单或资金任务。 | | `QuantStrategyLifecycle` 本机目录 | 退役/孤立 | 没有对应的 GitHub 主线仓；其中 autopilot/auto-approve 描述不得作为当前能力或设计依据。 | 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 6860c5e..a83b591 100644 --- a/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md +++ b/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md @@ -87,3 +87,5 @@ Composer 只产生建议，不能直接调用 `deterministic_risk_gate` 或覆 2. 任何扩大风险尺度、放宽回撤倍数、变更基准或进入 P4/P5/P6，都必须产生新的 P1/P2/P3 绑定与独立 policy-gate receipt； 3. P6 live 始终保留所有者明确决定，即使建议结果通过； 4. 输出不进入 AI 或控制台的原始数据上下文，只可发布脱敏摘要和 digest。 + +更丰富的候选类型使用并行的 [通用长期风险观察件 V2](qsl_long_horizon_risk_observation_v2.zh-CN.md)。v2 将可移植风险偏好、风险能力和基准政策分开绑定；在专用数学实现前，组合、现金流和非线性候选会闭合为 `PARKED`，不会被伪装为 v1 的线性收益路径。 diff --git a/docs/qsl_long_horizon_risk_observation_v2.zh-CN.md b/docs/qsl_long_horizon_risk_observation_v2.zh-CN.md new file mode 100644 index 0000000..e8d9319 --- /dev/null +++ b/docs/qsl_long_horizon_risk_observation_v2.zh-CN.md @@ -0,0 +1,93 @@ +# QSL 通用长期风险观察件 V2 + +> 状态：`CORE_AND_EXACT_INGRESS_PORT_IMPLEMENTED_NO_RUNTIME_OR_POLICY_WRITE` + +`python/scripts/long_horizon_risk_composer_v2.py` 是长期风险 Composer 的并行 v2 合约。它把三件本来不应混在一起的事分开： + +```text +所有者/控制面选择的风险偏好 + + +策略、组合或插件候选的冻结 P3 风险能力与基准政策 + ↓ +仅在数学假设被证据支持时给出脱敏建议；否则 PARKED +``` + +它是纯离线函数：没有账户、券商、资金、凭据、网络、对象存储、调度、风险政策写入或交易能力。v2 不替换 `qsl.long_horizon_risk_observation.v1`，也不改变现有 v1 对象、私有读取口或任何运行时配置。 + +`python/scripts/long_horizon_risk_observation_ingress_v2.py` 提供对应的受限私有读取口：它只接受调用方注入的 `read_exact`，按 `long-horizon-risk-observations/v2//.json` 读取一个不超过 2 MiB 的精确对象，并核验其 candidate 与 P3 摘要。它不会列举、猜测最新、重试另一对象、写入、覆盖或删除。profile selection 由控制面单独传入，不能从存储中发现或替换。当前没有实际云存储 adapter、运行身份或调度接线。 + +## 一个人工选择，所有候选复用 + +控制面保存可移植的 `qsl.risk_profile_selection.v1`，只有下列不可随候选历史反向拟合的字段： + +| 字段 | 含义 | +| --- | --- | +| `profile_id` | 与偏好一一对应且带版本，例如 `balanced_compounding_v1` | +| `risk_preference` | `CAPITAL_PRESERVATION`、`BALANCED_COMPOUNDING` 或 `GROWTH_COMPOUNDING` | +| `selection_sha256` | 防止把同一名称偷换为另一档位 | + +这个工件不包含账户、平台、资金或 live 授权。控制面在账户或组合层保存“哪个范围使用哪个 selection digest”；候选的 P3 观察件只读取该 digest 对应的偏好。策略、插件、平台适配器不能创建、变更或放宽 profile。 + +## V2 私有观察件 + +`qsl.long_horizon_risk_observation.v2` 继续绑定 candidate 和 P1/P2/P3/plugin 摘要、成对净收益路径及哈希，但将 v1 的单一基准描述展开为两份冻结声明。 + +### `risk_capability` + +| 字段 | 允许值 | 作用 | +| --- | --- | --- | +| `portfolio_scope` | `SINGLE_CANDIDATE` / `PORTFOLIO` | 区分单策略和必须组合级处理的候选 | +| `return_evaluation` | `LINEAR_NET_RETURN` / `REPLAY_REQUIRED` | 明确收益是否可以随尺度线性重放 | +| `cashflow_treatment` | `NOT_APPLICABLE` / `TIME_WEIGHTED` / `CASHFLOW_MATCHED` | 防止 DCA 充值、提款混入风险收益路径 | +| `risk_factor_coverage` | 有序、去重的风险因子集合 | 声明 P3 已覆盖的集中度、流动性、杠杆、跳空、保证金、相关性等因素 | + +组合候选必须声明 `PORTFOLIO` 且覆盖 `CORRELATION`。这不会自动证明相关性计算正确；P1/P2/P3 的摘要仍是可重放证据的绑定点。 + +### `benchmark_policy` + +基准政策绑定 `benchmark_id`、类型、交易日历、币种、收益口径、预登记定义摘要与年度 session 数。它支持以下预留类别： + +- `UNLEVERED_REFERENCE`：SOXL/SOXX、TQQQ/QQQ 等方向性候选； +- `POLICY_BLEND`：轮动或多资产组合的预登记混合基准； +- `CASH_EQUIVALENT` 与 `ABSOLUTE_RETURN_HURDLE`：市场中性/绝对收益候选。 + +这使基准的选择可审计，不能因为某个候选回测表现更好而临时换为有利基准。 + +## 当前可组合范围与闭合行为 + +v2 复用已验证的 v1 计算器，**仅**在以下范围输出 `ADVISORY_RECOMMENDATION_READY`： + +1. `SINGLE_CANDIDATE`； +2. `LINEAR_NET_RETURN`； +3. `NOT_APPLICABLE` 现金流处理； +4. `UNLEVERED_REFERENCE` 与 `TOTAL_RETURN_NET_OF_COST` 基准。 + +这正好适合经 P3 路径验证的单策略方向性 ETF 候选。算法仍使用配对路径回撤上限和等权 walk-forward/bootstrap/stress 证据家族。 + +其它类别已经能用同一 schema 表达，但在专用 Composer 实现前必须返回 `PARKED`： + +| 声明的情况 | 固定原因码 | 为什么不能暂时套用线性计算 | +| --- | --- | --- | +| 非线性仓位、期权或动态波动率控制 | `RETURN_SCALE_REPLAY_REQUIRED` | 每个风险尺度必须重放 P3，而非直接缩放收益率 | +| 多策略组合 | `PORTFOLIO_COMPOSER_REQUIRED` | 必须重算相关性、边际风险贡献和组合净收益 | +| DCA 或外部现金流 | `CASHFLOW_COMPOSER_REQUIRED` | 必须使用现金流匹配、时间一致的路径 | +| 混合、现金或绝对收益基准 | `BENCHMARK_POLICY_COMPOSER_REQUIRED` | 不能伪装为权益无杠杆基准 | +| 非净成本总收益基准口径 | `BENCHMARK_RETURN_BASIS_COMPOSER_REQUIRED` | 当前回撤比较不具可比性 | + +`PARKED` 没有尺度、最大回撤或前沿；它不是失败后的默认继续运行，更不能触发实盘动作。 + +## 插件和平台的边界 + +- 观察/信号插件没有独立收益路径，不能提交 v2 观察件或获得风险建议。 +- 改变下单、再平衡或持仓收益的插件，必须成为“策略 + plugin bundle”的新候选，并重新完成 P1/P2/P3。 +- 平台只消费已签名、已批准的下游政策；它可以因为流动性、风控或健康原因进一步降风险，但不能更改 profile、基准或扩大尺度。 +- 任何平台、账户或策略迁移前都要先完成自己的 P3 生产器和私有 ingress。不得用 SOXL 结果替代 TQQQ、组合、DCA 或插件候选。 + +## 迁移顺序 + +1. 维持 v1 运行，先让单策略 P3 生产器可同时构造并验证 v2 观察件； +2. 接入方向性单策略的 v2 建议，比较其与 v1 脱敏输出，仍只做建议； +3. 实现组合、现金流和重放型专用 Composer，并以新的 P3 数据生产器逐类接入； +4. 单独实现受限私有读取、政策作者和批准链路。风险建议永远不能直接进入 P4/P5/P6。 + +任何扩大尺度、改变 profile、基准、风险能力声明或 P3 路径的操作都必须产生新的 candidate/P1/P2/P3 绑定和独立批准；新证据退化时系统只可降尺度或 `PARKED`。 diff --git a/python/scripts/long_horizon_risk_composer_v2.py b/python/scripts/long_horizon_risk_composer_v2.py new file mode 100644 index 0000000..09fc622 --- /dev/null +++ b/python/scripts/long_horizon_risk_composer_v2.py @@ -0,0 +1,467 @@ +#!/usr/bin/env python3 +"""Validate and compose generic, private long-horizon risk observations. + +V2 is deliberately parallel to the v1 observation contract. It separates a +portable owner risk-profile selection from a candidate's risk capability and +benchmark policy. It has no storage, account, broker, credential, network, +policy-write, scheduler, or execution dependency. + +Only the subset whose economics can be evaluated safely by the proven v1 +linear return-path engine is composed today. All other declared capabilities +return a redacted ``PARKED`` recommendation instead of pretending that a +leveraged, cash-flow, portfolio, or nonlinear strategy can be linearly scaled. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from typing import Any + +from long_horizon_risk_composer import ( + RISK_COMPOSER_INPUT_SCHEMA_ID, + _FRONTIER_FIELDS, + _IDENTITY_PATTERN, + _RISK_PREFERENCES, + _RISK_SCALE_GRID_BPS, + _canonical_json, + _expect_exact_keys, + _expect_identity, + _expect_list, + _expect_nonnegative_integer, + _expect_object, + _expect_positive_integer, + _expect_sha256, + _fail, + _reject_forbidden_material, + _reject_non_finite_or_null, + _validate_candidate, + _validate_scenario, + _validate_source_evidence, + calculate_risk_composer_input_sha256, + compose_long_horizon_risk_recommendation, +) + + +RISK_PROFILE_SELECTION_SCHEMA_ID = "qsl.risk_profile_selection.v1" +RISK_OBSERVATION_V2_SCHEMA_ID = "qsl.long_horizon_risk_observation.v2" +RISK_RECOMMENDATION_V2_SCHEMA_ID = "qsl.long_horizon_risk_recommendation.v2" + +_PROFILE_SELECTION_FIELDS = {"schema", "profile_id", "risk_preference", "selection_sha256"} +_RISK_CAPABILITY_FIELDS = { + "portfolio_scope", + "return_evaluation", + "cashflow_treatment", + "risk_factor_coverage", +} +_BENCHMARK_POLICY_FIELDS = { + "benchmark_id", + "benchmark_kind", + "calendar_id", + "currency", + "return_basis", + "definition_sha256", + "sessions_per_year", +} +_OBSERVATION_V2_FIELDS = { + "schema", + "candidate", + "source_evidence", + "risk_capability", + "benchmark_policy", + "scenario_paths", + "observation_sha256", +} +_RECOMMENDATION_V2_FIELDS = { + "schema", + "candidate", + "source_evidence", + "risk_profile", + "risk_capability", + "benchmark_policy", + "observation_sha256", + "status", + "reason_codes", + "recommended_scale_bps", + "recommended_max_drawdown_bps", + "frontier", + "recommendation_sha256", +} +_PROFILE_IDS = { + "CAPITAL_PRESERVATION": "capital_preservation_v1", + "BALANCED_COMPOUNDING": "balanced_compounding_v1", + "GROWTH_COMPOUNDING": "growth_compounding_v1", +} +_PORTFOLIO_SCOPES = {"SINGLE_CANDIDATE", "PORTFOLIO"} +_RETURN_EVALUATIONS = {"LINEAR_NET_RETURN", "REPLAY_REQUIRED"} +_CASHFLOW_TREATMENTS = {"NOT_APPLICABLE", "TIME_WEIGHTED", "CASHFLOW_MATCHED"} +_BENCHMARK_KINDS = { + "UNLEVERED_REFERENCE", + "POLICY_BLEND", + "CASH_EQUIVALENT", + "ABSOLUTE_RETURN_HURDLE", +} +_RETURN_BASES = { + "TOTAL_RETURN_NET_OF_COST", + "TIME_WEIGHTED_TOTAL_RETURN", + "CASHFLOW_MATCHED_RETURN", +} +_RISK_FACTORS = { + "CONCENTRATION", + "CORRELATION", + "FINANCING", + "GAP", + "LEVERAGE", + "LIQUIDITY", + "MARGIN", + "OPTIONS_ASSIGNMENT", + "VOLATILITY", +} + + +def _calculate_digest(value: Mapping[str, Any], digest_field: str, label: str) -> str: + return hashlib.sha256(_canonical_json(value, digest_field, label).encode("utf-8")).hexdigest() + + +def calculate_risk_profile_selection_sha256(value: Mapping[str, Any]) -> str: + """Return the immutable digest of one owner-selected generic profile.""" + return _calculate_digest(value, "selection_sha256", "risk profile selection") + + +def calculate_risk_observation_v2_sha256(value: Mapping[str, Any]) -> str: + """Return the immutable digest of one private v2 P3 observation.""" + return _calculate_digest(value, "observation_sha256", "long-horizon risk observation v2") + + +def calculate_risk_recommendation_v2_sha256(value: Mapping[str, Any]) -> str: + """Return the digest of one redacted v2 advisory recommendation.""" + return _calculate_digest(value, "recommendation_sha256", "long-horizon risk recommendation v2") + + +def validate_risk_profile_selection(value: Any) -> dict[str, str]: + """Validate a portable named profile, without binding it to an account or policy. + + The control plane attaches this digest to an account or portfolio outside + this contract. Keeping account identity out of the artifact prevents a + risk recommendation from becoming an execution instruction. + """ + _reject_non_finite_or_null(value, "risk profile selection") + _reject_forbidden_material(value, "risk profile selection") + selection = _expect_object(value, "risk profile selection") + _expect_exact_keys(selection, _PROFILE_SELECTION_FIELDS, "risk profile selection") + if selection["schema"] != RISK_PROFILE_SELECTION_SCHEMA_ID: + _fail(f"risk profile selection.schema must be {RISK_PROFILE_SELECTION_SCHEMA_ID}") + preference = selection["risk_preference"] + if preference not in _RISK_PREFERENCES: + _fail("risk profile selection.risk_preference is not supported") + expected_profile_id = _PROFILE_IDS[preference] + if selection["profile_id"] != expected_profile_id: + _fail("risk profile selection.profile_id does not match risk_preference") + normalized = { + "schema": RISK_PROFILE_SELECTION_SCHEMA_ID, + "profile_id": _expect_identity(selection["profile_id"], "risk profile selection.profile_id"), + "risk_preference": preference, + "selection_sha256": _expect_sha256(selection["selection_sha256"], "risk profile selection.selection_sha256"), + } + if normalized["selection_sha256"] != calculate_risk_profile_selection_sha256(normalized): + _fail("risk profile selection.selection_sha256 mismatch") + return normalized + + +def _validate_risk_factor_coverage(value: Any) -> list[str]: + factors = _expect_list(value, "risk_capability.risk_factor_coverage") + if not factors or len(factors) > len(_RISK_FACTORS): + _fail("risk_capability.risk_factor_coverage must be a non-empty bounded array") + if not all(isinstance(item, str) and item in _RISK_FACTORS for item in factors): + _fail("risk_capability.risk_factor_coverage contains an unsupported risk factor") + if list(factors) != sorted(set(factors)): + _fail("risk_capability.risk_factor_coverage must be sorted and unique") + return list(factors) + + +def _validate_risk_capability(value: Any, *, candidate_kind: str) -> dict[str, Any]: + capability = _expect_object(value, "risk_capability") + _expect_exact_keys(capability, _RISK_CAPABILITY_FIELDS, "risk_capability") + portfolio_scope = capability["portfolio_scope"] + if portfolio_scope not in _PORTFOLIO_SCOPES: + _fail("risk_capability.portfolio_scope is not supported") + if candidate_kind == "combo" and portfolio_scope != "PORTFOLIO": + _fail("combo candidates must declare PORTFOLIO scope") + return_evaluation = capability["return_evaluation"] + if return_evaluation not in _RETURN_EVALUATIONS: + _fail("risk_capability.return_evaluation is not supported") + cashflow_treatment = capability["cashflow_treatment"] + if cashflow_treatment not in _CASHFLOW_TREATMENTS: + _fail("risk_capability.cashflow_treatment is not supported") + factors = _validate_risk_factor_coverage(capability["risk_factor_coverage"]) + if portfolio_scope == "PORTFOLIO" and "CORRELATION" not in factors: + _fail("PORTFOLIO risk capability must cover CORRELATION") + return { + "portfolio_scope": portfolio_scope, + "return_evaluation": return_evaluation, + "cashflow_treatment": cashflow_treatment, + "risk_factor_coverage": factors, + } + + +def _validate_benchmark_policy(value: Any, *, cashflow_treatment: str) -> dict[str, Any]: + policy = _expect_object(value, "benchmark_policy") + _expect_exact_keys(policy, _BENCHMARK_POLICY_FIELDS, "benchmark_policy") + if policy["benchmark_kind"] not in _BENCHMARK_KINDS: + _fail("benchmark_policy.benchmark_kind is not supported") + if ( + not isinstance(policy["calendar_id"], str) + or not policy["calendar_id"].isupper() + or not policy["calendar_id"].isalnum() + ): + _fail("benchmark_policy.calendar_id must be an uppercase calendar identity") + if not isinstance(policy["currency"], str) or len(policy["currency"]) != 3 or not policy["currency"].isupper(): + _fail("benchmark_policy.currency must be a three-letter uppercase currency") + return_basis = policy["return_basis"] + if return_basis not in _RETURN_BASES: + _fail("benchmark_policy.return_basis is not supported") + if cashflow_treatment == "CASHFLOW_MATCHED" and return_basis != "CASHFLOW_MATCHED_RETURN": + _fail("CASHFLOW_MATCHED capability requires CASHFLOW_MATCHED_RETURN benchmark basis") + if cashflow_treatment != "CASHFLOW_MATCHED" and return_basis == "CASHFLOW_MATCHED_RETURN": + _fail("CASHFLOW_MATCHED_RETURN benchmark basis requires CASHFLOW_MATCHED capability") + return { + "benchmark_id": _expect_identity(policy["benchmark_id"], "benchmark_policy.benchmark_id"), + "benchmark_kind": policy["benchmark_kind"], + "calendar_id": policy["calendar_id"], + "currency": policy["currency"], + "return_basis": return_basis, + "definition_sha256": _expect_sha256(policy["definition_sha256"], "benchmark_policy.definition_sha256"), + "sessions_per_year": _expect_positive_integer( + policy["sessions_per_year"], "benchmark_policy.sessions_per_year", maximum=366 + ), + } + + +def validate_long_horizon_risk_observation_v2(value: Any) -> dict[str, Any]: + """Validate a generic private P3 observation without choosing a profile.""" + _reject_non_finite_or_null(value, "long-horizon risk observation v2") + _reject_forbidden_material(value, "long-horizon risk observation v2") + observation = _expect_object(value, "long-horizon risk observation v2") + _expect_exact_keys(observation, _OBSERVATION_V2_FIELDS, "long-horizon risk observation v2") + if observation["schema"] != RISK_OBSERVATION_V2_SCHEMA_ID: + _fail(f"long-horizon risk observation v2.schema must be {RISK_OBSERVATION_V2_SCHEMA_ID}") + candidate = _validate_candidate(observation["candidate"]) + capability = _validate_risk_capability(observation["risk_capability"], candidate_kind=candidate["candidate_kind"]) + paths = _expect_list(observation["scenario_paths"], "observation.scenario_paths") + if not paths or len(paths) > 12: + _fail("observation.scenario_paths must contain between 1 and 12 paths") + normalized = { + "schema": RISK_OBSERVATION_V2_SCHEMA_ID, + "candidate": candidate, + "source_evidence": _validate_source_evidence(observation["source_evidence"]), + "risk_capability": capability, + "benchmark_policy": _validate_benchmark_policy( + observation["benchmark_policy"], cashflow_treatment=capability["cashflow_treatment"] + ), + "scenario_paths": [_validate_scenario(item, index) for index, item in enumerate(paths)], + "observation_sha256": _expect_sha256( + observation["observation_sha256"], "long-horizon risk observation v2.observation_sha256" + ), + } + if len({path["scenario_id"] for path in normalized["scenario_paths"]}) != len(normalized["scenario_paths"]): + _fail("observation.scenario_paths.scenario_id values must be unique") + if normalized["observation_sha256"] != calculate_risk_observation_v2_sha256(normalized): + _fail("long-horizon risk observation v2.observation_sha256 mismatch") + return normalized + + +def _parked_recommendation( + observation: Mapping[str, Any], profile: Mapping[str, str], reasons: list[str] +) -> dict[str, Any]: + recommendation: dict[str, Any] = { + "schema": RISK_RECOMMENDATION_V2_SCHEMA_ID, + "candidate": dict(observation["candidate"]), + "source_evidence": dict(observation["source_evidence"]), + "risk_profile": dict(profile), + "risk_capability": dict(observation["risk_capability"]), + "benchmark_policy": dict(observation["benchmark_policy"]), + "observation_sha256": observation["observation_sha256"], + "status": "PARKED", + "reason_codes": reasons, + "recommended_scale_bps": None, + "recommended_max_drawdown_bps": None, + "frontier": [], + "recommendation_sha256": "", + } + recommendation["recommendation_sha256"] = calculate_risk_recommendation_v2_sha256(recommendation) + return recommendation + + +def _v1_compatibility_reasons(observation: Mapping[str, Any]) -> list[str]: + capability = observation["risk_capability"] + policy = observation["benchmark_policy"] + reasons: list[str] = [] + if capability["return_evaluation"] != "LINEAR_NET_RETURN": + reasons.append("RETURN_SCALE_REPLAY_REQUIRED") + if capability["portfolio_scope"] != "SINGLE_CANDIDATE": + reasons.append("PORTFOLIO_COMPOSER_REQUIRED") + if capability["cashflow_treatment"] != "NOT_APPLICABLE": + reasons.append("CASHFLOW_COMPOSER_REQUIRED") + if policy["benchmark_kind"] != "UNLEVERED_REFERENCE": + reasons.append("BENCHMARK_POLICY_COMPOSER_REQUIRED") + if policy["return_basis"] != "TOTAL_RETURN_NET_OF_COST": + reasons.append("BENCHMARK_RETURN_BASIS_COMPOSER_REQUIRED") + return reasons + + +def compose_long_horizon_risk_recommendation_v2( + observation: Any, profile_selection: Any +) -> dict[str, Any]: + """Produce a redacted advisory or a fail-closed v2 parked result. + + This function is intentionally a facade around the v1 algorithm for its + narrow verified subset. Unsupported economics are identified by stable + reason codes; they are never coerced to a linear equity calculation. + """ + validated_observation = validate_long_horizon_risk_observation_v2(observation) + validated_profile = validate_risk_profile_selection(profile_selection) + reasons = _v1_compatibility_reasons(validated_observation) + if reasons: + return _parked_recommendation(validated_observation, validated_profile, reasons) + + legacy_input: dict[str, Any] = { + "schema": RISK_COMPOSER_INPUT_SCHEMA_ID, + "candidate": validated_observation["candidate"], + "source_evidence": validated_observation["source_evidence"], + "objective": { + "risk_preference": validated_profile["risk_preference"], + "benchmark_id": validated_observation["benchmark_policy"]["benchmark_id"], + "benchmark_kind": "unlevered_reference", + "sessions_per_year": validated_observation["benchmark_policy"]["sessions_per_year"], + }, + "scenario_paths": validated_observation["scenario_paths"], + "input_sha256": "", + } + legacy_input["input_sha256"] = calculate_risk_composer_input_sha256(legacy_input) + legacy = compose_long_horizon_risk_recommendation(legacy_input) + recommendation: dict[str, Any] = { + "schema": RISK_RECOMMENDATION_V2_SCHEMA_ID, + "candidate": dict(validated_observation["candidate"]), + "source_evidence": dict(validated_observation["source_evidence"]), + "risk_profile": dict(validated_profile), + "risk_capability": dict(validated_observation["risk_capability"]), + "benchmark_policy": dict(validated_observation["benchmark_policy"]), + "observation_sha256": validated_observation["observation_sha256"], + "status": legacy["status"], + "reason_codes": list(legacy["reason_codes"]), + "recommended_scale_bps": legacy["recommended_scale_bps"], + "recommended_max_drawdown_bps": legacy["recommended_max_drawdown_bps"], + "frontier": legacy["frontier"], + "recommendation_sha256": "", + } + recommendation["recommendation_sha256"] = calculate_risk_recommendation_v2_sha256(recommendation) + return validate_risk_recommendation_v2(recommendation) + + +def _validate_reason_codes(value: Any) -> list[str]: + reasons = _expect_list(value, "risk recommendation v2.reason_codes") + if not all(isinstance(reason, str) and _IDENTITY_PATTERN.fullmatch(reason.lower()) for reason in reasons): + _fail("risk recommendation v2.reason_codes must contain stable identifiers") + return list(reasons) + + +def _validate_frontier(value: Any) -> list[dict[str, Any]]: + frontier = _expect_list(value, "risk recommendation v2.frontier") + if len(frontier) != len(_RISK_SCALE_GRID_BPS): + _fail("ready v2 recommendation must have the complete frontier") + normalized: list[dict[str, Any]] = [] + for index, item in enumerate(frontier): + row = _expect_object(item, f"risk recommendation v2.frontier[{index}]") + _expect_exact_keys(row, _FRONTIER_FIELDS, f"risk recommendation v2.frontier[{index}]") + if row["scale_bps"] != _RISK_SCALE_GRID_BPS[index]: + _fail("v2 frontier scale grid must be immutable and ordered") + for field in ( + "positive_growth_scenarios", + "scenario_count", + "worst_max_drawdown_bps", + "worst_relative_drawdown_bps", + "worst_benchmark_drawdown_bps", + "worst_underwater_sessions", + ): + _expect_nonnegative_integer( + row[field], + f"risk recommendation v2.frontier[{index}].{field}", + maximum=10_000_000, + ) + if isinstance(row["median_log_growth_ppm"], bool) or not isinstance(row["median_log_growth_ppm"], int): + _fail(f"risk recommendation v2.frontier[{index}].median_log_growth_ppm must be an integer") + if not isinstance(row["eligible"], bool): + _fail(f"risk recommendation v2.frontier[{index}].eligible must be a boolean") + normalized.append(dict(row)) + return normalized + + +def validate_risk_recommendation_v2(value: Any) -> dict[str, Any]: + """Validate a safe-to-publish v2 recommendation without return paths.""" + _reject_non_finite_or_null(value, "risk recommendation v2") + _reject_forbidden_material(value, "risk recommendation v2") + recommendation = _expect_object(value, "risk recommendation v2") + _expect_exact_keys(recommendation, _RECOMMENDATION_V2_FIELDS, "risk recommendation v2") + if recommendation["schema"] != RISK_RECOMMENDATION_V2_SCHEMA_ID: + _fail(f"risk recommendation v2.schema must be {RISK_RECOMMENDATION_V2_SCHEMA_ID}") + candidate = _validate_candidate(recommendation["candidate"]) + capability = _validate_risk_capability( + recommendation["risk_capability"], candidate_kind=candidate["candidate_kind"] + ) + normalized = { + "schema": RISK_RECOMMENDATION_V2_SCHEMA_ID, + "candidate": candidate, + "source_evidence": _validate_source_evidence(recommendation["source_evidence"]), + "risk_profile": validate_risk_profile_selection(recommendation["risk_profile"]), + "risk_capability": capability, + "benchmark_policy": _validate_benchmark_policy( + recommendation["benchmark_policy"], cashflow_treatment=capability["cashflow_treatment"] + ), + "observation_sha256": _expect_sha256( + recommendation["observation_sha256"], "risk recommendation v2.observation_sha256" + ), + "status": recommendation["status"], + "reason_codes": _validate_reason_codes(recommendation["reason_codes"]), + "recommended_scale_bps": recommendation["recommended_scale_bps"], + "recommended_max_drawdown_bps": recommendation["recommended_max_drawdown_bps"], + "frontier": [], + "recommendation_sha256": _expect_sha256( + recommendation["recommendation_sha256"], "risk recommendation v2.recommendation_sha256" + ), + } + if normalized["status"] == "PARKED": + if ( + not normalized["reason_codes"] + or normalized["recommended_scale_bps"] is not None + or normalized["recommended_max_drawdown_bps"] is not None + or recommendation["frontier"] + ): + _fail("PARKED v2 recommendation must have reasons and no frontier or numeric recommendation") + elif normalized["status"] == "ADVISORY_RECOMMENDATION_READY": + if normalized["reason_codes"]: + _fail("ready v2 recommendation must not have reason codes") + normalized["frontier"] = _validate_frontier(recommendation["frontier"]) + _expect_positive_integer(normalized["recommended_scale_bps"], "recommended_scale_bps", maximum=10_000) + _expect_nonnegative_integer( + normalized["recommended_max_drawdown_bps"], "recommended_max_drawdown_bps", maximum=10_000 + ) + else: + _fail("risk recommendation v2.status must be PARKED or ADVISORY_RECOMMENDATION_READY") + if normalized["recommendation_sha256"] != calculate_risk_recommendation_v2_sha256(normalized): + _fail("risk recommendation v2.recommendation_sha256 mismatch") + return normalized + + +__all__ = [ + "RISK_OBSERVATION_V2_SCHEMA_ID", + "RISK_PROFILE_SELECTION_SCHEMA_ID", + "RISK_RECOMMENDATION_V2_SCHEMA_ID", + "calculate_risk_observation_v2_sha256", + "calculate_risk_profile_selection_sha256", + "calculate_risk_recommendation_v2_sha256", + "compose_long_horizon_risk_recommendation_v2", + "validate_long_horizon_risk_observation_v2", + "validate_risk_profile_selection", + "validate_risk_recommendation_v2", +] diff --git a/python/scripts/long_horizon_risk_observation_ingress_v2.py b/python/scripts/long_horizon_risk_observation_ingress_v2.py new file mode 100644 index 0000000..5684632 --- /dev/null +++ b/python/scripts/long_horizon_risk_observation_ingress_v2.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +"""Read one exact private v2 P3 observation through an injected read port. + +This is the v2 counterpart to the v1 ingress. It has no cloud SDK, +credential, network, bucket, listing, write, delete, policy, or execution +dependency. The caller injects one exact-object reader; profile selection is +passed separately by the control plane and cannot be discovered from storage. +""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from typing import Any + +from long_horizon_risk_composer import LongHorizonRiskComposerError, parse_risk_composer_input_json +from long_horizon_risk_composer_v2 import ( + compose_long_horizon_risk_recommendation_v2, + validate_long_horizon_risk_observation_v2, +) + + +PRIVATE_OBSERVATION_V2_OBJECT_PREFIX = "long-horizon-risk-observations/v2" +MAX_PRIVATE_OBSERVATION_V2_BYTES = 2 * 1024 * 1024 +_IDENTITY_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$") +_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") + + +class LongHorizonRiskObservationV2IngressError(ValueError): + """Fail-closed v2 ingress error without path or backend information.""" + + +def _fail() -> None: + raise LongHorizonRiskObservationV2IngressError("private long-horizon risk observation unavailable") + + +def private_observation_v2_object_name(*, candidate_id: str, p3_evidence_sha256: str) -> str: + """Return the only readable v2 object name for one candidate/P3 identity.""" + if not _IDENTITY_PATTERN.fullmatch(candidate_id) or not _SHA256_PATTERN.fullmatch(p3_evidence_sha256): + _fail() + return f"{PRIVATE_OBSERVATION_V2_OBJECT_PREFIX}/{candidate_id}/{p3_evidence_sha256}.json" + + +def load_private_long_horizon_risk_observation_v2( + *, + candidate_id: str, + p3_evidence_sha256: str, + read_exact: Callable[[str], bytes], +) -> dict[str, Any]: + """Read and validate one exact v2 object; no fallback object is possible.""" + if not callable(read_exact): + _fail() + object_name = private_observation_v2_object_name( + candidate_id=candidate_id, + p3_evidence_sha256=p3_evidence_sha256, + ) + try: + raw = read_exact(object_name) + except Exception as exc: # pragma: no cover - injected I/O boundary + raise LongHorizonRiskObservationV2IngressError( + "private long-horizon risk observation unavailable" + ) from exc + if not isinstance(raw, bytes) or not raw or len(raw) > MAX_PRIVATE_OBSERVATION_V2_BYTES: + _fail() + try: + observation = validate_long_horizon_risk_observation_v2( + parse_risk_composer_input_json(raw.decode("utf-8")) + ) + except (UnicodeDecodeError, LongHorizonRiskComposerError) as exc: + raise LongHorizonRiskObservationV2IngressError( + "private long-horizon risk observation unavailable" + ) from exc + if ( + observation["candidate"]["candidate_id"] != candidate_id + or observation["source_evidence"]["p3_evidence_sha256"] != p3_evidence_sha256 + ): + _fail() + return observation + + +def compose_from_private_long_horizon_risk_observation_v2( + *, + candidate_id: str, + p3_evidence_sha256: str, + profile_selection: Any, + read_exact: Callable[[str], bytes], +) -> dict[str, Any]: + """Return a v2 redacted advisory or parked result from one exact object.""" + observation = load_private_long_horizon_risk_observation_v2( + candidate_id=candidate_id, + p3_evidence_sha256=p3_evidence_sha256, + read_exact=read_exact, + ) + try: + return compose_long_horizon_risk_recommendation_v2(observation, profile_selection) + except LongHorizonRiskComposerError as exc: + raise LongHorizonRiskObservationV2IngressError( + "private long-horizon risk observation unavailable" + ) from exc + + +__all__ = [ + "LongHorizonRiskObservationV2IngressError", + "MAX_PRIVATE_OBSERVATION_V2_BYTES", + "PRIVATE_OBSERVATION_V2_OBJECT_PREFIX", + "compose_from_private_long_horizon_risk_observation_v2", + "load_private_long_horizon_risk_observation_v2", + "private_observation_v2_object_name", +] diff --git a/python/tests/test_long_horizon_risk_observation_ingress_v2.py b/python/tests/test_long_horizon_risk_observation_ingress_v2.py new file mode 100644 index 0000000..48b3b64 --- /dev/null +++ b/python/tests/test_long_horizon_risk_observation_ingress_v2.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +import sys +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" + + +def _load_module(name: str): + spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +composer = _load_module("long_horizon_risk_composer") +composer_v2 = _load_module("long_horizon_risk_composer_v2") +ingress_v2 = _load_module("long_horizon_risk_observation_ingress_v2") + + +class LongHorizonRiskObservationV2IngressTest(unittest.TestCase): + @staticmethod + def _returns(gain_bps: int, drawdown_bps: int) -> list[int]: + return [gain_bps] * 240 + [-drawdown_bps] * 12 + + def _profile(self) -> dict[str, object]: + value: dict[str, object] = { + "schema": "qsl.risk_profile_selection.v1", + "profile_id": "balanced_compounding_v1", + "risk_preference": "BALANCED_COMPOUNDING", + "selection_sha256": "", + } + value["selection_sha256"] = composer_v2.calculate_risk_profile_selection_sha256(value) + return value + + def _observation(self) -> dict[str, object]: + paths = [ + { + "scenario_id": f"soxl_soxx_{kind.lower()}_{index}", + "scenario_kind": kind, + "session_count": 253, + "strategy_returns_bps": self._returns(16 - index, 124 + index), + "benchmark_returns_bps": self._returns(10 - index, 100 + index), + } + for index, kind in enumerate(("WALK_FORWARD", "BOOTSTRAP", "STRESS"), start=1) + ] + value: dict[str, object] = { + "schema": "qsl.long_horizon_risk_observation.v2", + "candidate": { + "candidate_id": "soxl_soxx_longterm_compounding", + "candidate_kind": "individual", + "strategy_repository": "QuantStrategyLab/UsEquityStrategies", + "strategy_revision": "a" * 40, + }, + "source_evidence": { + "p1_input_digest": "1" * 64, + "p2_config_digest": "2" * 64, + "p3_evidence_sha256": "3" * 64, + "plugin_bundle_sha256": "4" * 64, + }, + "risk_capability": { + "portfolio_scope": "SINGLE_CANDIDATE", + "return_evaluation": "LINEAR_NET_RETURN", + "cashflow_treatment": "NOT_APPLICABLE", + "risk_factor_coverage": ["CONCENTRATION", "LIQUIDITY"], + }, + "benchmark_policy": { + "benchmark_id": "soxx", + "benchmark_kind": "UNLEVERED_REFERENCE", + "calendar_id": "XNYS", + "currency": "USD", + "return_basis": "TOTAL_RETURN_NET_OF_COST", + "definition_sha256": "5" * 64, + "sessions_per_year": 252, + }, + "scenario_paths": paths, + "observation_sha256": "", + } + value["observation_sha256"] = composer_v2.calculate_risk_observation_v2_sha256(value) + return value + + def test_reads_one_exact_v2_object_then_returns_only_a_redacted_recommendation(self): + observation = self._observation() + raw = json.dumps(observation, sort_keys=True, separators=(",", ":")).encode("utf-8") + calls: list[str] = [] + + result = ingress_v2.compose_from_private_long_horizon_risk_observation_v2( + candidate_id="soxl_soxx_longterm_compounding", + p3_evidence_sha256="3" * 64, + profile_selection=self._profile(), + read_exact=lambda object_name: calls.append(object_name) or raw, + ) + + self.assertEqual( + calls, + [ + "long-horizon-risk-observations/v2/soxl_soxx_longterm_compounding/" + + ("3" * 64) + + ".json" + ], + ) + self.assertEqual(result["status"], "ADVISORY_RECOMMENDATION_READY") + rendered = json.dumps(result, sort_keys=True).lower() + self.assertNotIn("strategy_returns", rendered) + self.assertNotIn("benchmark_returns", rendered) + self.assertNotIn("account", rendered) + self.assertNotIn("broker", rendered) + + def test_invalid_profile_or_tampered_observation_fails_closed_without_fallback(self): + observation = self._observation() + raw = json.dumps(observation).encode("utf-8") + invalid_profile = self._profile() + invalid_profile["selection_sha256"] = "0" * 64 + with self.assertRaisesRegex(ingress_v2.LongHorizonRiskObservationV2IngressError, "unavailable"): + ingress_v2.compose_from_private_long_horizon_risk_observation_v2( + candidate_id="soxl_soxx_longterm_compounding", + p3_evidence_sha256="3" * 64, + profile_selection=invalid_profile, + read_exact=lambda _object_name: raw, + ) + + tampered = copy.deepcopy(observation) + tampered["scenario_paths"][0]["strategy_returns_bps"][0] = 99 + with self.assertRaisesRegex(ingress_v2.LongHorizonRiskObservationV2IngressError, "unavailable"): + ingress_v2.load_private_long_horizon_risk_observation_v2( + candidate_id="soxl_soxx_longterm_compounding", + p3_evidence_sha256="3" * 64, + read_exact=lambda _object_name: json.dumps(tampered).encode("utf-8"), + ) + + def test_invalid_identity_never_reaches_the_injected_reader(self): + calls: list[str] = [] + with self.assertRaisesRegex(ingress_v2.LongHorizonRiskObservationV2IngressError, "unavailable"): + ingress_v2.load_private_long_horizon_risk_observation_v2( + candidate_id="../latest", + p3_evidence_sha256="3" * 64, + read_exact=lambda object_name: calls.append(object_name) or b"{}", + ) + self.assertEqual(calls, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/python/tests/test_long_horizon_risk_observation_v2.py b/python/tests/test_long_horizon_risk_observation_v2.py new file mode 100644 index 0000000..cc89081 --- /dev/null +++ b/python/tests/test_long_horizon_risk_observation_v2.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +import sys +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" + + +def _load_module(name: str): + spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +composer = _load_module("long_horizon_risk_composer") +composer_v2 = _load_module("long_horizon_risk_composer_v2") + + +class LongHorizonRiskObservationV2Test(unittest.TestCase): + @staticmethod + def _returns(gain_bps: int, drawdown_bps: int) -> list[int]: + return [gain_bps] * 240 + [-drawdown_bps] * 12 + + def _profile(self, preference: str = "BALANCED_COMPOUNDING") -> dict[str, object]: + value: dict[str, object] = { + "schema": "qsl.risk_profile_selection.v1", + "profile_id": { + "CAPITAL_PRESERVATION": "capital_preservation_v1", + "BALANCED_COMPOUNDING": "balanced_compounding_v1", + "GROWTH_COMPOUNDING": "growth_compounding_v1", + }[preference], + "risk_preference": preference, + "selection_sha256": "", + } + value["selection_sha256"] = composer_v2.calculate_risk_profile_selection_sha256(value) + return value + + def _observation(self) -> dict[str, object]: + paths = [ + { + "scenario_id": f"soxl_soxx_{kind.lower()}_{index}", + "scenario_kind": kind, + "session_count": 253, + "strategy_returns_bps": self._returns(16 - index, 124 + index), + "benchmark_returns_bps": self._returns(10 - index, 100 + index), + } + for index, kind in enumerate(("WALK_FORWARD", "BOOTSTRAP", "STRESS"), start=1) + ] + value: dict[str, object] = { + "schema": "qsl.long_horizon_risk_observation.v2", + "candidate": { + "candidate_id": "soxl_soxx_longterm_compounding", + "candidate_kind": "individual", + "strategy_repository": "QuantStrategyLab/UsEquityStrategies", + "strategy_revision": "a" * 40, + }, + "source_evidence": { + "p1_input_digest": "1" * 64, + "p2_config_digest": "2" * 64, + "p3_evidence_sha256": "3" * 64, + "plugin_bundle_sha256": "4" * 64, + }, + "risk_capability": { + "portfolio_scope": "SINGLE_CANDIDATE", + "return_evaluation": "LINEAR_NET_RETURN", + "cashflow_treatment": "NOT_APPLICABLE", + "risk_factor_coverage": ["CONCENTRATION", "LIQUIDITY"], + }, + "benchmark_policy": { + "benchmark_id": "soxx", + "benchmark_kind": "UNLEVERED_REFERENCE", + "calendar_id": "XNYS", + "currency": "USD", + "return_basis": "TOTAL_RETURN_NET_OF_COST", + "definition_sha256": "5" * 64, + "sessions_per_year": 252, + }, + "scenario_paths": paths, + "observation_sha256": "", + } + value["observation_sha256"] = composer_v2.calculate_risk_observation_v2_sha256(value) + return value + + def test_portable_profile_and_supported_observation_produce_a_redacted_advisory(self): + observation = self._observation() + profile = self._profile() + + recommendation = composer_v2.compose_long_horizon_risk_recommendation_v2(observation, profile) + + self.assertEqual(recommendation["status"], "ADVISORY_RECOMMENDATION_READY") + self.assertEqual(recommendation["risk_profile"], profile) + self.assertEqual(recommendation["benchmark_policy"]["benchmark_kind"], "UNLEVERED_REFERENCE") + self.assertEqual(recommendation, composer_v2.validate_risk_recommendation_v2(recommendation)) + rendered = json.dumps(recommendation, sort_keys=True).lower() + self.assertNotIn("strategy_returns", rendered) + self.assertNotIn("benchmark_returns", rendered) + self.assertNotIn("account", rendered) + self.assertNotIn("broker", rendered) + + def test_profile_is_hashed_and_cannot_claim_a_different_named_posture(self): + profile = self._profile() + tampered = copy.deepcopy(profile) + tampered["risk_preference"] = "GROWTH_COMPOUNDING" + with self.assertRaisesRegex(composer.LongHorizonRiskComposerError, "profile_id does not match"): + composer_v2.validate_risk_profile_selection(tampered) + + tampered = copy.deepcopy(profile) + tampered["profile_id"] = "growth_compounding_v1" + tampered["selection_sha256"] = composer_v2.calculate_risk_profile_selection_sha256(tampered) + with self.assertRaisesRegex(composer.LongHorizonRiskComposerError, "profile_id does not match"): + composer_v2.validate_risk_profile_selection(tampered) + + def test_nonlinear_replay_requirement_parks_instead_of_scaling_return_paths(self): + observation = self._observation() + observation["risk_capability"]["return_evaluation"] = "REPLAY_REQUIRED" + observation["observation_sha256"] = composer_v2.calculate_risk_observation_v2_sha256(observation) + + recommendation = composer_v2.compose_long_horizon_risk_recommendation_v2(observation, self._profile()) + + self.assertEqual(recommendation["status"], "PARKED") + self.assertEqual(recommendation["reason_codes"], ["RETURN_SCALE_REPLAY_REQUIRED"]) + self.assertIsNone(recommendation["recommended_scale_bps"]) + + def test_portfolios_require_correlation_coverage_and_a_dedicated_portfolio_composer(self): + observation = self._observation() + observation["candidate"]["candidate_kind"] = "combo" + observation["risk_capability"]["portfolio_scope"] = "PORTFOLIO" + observation["observation_sha256"] = composer_v2.calculate_risk_observation_v2_sha256(observation) + with self.assertRaisesRegex(composer.LongHorizonRiskComposerError, "must cover CORRELATION"): + composer_v2.validate_long_horizon_risk_observation_v2(observation) + + observation["risk_capability"]["risk_factor_coverage"] = ["CONCENTRATION", "CORRELATION", "LIQUIDITY"] + observation["benchmark_policy"]["benchmark_kind"] = "POLICY_BLEND" + observation["observation_sha256"] = composer_v2.calculate_risk_observation_v2_sha256(observation) + recommendation = composer_v2.compose_long_horizon_risk_recommendation_v2(observation, self._profile()) + self.assertEqual(recommendation["status"], "PARKED") + self.assertEqual( + recommendation["reason_codes"], + ["PORTFOLIO_COMPOSER_REQUIRED", "BENCHMARK_POLICY_COMPOSER_REQUIRED"], + ) + + def test_cashflow_matched_strategies_cannot_be_coerced_to_time_weighted_linear_math(self): + observation = self._observation() + observation["risk_capability"]["cashflow_treatment"] = "CASHFLOW_MATCHED" + observation["benchmark_policy"]["return_basis"] = "CASHFLOW_MATCHED_RETURN" + observation["observation_sha256"] = composer_v2.calculate_risk_observation_v2_sha256(observation) + + recommendation = composer_v2.compose_long_horizon_risk_recommendation_v2(observation, self._profile()) + + self.assertEqual(recommendation["status"], "PARKED") + self.assertEqual( + recommendation["reason_codes"], + ["CASHFLOW_COMPOSER_REQUIRED", "BENCHMARK_RETURN_BASIS_COMPOSER_REQUIRED"], + ) + + invalid = copy.deepcopy(observation) + invalid["benchmark_policy"]["return_basis"] = "TOTAL_RETURN_NET_OF_COST" + invalid["observation_sha256"] = composer_v2.calculate_risk_observation_v2_sha256(invalid) + with self.assertRaisesRegex(composer.LongHorizonRiskComposerError, "requires CASHFLOW_MATCHED_RETURN"): + composer_v2.validate_long_horizon_risk_observation_v2(invalid) + + +if __name__ == "__main__": + unittest.main()