From dbaf056fb86a4af210a3ab0693ff05d78205e84d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 26 Aug 2026 21:41:25 +0800 Subject: [PATCH] feat: harden forward observation lifecycle evidence Co-Authored-By: Codex --- ...ward_observation_runtime_contract.zh-CN.md | 8 +- .../strategy_lifecycle/__init__.py | 20 ++ .../strategy_lifecycle/forward_observation.py | 166 +++++++++- .../forward_observation_receipt.py | 300 ++++++++++++++++++ tests/test_forward_observation.py | 59 +++- tests/test_forward_observation_receipt.py | 130 ++++++++ 6 files changed, 662 insertions(+), 21 deletions(-) create mode 100644 src/quant_platform_kit/strategy_lifecycle/forward_observation_receipt.py create mode 100644 tests/test_forward_observation_receipt.py diff --git a/docs/forward_observation_runtime_contract.zh-CN.md b/docs/forward_observation_runtime_contract.zh-CN.md index 33edc02e..02bae90d 100644 --- 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 index ef8a8797..d1d6156c 100644 --- 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,12 +165,15 @@ "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", @@ -162,6 +181,7 @@ "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 index 92dfd0c3..7407e82e 100644 --- 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, ...]: diff --git a/src/quant_platform_kit/strategy_lifecycle/forward_observation_receipt.py b/src/quant_platform_kit/strategy_lifecycle/forward_observation_receipt.py new file mode 100644 index 00000000..0f9d2ef2 --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/forward_observation_receipt.py @@ -0,0 +1,300 @@ +"""Immutable, no-order receipts for candidate-bound forward observation. + +The receipt binds one observed session to the exact frozen candidate policy +and the digests of its evidence dependencies. It is intentionally local and +pure: persistence remains an adapter concern, and this module has no broker, +runtime-target, deployment, account, or market-data dependency. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import date +from hashlib import sha256 +import json +import re +from typing import Any + +from .forward_observation import ForwardObservationPolicy + + +FORWARD_OBSERVATION_RECEIPT_SCHEMA_VERSION = "forward_observation_receipt.v1" +FORWARD_OBSERVATION_DEPENDENCY_DIGESTS = frozenset( + { + "p1_manifest", + "p2_config", + "p3_evidence", + "risk_policy", + "strategy_release", + "plugin_bundle", + } +) +_FROZEN_CHAIN_DEPENDENCY_DIGESTS = frozenset( + {"p2_config", "p3_evidence", "risk_policy", "strategy_release", "plugin_bundle"} +) +FORWARD_OBSERVATION_EVIDENCE_MODES = frozenset( + {"shadow_decision", "simulated_replay", "broker_paper"} +) + +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_TOP_LEVEL_FIELDS = frozenset( + { + "schema_version", + "candidate_id", + "policy_sha256", + "observation_session", + "observation_index", + "previous_receipt_sha256", + "dependency_digests", + "evidence_modes", + "receipt_sha256", + } +) + + +class InvalidForwardObservationReceipt(ValueError): + """Raised when a forward-observation receipt cannot be trusted.""" + + +def _invalid(message: str) -> None: + raise InvalidForwardObservationReceipt(message) + + +def _canonical_bytes(value: object) -> bytes: + try: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise InvalidForwardObservationReceipt( + "receipt must contain only canonical JSON values" + ) from exc + + +def canonical_forward_observation_receipt_bytes(value: Mapping[str, object]) -> bytes: + """Return canonical bytes for a validated receipt, including its digest.""" + + return _canonical_bytes(validate_forward_observation_receipt(value)) + + +def forward_observation_receipt_sha256(value: Mapping[str, object]) -> str: + """Return the deterministic SHA-256 identity of a validated receipt.""" + + return str(validate_forward_observation_receipt(value)["receipt_sha256"]) + + +def forward_observation_policy_sha256(policy: ForwardObservationPolicy) -> str: + """Return the immutable identity of the full candidate observation policy.""" + + return sha256(_canonical_bytes(policy.to_dict())).hexdigest() + + +def _text(value: object, field: str) -> str: + if not isinstance(value, str) or not value.strip(): + _invalid(f"{field} must be a non-empty string") + if any(ord(character) < 0x20 or ord(character) == 0x7F for character in value): + _invalid(f"{field} contains a control character") + return value.strip() + + +def _digest(value: object, field: str) -> str: + text = _text(value, field) + if _SHA256.fullmatch(text) is None: + _invalid(f"{field} must be a lowercase SHA-256 digest") + return text + + +def _session(value: object, field: str) -> str: + text = _text(value, field) + try: + return date.fromisoformat(text).isoformat() + except ValueError as exc: + raise InvalidForwardObservationReceipt( + f"{field} must be an ISO-8601 date" + ) from exc + + +def _index(value: object, field: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + _invalid(f"{field} must be a positive integer") + return value + + +def _dependency_digests(value: object) -> dict[str, str]: + if not isinstance(value, Mapping) or set(value) != FORWARD_OBSERVATION_DEPENDENCY_DIGESTS: + _invalid("dependency_digests must contain the complete closed digest set") + return { + key: _digest(value[key], f"dependency_digests.{key}") + for key in sorted(FORWARD_OBSERVATION_DEPENDENCY_DIGESTS) + } + + +def _evidence_modes(value: object) -> tuple[str, ...]: + if isinstance(value, (str, bytes)) or not isinstance(value, Sequence): + _invalid("evidence_modes must be an array") + modes = tuple(_text(item, "evidence_modes").lower() for item in value) + if ( + len(modes) != 2 + or len(set(modes)) != 2 + or set(modes) - FORWARD_OBSERVATION_EVIDENCE_MODES + or "shadow_decision" not in modes + or not ({"simulated_replay", "broker_paper"} & set(modes)) + ): + _invalid( + "evidence_modes must contain shadow_decision and exactly one paper mode" + ) + return modes + + +def _receipt_core(value: Mapping[str, object]) -> dict[str, object]: + return {key: value[key] for key in sorted(_TOP_LEVEL_FIELDS - {"receipt_sha256"})} + + +def build_forward_observation_receipt( + *, + policy: ForwardObservationPolicy, + observation_session: str, + observation_index: int, + dependency_digests: Mapping[str, str], + evidence_modes: Sequence[str], + previous_receipt: Mapping[str, object] | None = None, +) -> dict[str, object]: + """Build one candidate-bound receipt and append it to a verified chain. + + The caller supplies digest identities only. Raw prices, account IDs, + orders, credentials, and execution instructions are intentionally absent. + """ + + session = _session(observation_session, "observation_session") + index = _index(observation_index, "observation_index") + policy_digest = forward_observation_policy_sha256(policy) + normalized_dependencies = _dependency_digests(dependency_digests) + normalized_modes = _evidence_modes(evidence_modes) + if normalized_modes != tuple(policy.non_live_evidence_modes): + _invalid("evidence_modes must exactly match policy.non_live_evidence_modes") + _validate_policy_window(policy, session, index) + + if previous_receipt is None: + if index != 1: + _invalid("first receipt must have observation_index=1") + previous_digest: str | None = None + else: + previous = validate_forward_observation_receipt(previous_receipt, policy=policy) + previous_index = _index(previous["observation_index"], "previous.observation_index") + if index != previous_index + 1: + _invalid("observation_index must increment by one from the previous receipt") + previous_session = _session(previous["observation_session"], "previous.observation_session") + if session <= previous_session: + _invalid("observation_session must advance from the previous receipt") + previous_digest = _digest(previous["receipt_sha256"], "previous.receipt_sha256") + + receipt: dict[str, object] = { + "schema_version": FORWARD_OBSERVATION_RECEIPT_SCHEMA_VERSION, + "candidate_id": policy.candidate_id, + "policy_sha256": policy_digest, + "observation_session": session, + "observation_index": index, + "previous_receipt_sha256": previous_digest, + "dependency_digests": normalized_dependencies, + "evidence_modes": list(normalized_modes), + "receipt_sha256": "", + } + receipt["receipt_sha256"] = sha256(_canonical_bytes(_receipt_core(receipt))).hexdigest() + return validate_forward_observation_receipt(receipt, policy=policy, previous_receipt=previous_receipt) + + +def validate_forward_observation_receipt( + value: Mapping[str, object], + *, + policy: ForwardObservationPolicy | None = None, + previous_receipt: Mapping[str, object] | None = None, +) -> dict[str, object]: + """Validate a receipt and, when provided, its exact policy and predecessor.""" + + if not isinstance(value, Mapping) or set(value) != _TOP_LEVEL_FIELDS: + _invalid("receipt must be a closed object") + if value.get("schema_version") != FORWARD_OBSERVATION_RECEIPT_SCHEMA_VERSION: + _invalid( + f"schema_version must equal {FORWARD_OBSERVATION_RECEIPT_SCHEMA_VERSION}" + ) + candidate_id = _text(value.get("candidate_id"), "candidate_id") + policy_digest = _digest(value.get("policy_sha256"), "policy_sha256") + session = _session(value.get("observation_session"), "observation_session") + index = _index(value.get("observation_index"), "observation_index") + previous_digest = value.get("previous_receipt_sha256") + if previous_digest is not None: + previous_digest = _digest(previous_digest, "previous_receipt_sha256") + elif index != 1: + _invalid("receipt after the first must include previous_receipt_sha256") + dependencies = _dependency_digests(value.get("dependency_digests")) + modes = _evidence_modes(value.get("evidence_modes")) + claimed_digest = _digest(value.get("receipt_sha256"), "receipt_sha256") + if claimed_digest != sha256(_canonical_bytes(_receipt_core(value))).hexdigest(): + _invalid("receipt_sha256 does not match canonical receipt content") + + if policy is not None: + if candidate_id != policy.candidate_id: + _invalid("candidate_id does not match policy") + if policy_digest != forward_observation_policy_sha256(policy): + _invalid("policy_sha256 does not match policy") + if modes != tuple(policy.non_live_evidence_modes): + _invalid("evidence_modes do not match policy") + _validate_policy_window(policy, session, index) + + if previous_receipt is not None: + previous = validate_forward_observation_receipt(previous_receipt, policy=policy) + if previous_digest != previous["receipt_sha256"]: + _invalid("previous_receipt_sha256 does not match the predecessor") + if index != int(previous["observation_index"]) + 1: + _invalid("observation_index does not increment from the predecessor") + if session <= str(previous["observation_session"]): + _invalid("observation_session does not advance from the predecessor") + if candidate_id != previous["candidate_id"] or policy_digest != previous["policy_sha256"]: + _invalid("candidate or policy identity changed within the receipt chain") + previous_dependencies = _dependency_digests(previous["dependency_digests"]) + if any( + dependencies[field] != previous_dependencies[field] + for field in _FROZEN_CHAIN_DEPENDENCY_DIGESTS + ): + _invalid("frozen dependency digest changed within the receipt chain") + + return { + "schema_version": FORWARD_OBSERVATION_RECEIPT_SCHEMA_VERSION, + "candidate_id": candidate_id, + "policy_sha256": policy_digest, + "observation_session": session, + "observation_index": index, + "previous_receipt_sha256": previous_digest, + "dependency_digests": dependencies, + "evidence_modes": list(modes), + "receipt_sha256": claimed_digest, + } + + +def _validate_policy_window( + policy: ForwardObservationPolicy, session: str, index: int +) -> None: + if index > policy.required_trading_sessions: + _invalid("observation_index exceeds the frozen policy window") + if policy.observation_window_type == "fixed": + start = policy.observation_start_session + assert start is not None + if session < start: + _invalid("observation_session precedes the fixed policy window") + + +__all__ = [ + "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", +] diff --git a/tests/test_forward_observation.py b/tests/test_forward_observation.py index 664c6d98..003aa6ff 100644 --- a/tests/test_forward_observation.py +++ b/tests/test_forward_observation.py @@ -17,6 +17,14 @@ def _policy(**changes: object) -> ForwardObservationPolicy: "domain": "us_equity", "benchmark_symbol": "SOXX", "required_trading_sessions": 252, + "review_milestones": (20, 60), + "automatic_non_live_modes": ("shadow", "paper"), + "auto_resume_clean_sessions": 3, + "observation_calendar": "XNYS", + "observation_window_type": "fixed", + "observation_start_session": "2026-08-26", + "window_rationale_ref": "sha256:soxl-v7-forward-window-rationale", + "non_live_evidence_modes": ("shadow_decision", "simulated_replay"), } values.update(changes) return ForwardObservationPolicy(**values) # type: ignore[arg-type] @@ -52,10 +60,10 @@ def test_missing_p3_evidence_parks_without_starting_non_live_modes() -> None: assert result.notifications == ("historical_evidence_required",) -def test_any_operational_or_risk_failure_pauses_both_non_live_modes() -> None: +def test_transient_operational_failure_pauses_both_non_live_modes() -> None: result = evaluate_forward_observation( _policy(), - _snapshot(data_status="stale", paper_status="mismatch", risk_status="blocked"), + _snapshot(data_status="stale", paper_status="mismatch"), ) assert result.state == "PAUSED" @@ -63,11 +71,23 @@ def test_any_operational_or_risk_failure_pauses_both_non_live_modes() -> None: assert set(result.reasons) == { "data_status=stale", "paper_status=mismatch", - "risk_status=blocked", } assert result.live_authority_granted is False +def test_risk_block_requires_human_review_and_never_auto_resumes() -> None: + blocked = evaluate_forward_observation(_policy(), _snapshot(risk_status="blocked")) + still_blocked = evaluate_forward_observation( + _policy(), _snapshot(previous_state="risk_blocked") + ) + + assert blocked.state == "RISK_BLOCKED" + assert blocked.non_live_actions == ("keep_shadow_stopped", "keep_paper_stopped") + assert blocked.notifications == ("forward_observation_risk_blocked",) + assert still_blocked.state == "RISK_BLOCKED" + assert still_blocked.non_live_actions == ("keep_shadow_stopped", "keep_paper_stopped") + + def test_non_live_pause_resumes_automatically_only_after_clean_recovery_window() -> None: waiting = evaluate_forward_observation( _policy(auto_resume_clean_sessions=3), @@ -97,6 +117,7 @@ def test_milestones_and_full_window_never_promote_live() -> None: assert milestone.notifications == ("forward_review_20_sessions",) assert completed.state == "FORWARD_COMPLETE_HUMAN_REVIEW" + assert completed.non_live_actions == ("keep_shadow_stopped", "keep_paper_stopped") assert completed.notifications == ("forward_window_complete_human_live_review_required",) assert completed.live_action == "human_approval_required" assert completed.live_authority_granted is False @@ -112,6 +133,11 @@ def test_each_candidate_supplies_its_own_forward_window_without_soxl_defaults() review_milestones=(15, 42), automatic_non_live_modes=("shadow", "paper"), auto_resume_clean_sessions=2, + observation_calendar="XNYS", + observation_window_type="fixed", + observation_start_session="2026-08-26", + window_rationale_ref="sha256:global-etf-monthly-v1-forward-window-rationale", + non_live_evidence_modes=("shadow_decision", "simulated_replay"), ) result = evaluate_forward_observation( @@ -131,10 +157,37 @@ def test_each_candidate_supplies_its_own_forward_window_without_soxl_defaults() assert result.live_authority_granted is False +@pytest.mark.parametrize( + ("control_status", "expected_state"), + [ + ("manual_hold", "MANUAL_HOLD"), + ("identity_mismatch", "IDENTITY_MISMATCH"), + ("revoked", "REVOKED"), + ("superseded", "SUPERSEDED"), + ], +) +def test_non_transient_control_states_never_auto_resume( + control_status: str, expected_state: str +) -> None: + held = evaluate_forward_observation(_policy(), _snapshot(control_status=control_status)) + persisted = evaluate_forward_observation( + _policy(), _snapshot(previous_state=control_status) + ) + + assert held.state == expected_state + assert held.non_live_actions == ("keep_shadow_stopped", "keep_paper_stopped") + assert persisted.state == expected_state + assert persisted.non_live_actions == ("keep_shadow_stopped", "keep_paper_stopped") + + def test_policy_and_snapshot_reject_ambiguous_configuration() -> None: with pytest.raises(ForwardObservationPolicyError, match="automatic_non_live_modes"): _policy(automatic_non_live_modes=("shadow",)) with pytest.raises(ForwardObservationPolicyError, match="review_milestones"): _policy(review_milestones=(60, 20)) + with pytest.raises(ForwardObservationPolicyError, match="paper mode"): + _policy(non_live_evidence_modes=("shadow_decision", "broker_paper", "simulated_replay")) + with pytest.raises(ForwardObservationPolicyError, match="rolling"): + _policy(observation_window_type="rolling", observation_start_session="2026-08-26") with pytest.raises(ForwardObservationPolicyError, match="cannot exceed"): _snapshot(observations_completed=20, previous_observations_completed=21) diff --git a/tests/test_forward_observation_receipt.py b/tests/test_forward_observation_receipt.py new file mode 100644 index 00000000..f7c15d0b --- /dev/null +++ b/tests/test_forward_observation_receipt.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import copy + +import pytest + +from quant_platform_kit.strategy_lifecycle.forward_observation import ( + ForwardObservationPolicy, +) +from quant_platform_kit.strategy_lifecycle.forward_observation_receipt import ( + FORWARD_OBSERVATION_DEPENDENCY_DIGESTS, + InvalidForwardObservationReceipt, + build_forward_observation_receipt, + forward_observation_policy_sha256, + forward_observation_receipt_sha256, + validate_forward_observation_receipt, +) + + +def _policy(**changes: object) -> ForwardObservationPolicy: + values: dict[str, object] = { + "candidate_id": "global-etf-monthly-v1", + "strategy_profile": "global_etf_rotation", + "domain": "us_equity", + "benchmark_symbol": "ACWI", + "required_trading_sessions": 63, + "review_milestones": (15, 42), + "automatic_non_live_modes": ("shadow", "paper"), + "auto_resume_clean_sessions": 2, + "observation_calendar": "XNYS", + "observation_window_type": "fixed", + "observation_start_session": "2026-08-26", + "window_rationale_ref": "sha256:global-etf-monthly-v1-forward-window-rationale", + "non_live_evidence_modes": ("shadow_decision", "simulated_replay"), + } + values.update(changes) + return ForwardObservationPolicy(**values) # type: ignore[arg-type] + + +def _dependencies() -> dict[str, str]: + return { + field: character * 64 + for field, character in zip(sorted(FORWARD_OBSERVATION_DEPENDENCY_DIGESTS), "abcdef") + } + + +def _receipt(*, previous=None, index: int = 1, session: str = "2026-08-26"): + return build_forward_observation_receipt( + policy=_policy(), + observation_session=session, + observation_index=index, + dependency_digests=_dependencies(), + evidence_modes=("shadow_decision", "simulated_replay"), + previous_receipt=previous, + ) + + +def test_receipt_binds_exact_policy_dependencies_and_sanitized_modes() -> None: + receipt = _receipt() + + assert receipt["policy_sha256"] == forward_observation_policy_sha256(_policy()) + assert receipt["evidence_modes"] == ["shadow_decision", "simulated_replay"] + assert set(receipt["dependency_digests"]) == FORWARD_OBSERVATION_DEPENDENCY_DIGESTS + assert forward_observation_receipt_sha256(receipt) == receipt["receipt_sha256"] + assert all(token not in str(receipt).lower() for token in ("account", "order", "price")) + + +def test_receipts_append_only_with_a_stable_candidate_policy_and_hash_chain() -> None: + first = _receipt() + second = _receipt(previous=first, index=2, session="2026-08-27") + + assert second["previous_receipt_sha256"] == first["receipt_sha256"] + assert validate_forward_observation_receipt(second, policy=_policy(), previous_receipt=first) == second + + tampered = copy.deepcopy(second) + tampered["previous_receipt_sha256"] = "0" * 64 + with pytest.raises(InvalidForwardObservationReceipt, match="receipt_sha256"): + validate_forward_observation_receipt(tampered, policy=_policy(), previous_receipt=first) + + changed_dependencies = _dependencies() + changed_dependencies["p2_config"] = "f" * 64 + with pytest.raises(InvalidForwardObservationReceipt, match="frozen dependency"): + build_forward_observation_receipt( + policy=_policy(), + observation_session="2026-08-27", + observation_index=2, + dependency_digests=changed_dependencies, + evidence_modes=("shadow_decision", "simulated_replay"), + previous_receipt=first, + ) + + +@pytest.mark.parametrize( + ("index", "session", "error"), + [ + (0, "2026-08-26", "positive integer"), + (1, "2026-08-25", "precedes"), + (64, "2026-11-23", "exceeds"), + ], +) +def test_receipt_rejects_invalid_frozen_window(index: int, session: str, error: str) -> None: + with pytest.raises(InvalidForwardObservationReceipt, match=error): + _receipt(index=index, session=session) + + +def test_receipt_rejects_ambiguous_modes_missing_digests_and_content_tampering() -> None: + with pytest.raises(InvalidForwardObservationReceipt, match="exactly match"): + build_forward_observation_receipt( + policy=_policy(), + observation_session="2026-08-26", + observation_index=1, + dependency_digests=_dependencies(), + evidence_modes=("shadow_decision", "broker_paper"), + ) + + missing = _dependencies() + missing.pop("plugin_bundle") + with pytest.raises(InvalidForwardObservationReceipt, match="closed digest set"): + build_forward_observation_receipt( + policy=_policy(), + observation_session="2026-08-26", + observation_index=1, + dependency_digests=missing, + evidence_modes=("shadow_decision", "simulated_replay"), + ) + + tampered = _receipt() + tampered["dependency_digests"]["p1_manifest"] = "f" * 64 + with pytest.raises(InvalidForwardObservationReceipt, match="receipt_sha256"): + validate_forward_observation_receipt(tampered, policy=_policy())