QuantStrategyLab · Pigbibi · Aug 26, 2026 · Aug 26, 2026
diff --git a/docs/forward_observation_runtime_contract.zh-CN.md b/docs/forward_observation_runtime_contract.zh-CN.md
@@ -2,14 +2,16 @@
 
 `quant_platform_kit.strategy_lifecycle.forward_observation` 是策略、插件和平台共用的纯决策层。它不连接券商、不写运行时目标、不部署服务，也不产生订单。
 
-每个冻结候选必须显式提供自己的 `ForwardObservationPolicy`：候选 ID、策略 profile、无杠杆基准、前瞻交易日数量、复核里程碑，以及恢复前需要的连续健康周期。控制器没有“252 天”“20/60 天”或某个策略的隐性默认值；新候选缺少这些字段会被拒绝，不能继承 SOXL 的参数。没有已验证的 P3 历史证据和证据引用时，状态固定为 `PARKED`。
+每个冻结候选必须显式提供自己的 `ForwardObservationPolicy`：候选 ID、策略 profile、无杠杆基准、前瞻交易日数量、复核里程碑、恢复前需要的连续健康周期、市场日历、固定或滚动窗口、窗口起点、窗口理由引用，以及精确的非 Live 证据模式。控制器没有“252 天”“20/60 天”或某个策略的隐性默认值；新候选缺少这些字段会被拒绝，不能继承 SOXL 的参数。没有已验证的 P3 历史证据和证据引用时，状态固定为 `PARKED`。
 
-P3 通过后，控制器可以自动给出 `start_shadow`、`start_paper`、`continue_*` 和在短暂故障恢复后的 `resume_*` 意图。数据过期、Paper/Shadow 不一致或风险门阻断时，它只会给出 `pause_*`，并产生告警；它不会替换数据源、修改参数、授予 IAM 权限或修改仓位。
+P3 通过后，控制器可以自动给出 `start_shadow`、`start_paper`、`continue_*` 和在短暂数据/运行故障恢复后的 `resume_*` 意图。证据模式必须明确为 `shadow_decision + simulated_replay` 或 `shadow_decision + broker_paper`，不能把模拟回放、订单预览和券商 Paper 混称为同一种 Paper。数据过期或 Shadow/Paper 不一致时，才会进入可自动恢复的 `PAUSED`；风险阻断、人工冻结、身份不匹配、撤销或被新候选替代，分别进入不可自动恢复的终止状态。
 
-达到候选自己的完整前瞻窗口（例如 SOXL V7 基于其回测与风险验证采用 252 个交易日）只会进入 `FORWARD_COMPLETE_HUMAN_REVIEW`。不同策略可以采用不同观察期，但必须由冻结候选与回测证据明确证明，不能在运行中自动修改。返回值永久包含：
+达到候选自己的完整前瞻窗口（例如 SOXL V7 基于其回测与风险验证采用 252 个交易日）后，控制器会停止两种非 Live 意图，并进入 `FORWARD_COMPLETE_HUMAN_REVIEW`。不同策略可以采用不同观察期，但必须由冻结候选与回测证据明确证明，不能在运行中自动修改。返回值永久包含：
 
 - `no_order=true`
 - `live_authority_granted=false`
 - `live_action=human_approval_required`
 
 因此调度器可以自动完成非实盘观察、记录和安全暂停/恢复；任何平台适配器都必须把该结果视为非 Live 意图。首次 Live、重新启用 Live、资金扩大和策略参数修改仍须独立人工批准与重新验证。
+
+每个有效观察周期还应输出 `forward_observation_receipt.v1`。收据只保存候选、完整 policy 摘要、观察交易日/序号、前一收据摘要、P1/P2/P3/风控/发布/插件的摘要和精确证据模式；它不保存原始价格、账户、订单或密钥。验证器要求收据连续追加、候选与 policy 一致，且不能越过冻结窗口。
diff --git a/src/quant_platform_kit/strategy_lifecycle/__init__.py b/src/quant_platform_kit/strategy_lifecycle/__init__.py
@@ -63,6 +63,17 @@
     ForwardObservationSnapshot,
     evaluate_forward_observation,
 )
