feat(risk): bind owner preference after private P3 ingress

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

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 的使用方式

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

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)