From f9095b3700719265bbb5f002e0787408b5e5139e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 27 Aug 2026 17:28:02 +0800 Subject: [PATCH] feat(risk): bind owner preference after private P3 ingress Co-Authored-By: Codex --- ...qsl_long_horizon_risk_composer_v1.zh-CN.md | 8 +- python/scripts/long_horizon_risk_composer.py | 132 +++++++++++++++++- .../tests/test_long_horizon_risk_composer.py | 81 +++++++++++ 3 files changed, 213 insertions(+), 8 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 73b5a08..cf16dda 100644 --- a/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md +++ b/docs/qsl_long_horizon_risk_composer_v1.zh-CN.md @@ -1,11 +1,13 @@ # QSL 长期复利风险政策 Composer V1 -> 状态：`ADVISORY_CORE_IMPLEMENTED_NOT_WIRED` +> 状态：`PRIVATE_P3_OBSERVATION_INGRESS_IMPLEMENTED_POLICY_WRITE_NOT_WIRED` `python/scripts/long_horizon_risk_composer.py` 是所有策略、组合与插件可复用的离线风险设计内核。 它把已经冻结、净成本后的 P3 策略收益路径与同周期的无杠杆基准路径，计算成一个脱敏的风险尺度前沿。 它不读取账户、资金、券商、凭据或网络；不写入风险政策、不改策略参数、不启动调度，也不能授予 P4、P5 或 P6。 +研究管道与控制面之间使用 `qsl.long_horizon_risk_observation.v1`。研究管道只负责产出候选、P1/P2/P3/plugin 摘要、无杠杆基准和成对净收益路径；控制面必须显式叠加所有者选择的风险偏好，才可转换为 Composer 输入。观察件是**私有 ingress 工件**：不能上传到公开仓库、Actions 公开摘要、控制台或 AI 上下文；Composer 输出才是可发布的脱敏摘要。 + ## 人和系统的分工 人工只选择三个简单、重要的偏好之一： @@ -18,6 +20,8 @@ 这三个倍数是透明、版本化的偏好模板，不是模型从历史数据“发现”的真理。系统计算的内容是每个候选在每个尺度下的实际净成本路径、最大回撤、相对基准回撤、水下持续期和每 session 对数几何增长；它不会把一次历史最优结果伪装成未来保证。 +命令行也遵守这条分工：已有 owner-bound 输入可用 `--input`；私有观察件必须同时给出 `--observation` 与 `--risk-preference`。缺少偏好即失败，不会静默选择“均衡”或任何默认档位。 + ## 必要证据和计算方法 输入必须精确绑定 candidate revision 与 P1/P2/P3/plugin 摘要，并至少包含每类一个完整的 252-session 以上路径： @@ -33,6 +37,8 @@ - 最坏基准最大回撤与最长水下期； - 至少三分之二情景为正增长，且策略最坏回撤不超过所选偏好的基准回撤倍数时，才标记该尺度合格。 +每个情景明确携带 `session_count`，且必须恰好比收益率数组多一个起点：例如 252 个 XNYS 观测日对应 251 个相邻日收益率。长度门检查的是已签名的观测日数量，避免把完整的一年前瞻窗口误判为不足，也避免把收益率数组伪装成更多交易日。 + 在所有合格尺度中，选择下中位数对数几何增长最高者；同分时选择更低风险尺度。输出的 `recommended_max_drawdown_bps` 是由冻结的基准路径和偏好模板计算出的候选上限，不是订单阈值或已启用政策。 ## SOXL 与 TQQQ 的使用方式 diff --git a/python/scripts/long_horizon_risk_composer.py b/python/scripts/long_horizon_risk_composer.py index 250ca45..c164e36 100644 --- a/python/scripts/long_horizon_risk_composer.py +++ b/python/scripts/long_horizon_risk_composer.py @@ -29,6 +29,7 @@ RISK_COMPOSER_INPUT_SCHEMA_ID = "qsl.long_horizon_risk_composer_input.v1" RISK_COMPOSER_RECOMMENDATION_SCHEMA_ID = "qsl.long_horizon_risk_composer_recommendation.v1" +RISK_OBSERVATION_SCHEMA_ID = "qsl.long_horizon_risk_observation.v1" _IDENTITY_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$") _REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$") _REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$") @@ -41,10 +42,18 @@ re.IGNORECASE, ) _INPUT_FIELDS = {"schema", "candidate", "source_evidence", "objective", "scenario_paths", "input_sha256"} +_OBSERVATION_FIELDS = {"schema", "candidate", "source_evidence", "benchmark", "scenario_paths", "observation_sha256"} _CANDIDATE_FIELDS = {"candidate_id", "candidate_kind", "strategy_repository", "strategy_revision"} _SOURCE_EVIDENCE_FIELDS = {"p1_input_digest", "p2_config_digest", "p3_evidence_sha256", "plugin_bundle_sha256"} _OBJECTIVE_FIELDS = {"risk_preference", "benchmark_id", "benchmark_kind", "sessions_per_year"} -_SCENARIO_FIELDS = {"scenario_id", "scenario_kind", "strategy_returns_bps", "benchmark_returns_bps"} +_BENCHMARK_FIELDS = {"benchmark_id", "benchmark_kind", "sessions_per_year"} +_SCENARIO_FIELDS = { + "scenario_id", + "scenario_kind", + "session_count", + "strategy_returns_bps", + "benchmark_returns_bps", +} _RECOMMENDATION_FIELDS = { "schema", "candidate", @@ -190,6 +199,13 @@ def calculate_risk_composer_recommendation_sha256(value: Mapping[str, Any]) -> s ).hexdigest() +def calculate_risk_observation_sha256(value: Mapping[str, Any]) -> str: + """Return the stable identity of one private P3 return-path observation.""" + return hashlib.sha256( + _canonical_json(value, "observation_sha256", "long-horizon risk observation").encode("utf-8") + ).hexdigest() + + def _validate_candidate(value: Any) -> dict[str, str]: candidate = _expect_object(value, "candidate") _expect_exact_keys(candidate, _CANDIDATE_FIELDS, "candidate") @@ -234,6 +250,20 @@ def _validate_objective(value: Any) -> dict[str, Any]: } +def _validate_benchmark(value: Any) -> dict[str, Any]: + benchmark = _expect_object(value, "benchmark") + _expect_exact_keys(benchmark, _BENCHMARK_FIELDS, "benchmark") + if benchmark["benchmark_kind"] != "unlevered_reference": + _fail("benchmark.benchmark_kind must be unlevered_reference") + return { + "benchmark_id": _expect_identity(benchmark["benchmark_id"], "benchmark.benchmark_id"), + "benchmark_kind": _expect_identity(benchmark["benchmark_kind"], "benchmark.benchmark_kind"), + "sessions_per_year": _expect_positive_integer( + benchmark["sessions_per_year"], "benchmark.sessions_per_year", maximum=366 + ), + } + + def _validate_scenario(value: Any, index: int) -> dict[str, Any]: path = f"scenario_paths[{index}]" scenario = _expect_object(value, path) @@ -245,11 +275,17 @@ def _validate_scenario(value: Any, index: int) -> dict[str, Any]: benchmark_returns = _expect_list(scenario["benchmark_returns_bps"], f"{path}.benchmark_returns_bps") if len(strategy_returns) != len(benchmark_returns): _fail(f"{path} strategy and benchmark returns must have the same length") - if len(strategy_returns) > _MAX_SESSIONS_PER_SCENARIO: + session_count = _expect_positive_integer( + scenario["session_count"], f"{path}.session_count", maximum=_MAX_SESSIONS_PER_SCENARIO + ) + if len(strategy_returns) != session_count - 1: + _fail(f"{path} must contain exactly one fewer return than its observed sessions") + if len(strategy_returns) > _MAX_SESSIONS_PER_SCENARIO - 1: _fail(f"{path} exceeds the bounded session count") return { "scenario_id": _expect_identity(scenario["scenario_id"], f"{path}.scenario_id"), "scenario_kind": kind, + "session_count": session_count, "strategy_returns_bps": [ _expect_return_bps(item, f"{path}.strategy_returns_bps[{return_index}]") for return_index, item in enumerate(strategy_returns) @@ -287,6 +323,68 @@ def validate_risk_composer_input(value: Any) -> dict[str, Any]: return normalized +def validate_long_horizon_risk_observation(value: Any) -> dict[str, Any]: + """Validate a private P3 observation before an owner preference is bound. + + The observation contains only frozen candidate identity, evidence digests, + a same-window unlevered reference, and paired net-return paths. It is an + internal ingress artifact: it is never suitable for a public console or + AI prompt. Unlike a composer input it intentionally contains no risk + preference, because that is a control-plane/owner decision. + """ + _reject_non_finite_or_null(value, "long-horizon risk observation") + _reject_forbidden_material(value, "long-horizon risk observation") + observation = _expect_object(value, "long-horizon risk observation") + _expect_exact_keys(observation, _OBSERVATION_FIELDS, "long-horizon risk observation") + if observation["schema"] != RISK_OBSERVATION_SCHEMA_ID: + _fail(f"long-horizon risk observation.schema must be {RISK_OBSERVATION_SCHEMA_ID}") + paths = _expect_list(observation["scenario_paths"], "observation.scenario_paths") + if not paths or len(paths) > _MAX_SCENARIOS: + _fail(f"observation.scenario_paths must contain between 1 and {_MAX_SCENARIOS} paths") + normalized = { + "schema": RISK_OBSERVATION_SCHEMA_ID, + "candidate": _validate_candidate(observation["candidate"]), + "source_evidence": _validate_source_evidence(observation["source_evidence"]), + "benchmark": _validate_benchmark(observation["benchmark"]), + "scenario_paths": [_validate_scenario(item, index) for index, item in enumerate(paths)], + "observation_sha256": _expect_sha256( + observation["observation_sha256"], "long-horizon risk observation.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_sha256(normalized): + _fail("long-horizon risk observation.observation_sha256 mismatch") + return normalized + + +def build_risk_composer_input_from_observation( + observation: Any, *, risk_preference: str +) -> dict[str, Any]: + """Attach one explicit owner preference to a frozen private observation. + + This is deliberately a pure conversion. It does not refresh P3 data, + choose a preference, write a policy, or authorize any lifecycle phase. + """ + normalized = validate_long_horizon_risk_observation(observation) + objective = _validate_objective( + { + "risk_preference": risk_preference, + **normalized["benchmark"], + } + ) + result: dict[str, Any] = { + "schema": RISK_COMPOSER_INPUT_SCHEMA_ID, + "candidate": normalized["candidate"], + "source_evidence": normalized["source_evidence"], + "objective": objective, + "scenario_paths": normalized["scenario_paths"], + "input_sha256": "", + } + result["input_sha256"] = calculate_risk_composer_input_sha256(result) + return validate_risk_composer_input(result) + + def _scaled_return_bps(return_bps: int, scale_bps: int) -> int: product = return_bps * scale_bps return product // 10_000 if product >= 0 else -((-product + 9_999) // 10_000) @@ -368,7 +466,7 @@ def compose_long_horizon_risk_recommendation(value: Any) -> dict[str, Any]: reasons: list[str] = [] if kinds != _SCENARIO_KINDS: reasons.append("SCENARIO_KIND_COVERAGE_INCOMPLETE") - if any(len(path["strategy_returns_bps"]) < _MIN_SESSIONS_PER_SCENARIO for path in paths): + if any(path["session_count"] < _MIN_SESSIONS_PER_SCENARIO for path in paths): reasons.append("LONG_HORIZON_SESSION_COVERAGE_INCOMPLETE") if reasons: return _parked_recommendation(validated, reasons) @@ -522,13 +620,33 @@ def parse_risk_composer_input_json(text: str) -> dict[str, Any]: def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Compose a non-executing long-horizon risk recommendation") - parser.add_argument("--input", type=Path, required=True, help="frozen P3 return-path evidence JSON") + input_source = parser.add_mutually_exclusive_group(required=True) + input_source.add_argument("--input", type=Path, help="private, owner-bound P3 return-path evidence JSON") + input_source.add_argument( + "--observation", + type=Path, + help="private P3 observation JSON; requires an explicit --risk-preference", + ) + parser.add_argument( + "--risk-preference", + choices=tuple(sorted(_RISK_PREFERENCES)), + help="owner-selected preference when converting a private observation", + ) parser.add_argument("--output", type=Path, required=True, help="advisory recommendation JSON") args = parser.parse_args(argv) try: - recommendation = compose_long_horizon_risk_recommendation( - parse_risk_composer_input_json(args.input.read_text(encoding="utf-8")) - ) + if args.input is not None: + if args.risk_preference is not None: + _fail("--risk-preference is only valid with --observation") + composer_input = parse_risk_composer_input_json(args.input.read_text(encoding="utf-8")) + else: + if args.risk_preference is None: + _fail("--observation requires --risk-preference") + composer_input = build_risk_composer_input_from_observation( + parse_risk_composer_input_json(args.observation.read_text(encoding="utf-8")), + risk_preference=args.risk_preference, + ) + recommendation = compose_long_horizon_risk_recommendation(composer_input) validated = validate_risk_composer_recommendation(recommendation) args.output.write_text( json.dumps(validated, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n", diff --git a/python/tests/test_long_horizon_risk_composer.py b/python/tests/test_long_horizon_risk_composer.py index 04cda5a..3a17324 100644 --- a/python/tests/test_long_horizon_risk_composer.py +++ b/python/tests/test_long_horizon_risk_composer.py @@ -4,6 +4,7 @@ import importlib.util import json import sys +import tempfile import unittest from pathlib import Path @@ -44,6 +45,7 @@ def _input(self, *, preference: str = "BALANCED_COMPOUNDING") -> dict[str, objec { "scenario_id": f"soxl_soxx_longterm_{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), } @@ -74,6 +76,23 @@ def _input(self, *, preference: str = "BALANCED_COMPOUNDING") -> dict[str, objec value["input_sha256"] = composer.calculate_risk_composer_input_sha256(value) return value + def _observation(self) -> dict[str, object]: + composer_input = self._input() + value: dict[str, object] = { + "schema": "qsl.long_horizon_risk_observation.v1", + "candidate": composer_input["candidate"], + "source_evidence": composer_input["source_evidence"], + "benchmark": { + "benchmark_id": "soxx", + "benchmark_kind": "unlevered_reference", + "sessions_per_year": 252, + }, + "scenario_paths": composer_input["scenario_paths"], + "observation_sha256": "", + } + value["observation_sha256"] = composer.calculate_risk_observation_sha256(value) + return value + def test_balanced_composer_selects_the_highest_robust_growth_scale_within_benchmark_drawdown_envelope(self): recommendation = composer.compose_long_horizon_risk_recommendation(self._input()) @@ -120,8 +139,17 @@ def test_short_or_duplicate_scenarios_cannot_supply_a_long_horizon_recommendatio short = self._input() short["scenario_paths"][0]["strategy_returns_bps"] = [10] * 251 short["scenario_paths"][0]["benchmark_returns_bps"] = [8] * 251 + short["scenario_paths"][0]["session_count"] = 252 short["input_sha256"] = composer.calculate_risk_composer_input_sha256(short) recommendation = composer.compose_long_horizon_risk_recommendation(short) + self.assertEqual(recommendation["status"], "ADVISORY_RECOMMENDATION_READY") + + insufficient_sessions = self._input() + insufficient_sessions["scenario_paths"][0]["strategy_returns_bps"] = [10] * 250 + insufficient_sessions["scenario_paths"][0]["benchmark_returns_bps"] = [8] * 250 + insufficient_sessions["scenario_paths"][0]["session_count"] = 251 + insufficient_sessions["input_sha256"] = composer.calculate_risk_composer_input_sha256(insufficient_sessions) + recommendation = composer.compose_long_horizon_risk_recommendation(insufficient_sessions) self.assertEqual(recommendation["status"], "PARKED") self.assertIn("LONG_HORIZON_SESSION_COVERAGE_INCOMPLETE", recommendation["reason_codes"]) @@ -144,6 +172,59 @@ def test_tampering_with_evidence_or_smuggling_capital_fails_closed(self): with self.assertRaisesRegex(composer.LongHorizonRiskComposerError, "capital_amount is forbidden"): composer.compose_long_horizon_risk_recommendation(unsafe) + def test_private_observation_needs_an_explicit_preference_before_composition(self): + observation = self._observation() + validated = composer.validate_long_horizon_risk_observation(observation) + composer_input = composer.build_risk_composer_input_from_observation( + validated, + risk_preference="CAPITAL_PRESERVATION", + ) + + self.assertEqual(composer_input["objective"]["risk_preference"], "CAPITAL_PRESERVATION") + self.assertEqual(composer_input["objective"]["benchmark_id"], "soxx") + self.assertEqual(composer_input, composer.validate_risk_composer_input(composer_input)) + self.assertEqual( + composer.compose_long_horizon_risk_recommendation(composer_input)["status"], + "ADVISORY_RECOMMENDATION_READY", + ) + + tampered = copy.deepcopy(observation) + tampered["benchmark"]["benchmark_id"] = "qqq" + with self.assertRaisesRegex(composer.LongHorizonRiskComposerError, "observation_sha256 mismatch"): + composer.build_risk_composer_input_from_observation( + tampered, + risk_preference="BALANCED_COMPOUNDING", + ) + + def test_cli_accepts_private_observation_but_not_an_implicit_preference(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + observation = root / "observation.json" + output = root / "recommendation.json" + observation.write_text(json.dumps(self._observation()), encoding="utf-8") + + self.assertEqual( + composer.main(["--observation", str(observation), "--output", str(output)]), + 1, + ) + self.assertFalse(output.exists()) + self.assertEqual( + composer.main( + [ + "--observation", + str(observation), + "--risk-preference", + "GROWTH_COMPOUNDING", + "--output", + str(output), + ] + ), + 0, + ) + result = json.loads(output.read_text(encoding="utf-8")) + self.assertEqual(result["objective"]["risk_preference"], "GROWTH_COMPOUNDING") + self.assertNotIn("strategy_returns", json.dumps(result)) + def test_recommendation_digest_binds_the_frontier_and_prevents_policy_promotion_by_mutation(self): recommendation = composer.compose_long_horizon_risk_recommendation(self._input()) tampered = copy.deepcopy(recommendation)