+from quant_platform_kit.strategy_lifecycle.forward_observation_receipt import (
+    FORWARD_OBSERVATION_DEPENDENCY_DIGESTS,
+    FORWARD_OBSERVATION_EVIDENCE_MODES,
+    FORWARD_OBSERVATION_RECEIPT_SCHEMA_VERSION,
+    InvalidForwardObservationReceipt,
+    build_forward_observation_receipt,
+    canonical_forward_observation_receipt_bytes,
+    forward_observation_policy_sha256,
+    forward_observation_receipt_sha256,
+    validate_forward_observation_receipt,
+)
 from quant_platform_kit.strategy_lifecycle.live_candidate_notifications import (
     LiveCandidateNotificationEvent,
     build_live_candidate_notification,
@@ -116,6 +127,9 @@
     "EvidenceGateResult",
     "EvidencePackage",
     "FORWARD_OBSERVATION_POLICY_SCHEMA_VERSION",
+    "FORWARD_OBSERVATION_RECEIPT_SCHEMA_VERSION",
+    "FORWARD_OBSERVATION_DEPENDENCY_DIGESTS",
+    "FORWARD_OBSERVATION_EVIDENCE_MODES",
     "FORWARD_RISK_SCHEMA_VERSION",
     "FORWARD_RISK_TERMINAL_STATUSES",
     "P4_OBSERVATION_MODES",
@@ -126,6 +140,7 @@
     "ForwardObservationPolicy",
     "ForwardObservationPolicyError",
     "ForwardObservationSnapshot",
+    "InvalidForwardObservationReceipt",
     "CANONICAL_LIFECYCLE_STATES",
     "LEGACY_CATALOG_STATUS_MAP",
     "InvalidResearchDriverArtifact",
@@ -140,6 +155,7 @@
     "build_strategy_benchmark_catalog",
     "assess_strategy_release_readiness",
     "build_forward_risk_terminal_artifact",
+    "build_forward_observation_receipt",
     "evaluate_forward_observation",
     "build_nonready_forward_risk_stage",
     "build_nonready_research_stage",
@@ -149,19 +165,23 @@
     "build_research_driver_terminal_artifact",
     "canonical_research_driver_terminal_bytes",
     "canonical_forward_risk_terminal_bytes",
+    "canonical_forward_observation_receipt_bytes",
     "catalog_status_grants_execution_permission",
     "migrate_legacy_lifecycle_status",
     "normalize_catalog_lifecycle_status",
     "require_canonical_lifecycle_write",
     "research_driver_terminal_sha256",
     "forward_risk_terminal_sha256",
+    "forward_observation_policy_sha256",
+    "forward_observation_receipt_sha256",
     "validate_evidence_package",
     "validate_evidence_package_file",
     "validate_evidence_package_v2",
     "validate_optimization_spec",
     "validate_research_spec",
     "validate_research_driver_terminal_artifact",
     "validate_forward_risk_terminal_artifact",
+    "validate_forward_observation_receipt",
     "validate_strategy_spec",
     "validate_strategy_spec_file",
 ]
diff --git a/src/quant_platform_kit/strategy_lifecycle/forward_observation.py b/src/quant_platform_kit/strategy_lifecycle/forward_observation.py
@@ -10,16 +10,40 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import date
 from typing import Any
 
 
 FORWARD_OBSERVATION_POLICY_SCHEMA_VERSION = "forward_observation_policy.v1"
 
 _NON_LIVE_MODES = frozenset({"shadow", "paper"})
+_NON_LIVE_EVIDENCE_MODES = frozenset(
+    {"shadow_decision", "simulated_replay", "broker_paper"}
+)
 _DATA_STATUSES = frozenset({"ready", "stale", "unavailable"})
 _MODE_STATUSES = frozenset({"healthy", "mismatch", "unavailable"})
 _RISK_STATUSES = frozenset({"pass", "blocked"})
-_PREVIOUS_STATES = frozenset({"not_started", "active", "paused", "complete"})
+_WINDOW_TYPES = frozenset({"fixed", "rolling"})
+_CONTROL_STATUSES = frozenset(
+    {"clear", "manual_hold", "identity_mismatch", "revoked", "superseded"}
+)
+_PREVIOUS_STATES = frozenset(
+    {
+        "not_started",
+        "active",
+        "paused",
+        "complete",
+        "manual_hold",
+        "identity_mismatch",
+        "risk_blocked",
+        "revoked",
+        "superseded",
+    }
+)
+_STOPPED_ACTIONS = ("keep_shadow_stopped", "keep_paper_stopped")
+_PERMANENT_PREVIOUS_STATES = frozenset(
+    {"manual_hold", "identity_mismatch", "risk_blocked", "revoked", "superseded"}
+)
 
 
 class ForwardObservationPolicyError(ValueError):
@@ -38,6 +62,17 @@ def _non_negative_int(value: object, label: str) -> int:
     return value
 
 
+def _session_date(value: object, label: str) -> str:
+    if not isinstance(value, str) or not value:
+        raise ForwardObservationPolicyError(f"{label} must be an ISO-8601 date")
+    try:
+        return date.fromisoformat(value).isoformat()
+    except ValueError as exc:
+        raise ForwardObservationPolicyError(
+            f"{label} must be an ISO-8601 date"
+        ) from exc
+
+
 @dataclass(frozen=True)
 class ForwardObservationPolicy:
     """Candidate-specific rules for a no-capital forward-observation window.
@@ -52,12 +87,24 @@ class ForwardObservationPolicy:
     domain: str
     benchmark_symbol: str
     required_trading_sessions: int
-    review_milestones: tuple[int, ...] = (20, 60)
-    automatic_non_live_modes: tuple[str, ...] = ("shadow", "paper")
-    auto_resume_clean_sessions: int = 3
+    review_milestones: tuple[int, ...]
+    automatic_non_live_modes: tuple[str, ...]
+    auto_resume_clean_sessions: int
+    observation_calendar: str
+    observation_window_type: str
+    observation_start_session: str | None
+    window_rationale_ref: str
+    non_live_evidence_modes: tuple[str, ...]
 
     def __post_init__(self) -> None:
-        for field_name in ("candidate_id", "strategy_profile", "domain", "benchmark_symbol"):
+        for field_name in (
+            "candidate_id",
+            "strategy_profile",
+            "domain",
+            "benchmark_symbol",
+            "observation_calendar",
+            "window_rationale_ref",
+        ):
             _required_text(getattr(self, field_name), field_name)
         required = _non_negative_int(
             self.required_trading_sessions, "required_trading_sessions"
@@ -92,6 +139,33 @@ def __post_init__(self) -> None:
             raise ForwardObservationPolicyError(
                 "review_milestones must be positive integers below required_trading_sessions"
             )
+        window_type = _required_text(
+            self.observation_window_type, "observation_window_type"
+        ).lower()
+        if window_type not in _WINDOW_TYPES:
+            raise ForwardObservationPolicyError(
+                "observation_window_type must be fixed or rolling"
+            )
+        if window_type == "fixed":
+            _session_date(self.observation_start_session, "observation_start_session")
+        elif self.observation_start_session is not None:
+            raise ForwardObservationPolicyError(
+                "rolling observation_window_type must not set observation_start_session"
+            )
+
+        evidence_modes = tuple(
+            str(mode).strip().lower() for mode in self.non_live_evidence_modes
+        )
+        if (
+            len(evidence_modes) != 2
+            or len(set(evidence_modes)) != 2
+            or set(evidence_modes) - _NON_LIVE_EVIDENCE_MODES
+            or "shadow_decision" not in evidence_modes
+            or not ({"simulated_replay", "broker_paper"} & set(evidence_modes))
+        ):
+            raise ForwardObservationPolicyError(
+                "non_live_evidence_modes must contain shadow_decision and exactly one paper mode"
+            )
 
     def to_dict(self) -> dict[str, object]:
         return {
@@ -104,6 +178,11 @@ def to_dict(self) -> dict[str, object]:
             "review_milestones": list(self.review_milestones),
             "automatic_non_live_modes": list(self.automatic_non_live_modes),
             "auto_resume_clean_sessions": self.auto_resume_clean_sessions,
+            "observation_calendar": self.observation_calendar,
+            "observation_window_type": self.observation_window_type,
+            "observation_start_session": self.observation_start_session,
+            "window_rationale_ref": self.window_rationale_ref,
+            "non_live_evidence_modes": list(self.non_live_evidence_modes),
             "live_authority_granted": False,
         }
 
@@ -122,6 +201,7 @@ class ForwardObservationSnapshot:
     shadow_status: str = "healthy"
     paper_status: str = "healthy"
     risk_status: str = "pass"
+    control_status: str = "clear"
 
     def __post_init__(self) -> None:
         if not isinstance(self.historical_evidence_verified, bool):
@@ -152,6 +232,8 @@ def __post_init__(self) -> None:
             raise ForwardObservationPolicyError("unsupported paper_status")
         if self.risk_status not in _RISK_STATUSES:
             raise ForwardObservationPolicyError("unsupported risk_status")
+        if self.control_status not in _CONTROL_STATUSES:
+            raise ForwardObservationPolicyError("unsupported control_status")
 
 
 @dataclass(frozen=True)
@@ -209,11 +291,37 @@ def evaluate_forward_observation(
             policy,
             snapshot,
             state="PARKED",
-            actions=("keep_shadow_stopped", "keep_paper_stopped"),
+            actions=_STOPPED_ACTIONS,
             notifications=("historical_evidence_required",),
             reasons=("verified P3 historical evidence is required before P4",),
         )
 
+    controlled = _controlled_stop(snapshot)
+    if controlled is not None:
+        state, notification, reason = controlled
+        return _decision(
+            policy,
+            snapshot,
+            state=state,
+            actions=_STOPPED_ACTIONS,
+            notifications=(notification,) if notification else (),
+            reasons=(reason,),
+        )
+
+    if snapshot.risk_status == "blocked":
+        return _decision(
+            policy,
+            snapshot,
+            state="RISK_BLOCKED",
+            actions=_STOPPED_ACTIONS,
+            notifications=(
+                ("forward_observation_risk_blocked",)
+                if snapshot.previous_state != "risk_blocked"
+                else ()
+            ),
+            reasons=("risk_status=blocked; explicit human review is required",),
+        )
+
     health_reasons = _health_reasons(snapshot)
     if health_reasons:
         return _decision(
@@ -241,6 +349,21 @@ def evaluate_forward_observation(
             ),
         )
 
+    if snapshot.observations_completed >= policy.required_trading_sessions:
+        notifications = list(_crossed_milestones(policy, snapshot))
+        if snapshot.previous_observations_completed < policy.required_trading_sessions:
+            notifications.append("forward_window_complete_human_live_review_required")
+        return _decision(
+            policy,
+            snapshot,
+            state="FORWARD_COMPLETE_HUMAN_REVIEW",
+            actions=_STOPPED_ACTIONS,
+            notifications=tuple(notifications),
+            reasons=(
+                "forward window is complete; non-live observation is stopped and live remains blocked pending explicit human approval",
+            ),
+        )
+
     actions = (
         ("resume_shadow", "resume_paper")
         if snapshot.previous_state == "paused"
@@ -255,13 +378,6 @@ def evaluate_forward_observation(
     reasons = [
         "P3 evidence is verified; non-live shadow and paper observation may run automatically"
     ]
-    if snapshot.observations_completed >= policy.required_trading_sessions:
-        state = "FORWARD_COMPLETE_HUMAN_REVIEW"
-        if snapshot.previous_observations_completed < policy.required_trading_sessions:
-            notifications.append("forward_window_complete_human_live_review_required")
-        reasons.append(
-            "forward window is complete; live remains blocked pending explicit human approval"
-        )
     return _decision(
         policy,
         snapshot,
@@ -280,11 +396,31 @@ def _health_reasons(snapshot: ForwardObservationSnapshot) -> list[str]:
         reasons.append(f"shadow_status={snapshot.shadow_status}")
     if snapshot.paper_status != "healthy":
         reasons.append(f"paper_status={snapshot.paper_status}")
-    if snapshot.risk_status != "pass":
-        reasons.append(f"risk_status={snapshot.risk_status}")
     return reasons
 
 
+def _controlled_stop(
+    snapshot: ForwardObservationSnapshot,
+) -> tuple[str, str | None, str] | None:
+    control = snapshot.control_status
+    if control == "clear" and snapshot.previous_state in _PERMANENT_PREVIOUS_STATES:
+        control = snapshot.previous_state
+    if control == "clear":
+        return None
+    state, notification = {
+        "manual_hold": ("MANUAL_HOLD", "forward_observation_manual_hold"),
+        "identity_mismatch": (
+            "IDENTITY_MISMATCH",
+            "forward_observation_identity_mismatch",
+        ),
+        "revoked": ("REVOKED", "forward_observation_revoked"),
+        "superseded": ("SUPERSEDED", "forward_observation_superseded"),
+        "risk_blocked": ("RISK_BLOCKED", "forward_observation_risk_blocked"),
+    }[control]
+    should_notify = control != snapshot.previous_state
+    return state, notification if should_notify else None, f"control_status={control}"
+
+
 def _crossed_milestones(
     policy: ForwardObservationPolicy, snapshot: ForwardObservationSnapshot
 ) -> tuple[str, ...]: