From f74c87486b00d3aa5368e2a2041fcd5f1621dda9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 27 Aug 2026 15:53:33 +0800 Subject: [PATCH] feat: bind non-live evidence to runtime identity Co-Authored-By: Codex --- ...n_live_execution_evidence_binding.zh-CN.md | 35 ++ .../strategy_lifecycle/__init__.py | 22 + .../non_live_execution_evidence.py | 465 ++++++++++++++++++ tests/test_non_live_execution_evidence.py | 245 +++++++++ 4 files changed, 767 insertions(+) create mode 100644 docs/non_live_execution_evidence_binding.zh-CN.md create mode 100644 src/quant_platform_kit/strategy_lifecycle/non_live_execution_evidence.py create mode 100644 tests/test_non_live_execution_evidence.py diff --git a/docs/non_live_execution_evidence_binding.zh-CN.md b/docs/non_live_execution_evidence_binding.zh-CN.md new file mode 100644 index 0000000..10199b8 --- /dev/null +++ b/docs/non_live_execution_evidence_binding.zh-CN.md @@ -0,0 +1,35 @@ +# 非 Live 运行证据绑定 + +`quant_platform_kit.strategy_lifecycle.non_live_execution_evidence` 定义 +`non_live_execution_evidence_binding.v1`。它把已经存在、但此前分散的三类事实冻结为一条可验证记录： + +- 冻结候选及其不可变候选修订摘要； +- 一个目标平台、一个不暴露账户信息的 `runtime_scope_sha256`、一个平台适配器摘要和一个非 Live 通道； +- 一份精确的策略发布身份，以及同一 P4 前瞻收据和非 Live 证据工件的摘要。 + +这是所有单策略、组合策略与 `plugin_composite` 候选共用的契约；它没有 SOXL、TQQQ 或某个平台的硬编码。`plugin_composite` 是插件参与运行时唯一可用的候选形态，不能把单个插件伪装成独立执行策略。 + +## 冻结关系 + +构建器会校验收据中的 `p2_config`、`p3_evidence`、`risk_policy`、`strategy_release` 和 `plugin_bundle` 摘要，分别精确等于 `StrategyReleaseIdentity` 的 config、evidence、risk、manifest 与 plugin 摘要。这样，同一候选不能把另一个策略版本、不同风险政策或不同插件包混入前瞻观察。 + +`execution_channel` 只能是 `shadow` 或 `paper`，且必须是该候选 policy 已启用的通道。Shadow 需要收据中有 `shadow_decision`；Paper 还必须有 policy 声明的 `simulated_replay` 或 `broker_paper`。任何路径都固定为： + +```json +{ + "no_order": true, + "live_authority_granted": false +} +``` + +因此它不是订单许可、账户许可、调度器或 Live 晋级器。它也不会读取/写入对象存储、Cloud Run、GitHub、市场数据或券商。 + +## 账户与运行范围 + +记录只保存 `runtime_scope_sha256`，不保存账户号、账户 selector、服务名、部署 URL、凭证或券商原始响应。平台在其受控环境内从自己的运行时目标计算该摘要；审计/存储层只按摘要匹配。实际的账户身份核验、Paper 命令范围和订单准入仍由已有的 runtime-target、account-identity、paper-command 和 risk-gate 契约负责。 + +## 当前 Shadow 与未来 Paper + +通用构建器只接受经过调用方验证的 `schema_version + sha256` 证据引用，使未来的 Paper adapter 可以采用自己的、严格验证后的证据 schema。对当前 `paired_shadow_evidence.v1`，平台必须优先使用 `build_paired_shadow_execution_evidence_binding(...)`：它会再验证 paired-shadow 工件、policy 和前瞻收据三者完全一致。 + +平台可以调用 `build_non_live_execution_evidence_report_artifacts(...)`，将 canonical JSON 与摘要放进既有 `runtime_reports.artifacts`。持久化服务仍应以 `(candidate_id, platform_id, runtime_scope_sha256, execution_channel)` 作为隔离/查询键，并用 create-only 或条件追加保存收据、证据与绑定；纯函数不能替代对象存储的原子性和访问控制。 diff --git a/src/quant_platform_kit/strategy_lifecycle/__init__.py b/src/quant_platform_kit/strategy_lifecycle/__init__.py index a5405c3..d0c81f2 100644 --- a/src/quant_platform_kit/strategy_lifecycle/__init__.py +++ b/src/quant_platform_kit/strategy_lifecycle/__init__.py @@ -107,6 +107,18 @@ paired_shadow_evidence_sha256, validate_paired_shadow_evidence, ) +from quant_platform_kit.strategy_lifecycle.non_live_execution_evidence import ( + NON_LIVE_CANDIDATE_SUBJECTS, + NON_LIVE_EXECUTION_CHANNELS, + NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION, + InvalidNonLiveExecutionEvidenceBinding, + build_non_live_execution_evidence_binding, + build_non_live_execution_evidence_report_artifacts, + build_paired_shadow_execution_evidence_binding, + canonical_non_live_execution_evidence_binding_bytes, + non_live_execution_evidence_binding_sha256, + validate_non_live_execution_evidence_binding, +) from quant_platform_kit.strategy_lifecycle.live_candidate_notifications import ( LiveCandidateNotificationEvent, build_live_candidate_notification, @@ -182,6 +194,9 @@ "PAIRED_SHADOW_EVIDENCE_KIND", "PAIRED_SHADOW_EVIDENCE_SCHEMA_VERSION", "PAIRED_SHADOW_LEG_FIELDS", + "NON_LIVE_CANDIDATE_SUBJECTS", + "NON_LIVE_EXECUTION_CHANNELS", + "NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION", "FORWARD_RISK_SCHEMA_VERSION", "FORWARD_RISK_TERMINAL_STATUSES", "P4_OBSERVATION_MODES", @@ -194,6 +209,7 @@ "ForwardObservationSnapshot", "InvalidForwardObservationReceipt", "InvalidPairedShadowEvidence", + "InvalidNonLiveExecutionEvidenceBinding", "CANONICAL_LIFECYCLE_STATES", "LEGACY_CATALOG_STATUS_MAP", "InvalidResearchDriverArtifact", @@ -211,6 +227,9 @@ "build_forward_observation_receipt", "build_paired_shadow_evidence", "build_paired_shadow_evidence_report_artifacts", + "build_non_live_execution_evidence_binding", + "build_non_live_execution_evidence_report_artifacts", + "build_paired_shadow_execution_evidence_binding", "evaluate_forward_observation", "build_nonready_forward_risk_stage", "build_nonready_research_stage", @@ -222,6 +241,7 @@ "canonical_forward_risk_terminal_bytes", "canonical_forward_observation_receipt_bytes", "canonical_paired_shadow_evidence_bytes", + "canonical_non_live_execution_evidence_binding_bytes", "catalog_status_grants_execution_permission", "migrate_legacy_lifecycle_status", "normalize_catalog_lifecycle_status", @@ -231,6 +251,7 @@ "forward_observation_policy_sha256", "forward_observation_receipt_sha256", "paired_shadow_evidence_sha256", + "non_live_execution_evidence_binding_sha256", "validate_evidence_package", "validate_evidence_package_file", "validate_evidence_package_v2", @@ -241,6 +262,7 @@ "validate_forward_risk_terminal_artifact", "validate_forward_observation_receipt", "validate_paired_shadow_evidence", + "validate_non_live_execution_evidence_binding", "validate_source_receipt", "validate_strategy_candidate", "validate_strategy_candidate_v2", diff --git a/src/quant_platform_kit/strategy_lifecycle/non_live_execution_evidence.py b/src/quant_platform_kit/strategy_lifecycle/non_live_execution_evidence.py new file mode 100644 index 0000000..7515a9b --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/non_live_execution_evidence.py @@ -0,0 +1,465 @@ +"""Immutable platform/channel bindings for non-live lifecycle evidence. + +Forward-observation receipts prove that a frozen candidate was observed. +Strategy-release identities prove which strategy, risk, evidence and plugin +bundle a runtime loaded. This module joins those two facts with a *single* +platform runtime scope and a non-live channel. It deliberately contains no +account identifier, broker client, runtime target, storage implementation or +execution instruction. + +The generic binding stores an evidence identity only, so later paper adapters +can use their own verified evidence schema. The paired-shadow helper validates +the currently available ``paired_shadow_evidence.v1`` object before building +the generic binding. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from hashlib import sha256 +import json +import re + +from quant_platform_kit.common.strategy_release import ( + StrategyReleaseIdentity, + build_strategy_release_identity, +) + +from .forward_observation import ForwardObservationPolicy +from .forward_observation_receipt import ( + forward_observation_receipt_sha256, + validate_forward_observation_receipt, +) +from .paired_shadow_evidence import ( + PAIRED_SHADOW_EVIDENCE_SCHEMA_VERSION, + paired_shadow_evidence_sha256, + validate_paired_shadow_evidence, +) + + +NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION = ( + "non_live_execution_evidence_binding.v1" +) +NON_LIVE_CANDIDATE_SUBJECTS = frozenset( + {"strategy", "portfolio", "plugin_composite"} +) +NON_LIVE_EXECUTION_CHANNELS = frozenset({"shadow", "paper"}) + +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_PLATFORM_ID = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$") +_EVIDENCE_SCHEMA_VERSION = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$") +_TOP_LEVEL_FIELDS = frozenset( + { + "schema_version", + "candidate_id", + "candidate_subject", + "candidate_revision_sha256", + "platform_id", + "runtime_scope_sha256", + "platform_adapter_sha256", + "execution_channel", + "strategy_release", + "forward_observation_receipt_sha256", + "non_live_evidence_ref", + "no_order", + "live_authority_granted", + "binding_sha256", + } +) +_EVIDENCE_REF_FIELDS = frozenset({"schema_version", "sha256"}) + + +class InvalidNonLiveExecutionEvidenceBinding(ValueError): + """Raised when a non-live evidence binding cannot be trusted.""" + + +def _invalid(message: str) -> None: + raise InvalidNonLiveExecutionEvidenceBinding(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 InvalidNonLiveExecutionEvidenceBinding( + "binding must contain only canonical JSON values" + ) from exc + + +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 _platform(value: object) -> str: + platform = _text(value, "platform_id").lower() + if _PLATFORM_ID.fullmatch(platform) is None: + _invalid("platform_id must be a lowercase scoped identifier") + return platform + + +def _candidate_subject(value: object) -> str: + subject = _text(value, "candidate_subject").lower() + if subject not in NON_LIVE_CANDIDATE_SUBJECTS: + _invalid("candidate_subject must be strategy, portfolio, or plugin_composite") + return subject + + +def _channel(value: object) -> str: + channel = _text(value, "execution_channel").lower() + if channel not in NON_LIVE_EXECUTION_CHANNELS: + _invalid("execution_channel must be shadow or paper") + return channel + + +def _strategy_release(value: object) -> StrategyReleaseIdentity: + if isinstance(value, StrategyReleaseIdentity): + return value + if not isinstance(value, Mapping): + _invalid("strategy_release must be a closed release identity") + expected = { + "release_id", + "manifest_sha256", + "strategy_revision", + "config_sha256", + "risk_policy_sha256", + "evidence_sha256", + "plugin_bundle_sha256", + "effective_session", + } + if set(value) != expected: + _invalid("strategy_release must be a closed release identity") + try: + return build_strategy_release_identity(value) + except ValueError as exc: + raise InvalidNonLiveExecutionEvidenceBinding( + "strategy_release is invalid" + ) from exc + + +def _evidence_ref(value: object) -> dict[str, str]: + if not isinstance(value, Mapping) or set(value) != _EVIDENCE_REF_FIELDS: + _invalid("non_live_evidence_ref must be a closed evidence identity") + schema_version = _text(value.get("schema_version"), "non_live_evidence_ref.schema_version") + if _EVIDENCE_SCHEMA_VERSION.fullmatch(schema_version) is None: + _invalid("non_live_evidence_ref.schema_version is invalid") + return { + "schema_version": schema_version, + "sha256": _digest(value.get("sha256"), "non_live_evidence_ref.sha256"), + } + + +def _binding_core(value: Mapping[str, object]) -> dict[str, object]: + return { + key: value[key] + for key in sorted(_TOP_LEVEL_FIELDS - {"binding_sha256"}) + } + + +def _validate_release_dependencies( + receipt: Mapping[str, object], release: StrategyReleaseIdentity +) -> None: + dependencies = receipt["dependency_digests"] + assert isinstance(dependencies, Mapping) # validated receipt invariant + expected = { + "p2_config": release.config_sha256, + "p3_evidence": release.evidence_sha256, + "risk_policy": release.risk_policy_sha256, + "strategy_release": release.manifest_sha256, + "plugin_bundle": release.plugin_bundle_sha256, + } + if any(dependencies[key] != digest for key, digest in expected.items()): + _invalid("strategy_release does not match frozen receipt dependencies") + + +def _validate_channel_receipt_evidence( + *, channel: str, policy: ForwardObservationPolicy, receipt: Mapping[str, object] +) -> None: + modes = set(receipt["evidence_modes"]) + if channel not in policy.automatic_non_live_modes: + _invalid("execution_channel is not enabled by the forward-observation policy") + if channel == "shadow" and "shadow_decision" not in modes: + _invalid("shadow binding requires shadow_decision receipt evidence") + if channel == "paper" and not ( + {"simulated_replay", "broker_paper"} & modes + ): + _invalid("paper binding requires configured paper receipt evidence") + + +def build_non_live_execution_evidence_binding( + *, + policy: ForwardObservationPolicy, + forward_observation_receipt: Mapping[str, object], + candidate_subject: str, + candidate_revision_sha256: str, + platform_id: str, + runtime_scope_sha256: str, + platform_adapter_sha256: str, + execution_channel: str, + strategy_release: StrategyReleaseIdentity | Mapping[str, object], + non_live_evidence_schema_version: str, + non_live_evidence_sha256: str, +) -> dict[str, object]: + """Build a no-order evidence binding for one platform and non-live channel. + + ``runtime_scope_sha256`` is an opaque digest owned by the platform. Raw + account selectors, service names and broker identifiers must not enter this + artifact. The generic evidence reference can represent a future paper + evidence schema; callers must validate that schema before passing its + identity here. Use the paired-shadow helper below when the evidence is a + ``paired_shadow_evidence.v1`` artifact. + """ + + receipt = validate_forward_observation_receipt( + forward_observation_receipt, policy=policy + ) + release = _strategy_release(strategy_release) + channel = _channel(execution_channel) + _validate_channel_receipt_evidence(channel=channel, policy=policy, receipt=receipt) + _validate_release_dependencies(receipt, release) + evidence_ref = _evidence_ref( + { + "schema_version": non_live_evidence_schema_version, + "sha256": non_live_evidence_sha256, + } + ) + binding: dict[str, object] = { + "schema_version": NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION, + "candidate_id": policy.candidate_id, + "candidate_subject": _candidate_subject(candidate_subject), + "candidate_revision_sha256": _digest( + candidate_revision_sha256, "candidate_revision_sha256" + ), + "platform_id": _platform(platform_id), + "runtime_scope_sha256": _digest(runtime_scope_sha256, "runtime_scope_sha256"), + "platform_adapter_sha256": _digest( + platform_adapter_sha256, "platform_adapter_sha256" + ), + "execution_channel": channel, + "strategy_release": release.to_dict(), + "forward_observation_receipt_sha256": receipt["receipt_sha256"], + "non_live_evidence_ref": evidence_ref, + "no_order": True, + "live_authority_granted": False, + "binding_sha256": "", + } + binding["binding_sha256"] = sha256(_canonical_bytes(_binding_core(binding))).hexdigest() + return validate_non_live_execution_evidence_binding( + binding, + policy=policy, + forward_observation_receipt=forward_observation_receipt, + strategy_release=strategy_release, + ) + + +def build_paired_shadow_execution_evidence_binding( + *, + policy: ForwardObservationPolicy, + forward_observation_receipt: Mapping[str, object], + paired_shadow_evidence: Mapping[str, object], + candidate_subject: str, + candidate_revision_sha256: str, + platform_id: str, + runtime_scope_sha256: str, + platform_adapter_sha256: str, + strategy_release: StrategyReleaseIdentity | Mapping[str, object], +) -> dict[str, object]: + """Build a shadow binding after verifying a paired-shadow evidence object.""" + + receipt = validate_forward_observation_receipt( + forward_observation_receipt, policy=policy + ) + evidence = validate_paired_shadow_evidence( + paired_shadow_evidence, + policy=policy, + forward_observation_receipt=receipt, + ) + return build_non_live_execution_evidence_binding( + policy=policy, + forward_observation_receipt=receipt, + candidate_subject=candidate_subject, + candidate_revision_sha256=candidate_revision_sha256, + platform_id=platform_id, + runtime_scope_sha256=runtime_scope_sha256, + platform_adapter_sha256=platform_adapter_sha256, + execution_channel="shadow", + strategy_release=strategy_release, + non_live_evidence_schema_version=PAIRED_SHADOW_EVIDENCE_SCHEMA_VERSION, + non_live_evidence_sha256=str(evidence["paired_shadow_evidence_sha256"]), + ) + + +def validate_non_live_execution_evidence_binding( + value: Mapping[str, object], + *, + policy: ForwardObservationPolicy | None = None, + forward_observation_receipt: Mapping[str, object] | None = None, + strategy_release: StrategyReleaseIdentity | Mapping[str, object] | None = None, + paired_shadow_evidence: Mapping[str, object] | None = None, +) -> dict[str, object]: + """Validate a binding and optional primary artifacts supplied by its owner. + + Validation stays pure and cannot resolve storage. Therefore callers must + supply the receipt/release/evidence objects they resolved from their + create-only stores when proving an end-to-end runtime observation. + """ + + if not isinstance(value, Mapping) or set(value) != _TOP_LEVEL_FIELDS: + _invalid("binding must be a closed object") + if value.get("schema_version") != NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION: + _invalid( + "schema_version must equal " + f"{NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION}" + ) + candidate_id = _text(value.get("candidate_id"), "candidate_id") + subject = _candidate_subject(value.get("candidate_subject")) + candidate_revision = _digest( + value.get("candidate_revision_sha256"), "candidate_revision_sha256" + ) + platform = _platform(value.get("platform_id")) + scope_digest = _digest(value.get("runtime_scope_sha256"), "runtime_scope_sha256") + adapter_digest = _digest( + value.get("platform_adapter_sha256"), "platform_adapter_sha256" + ) + channel = _channel(value.get("execution_channel")) + release = _strategy_release(value.get("strategy_release")) + receipt_digest = _digest( + value.get("forward_observation_receipt_sha256"), + "forward_observation_receipt_sha256", + ) + evidence_ref = _evidence_ref(value.get("non_live_evidence_ref")) + if value.get("no_order") is not True: + _invalid("no_order must be true") + if value.get("live_authority_granted") is not False: + _invalid("live_authority_granted must be false") + claimed_digest = _digest(value.get("binding_sha256"), "binding_sha256") + if claimed_digest != sha256(_canonical_bytes(_binding_core(value))).hexdigest(): + _invalid("binding_sha256 does not match canonical binding content") + + if forward_observation_receipt is not None: + receipt = validate_forward_observation_receipt( + forward_observation_receipt, policy=policy + ) + if receipt_digest != forward_observation_receipt_sha256(receipt): + _invalid("forward_observation_receipt_sha256 does not match receipt") + if candidate_id != receipt["candidate_id"]: + _invalid("candidate_id does not match forward-observation receipt") + _validate_release_dependencies(receipt, release) + if policy is not None: + _validate_channel_receipt_evidence( + channel=channel, policy=policy, receipt=receipt + ) + elif policy is not None: + _invalid("policy validation requires forward_observation_receipt") + + if strategy_release is not None: + expected_release = _strategy_release(strategy_release) + if release != expected_release: + _invalid("strategy_release does not match expected release identity") + + if paired_shadow_evidence is not None: + if forward_observation_receipt is None or policy is None: + _invalid("paired-shadow validation requires policy and forward_observation_receipt") + evidence = validate_paired_shadow_evidence( + paired_shadow_evidence, + policy=policy, + forward_observation_receipt=forward_observation_receipt, + ) + if channel != "shadow": + _invalid("paired-shadow evidence can bind only to the shadow channel") + if evidence_ref["schema_version"] != PAIRED_SHADOW_EVIDENCE_SCHEMA_VERSION: + _invalid("paired-shadow evidence schema does not match evidence reference") + if evidence_ref["sha256"] != paired_shadow_evidence_sha256(evidence): + _invalid("paired-shadow evidence digest does not match evidence reference") + if candidate_id != evidence["candidate_id"]: + _invalid("candidate_id does not match paired-shadow evidence") + + return { + "schema_version": NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION, + "candidate_id": candidate_id, + "candidate_subject": subject, + "candidate_revision_sha256": candidate_revision, + "platform_id": platform, + "runtime_scope_sha256": scope_digest, + "platform_adapter_sha256": adapter_digest, + "execution_channel": channel, + "strategy_release": release.to_dict(), + "forward_observation_receipt_sha256": receipt_digest, + "non_live_evidence_ref": evidence_ref, + "no_order": True, + "live_authority_granted": False, + "binding_sha256": claimed_digest, + } + + +def canonical_non_live_execution_evidence_binding_bytes( + value: Mapping[str, object] +) -> bytes: + """Return canonical bytes for a schema-valid non-live evidence binding.""" + + return _canonical_bytes(validate_non_live_execution_evidence_binding(value)) + + +def non_live_execution_evidence_binding_sha256(value: Mapping[str, object]) -> str: + """Return the deterministic identity of a validated non-live binding.""" + + return str( + validate_non_live_execution_evidence_binding(value)["binding_sha256"] + ) + + +def build_non_live_execution_evidence_report_artifacts( + binding: Mapping[str, object], + **validation_context: object, +) -> dict[str, object]: + """Return a platform-neutral runtime-report attachment for one binding. + + The helper is intentionally serialization-only: it does not resolve a + runtime target, persist data, submit orders, or grant any live authority. + """ + + validated = validate_non_live_execution_evidence_binding( + binding, + **validation_context, # type: ignore[arg-type] + ) + return { + "non_live_execution_evidence_binding_json": ( + canonical_non_live_execution_evidence_binding_bytes(validated).decode("utf-8") + ), + "non_live_execution_evidence_binding_sha256": validated["binding_sha256"], + "non_live_execution_evidence_binding_schema_version": ( + NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION + ), + "non_live_execution_evidence_binding_no_order": True, + "non_live_execution_evidence_binding_live_authority_granted": False, + } + + +__all__ = [ + "NON_LIVE_CANDIDATE_SUBJECTS", + "NON_LIVE_EXECUTION_CHANNELS", + "NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION", + "InvalidNonLiveExecutionEvidenceBinding", + "build_non_live_execution_evidence_binding", + "build_non_live_execution_evidence_report_artifacts", + "build_paired_shadow_execution_evidence_binding", + "canonical_non_live_execution_evidence_binding_bytes", + "non_live_execution_evidence_binding_sha256", + "validate_non_live_execution_evidence_binding", +] diff --git a/tests/test_non_live_execution_evidence.py b/tests/test_non_live_execution_evidence.py new file mode 100644 index 0000000..affe7f0 --- /dev/null +++ b/tests/test_non_live_execution_evidence.py @@ -0,0 +1,245 @@ +from __future__ import annotations + +import copy +import json + +import pytest + +from quant_platform_kit.common.runtime_reports import build_runtime_report_base +from quant_platform_kit.common.strategy_release import build_strategy_release_identity +from quant_platform_kit.strategy_lifecycle.forward_observation import ( + ForwardObservationPolicy, +) +from quant_platform_kit.strategy_lifecycle.forward_observation_receipt import ( + FORWARD_OBSERVATION_DEPENDENCY_DIGESTS, + build_forward_observation_receipt, +) +from quant_platform_kit.strategy_lifecycle.non_live_execution_evidence import ( + NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION, + InvalidNonLiveExecutionEvidenceBinding, + build_non_live_execution_evidence_binding, + build_non_live_execution_evidence_report_artifacts, + build_paired_shadow_execution_evidence_binding, + canonical_non_live_execution_evidence_binding_bytes, + non_live_execution_evidence_binding_sha256, + validate_non_live_execution_evidence_binding, +) +from quant_platform_kit.strategy_lifecycle.paired_shadow_evidence import ( + build_paired_shadow_evidence, +) + + +def _policy(**changes: object) -> ForwardObservationPolicy: + values: dict[str, object] = { + "candidate_id": "soxl-v7-volatility-budget", + "strategy_profile": "soxl_tactical", + "domain": "us_equity", + "benchmark_symbol": "SOXX", + "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:soxl-v7-forward-window-rationale", + "non_live_evidence_modes": ("shadow_decision", "simulated_replay"), + } + values.update(changes) + return ForwardObservationPolicy(**values) # type: ignore[arg-type] + + +def _release() -> dict[str, str]: + return { + "release_id": "soxl-v7.20260826", + "manifest_sha256": "a" * 64, + "strategy_revision": "soxl-v7", + "config_sha256": "b" * 64, + "risk_policy_sha256": "c" * 64, + "evidence_sha256": "d" * 64, + "plugin_bundle_sha256": "e" * 64, + "effective_session": "2026-08-26", + } + + +def _dependencies() -> dict[str, str]: + release = _release() + dependencies = { + "p1_manifest": "f" * 64, + "p2_config": release["config_sha256"], + "p3_evidence": release["evidence_sha256"], + "risk_policy": release["risk_policy_sha256"], + "strategy_release": release["manifest_sha256"], + "plugin_bundle": release["plugin_bundle_sha256"], + } + assert set(dependencies) == FORWARD_OBSERVATION_DEPENDENCY_DIGESTS + return dependencies + + +def _receipt(*, policy: ForwardObservationPolicy | None = None) -> dict[str, object]: + active_policy = policy or _policy() + return build_forward_observation_receipt( + policy=active_policy, + observation_session="2026-08-26", + observation_index=1, + dependency_digests=_dependencies(), + evidence_modes=active_policy.non_live_evidence_modes, + ) + + +def _leg(source: str) -> dict[str, object]: + return { + "signal": {"source": source}, + "hypothetical_order": {"source": source}, + "position": {"source": source}, + "cost": {"source": source}, + "return": {"source": source}, + } + + +def _paired_evidence(*, receipt: dict[str, object] | None = None) -> dict[str, object]: + active_receipt = receipt or _receipt() + return build_paired_shadow_evidence( + policy=_policy(), + forward_observation_receipt=active_receipt, + baseline_id="soxl-v6-baseline", + observed_at="2026-08-26T20:00:00-04:00", + input_snapshot_sha256="9" * 64, + candidate=_leg("candidate"), + baseline=_leg("baseline"), + ) + + +def _binding(**changes: object) -> dict[str, object]: + receipt = _receipt() + values: dict[str, object] = { + "policy": _policy(), + "forward_observation_receipt": receipt, + "candidate_subject": "strategy", + "candidate_revision_sha256": "7" * 64, + "platform_id": "longbridge_sg", + "runtime_scope_sha256": "8" * 64, + "platform_adapter_sha256": "6" * 64, + "execution_channel": "shadow", + "strategy_release": _release(), + "non_live_evidence_schema_version": "shadow_observation.v1", + "non_live_evidence_sha256": "5" * 64, + } + values.update(changes) + return build_non_live_execution_evidence_binding(**values) # type: ignore[arg-type] + + +def test_generic_binding_joins_candidate_platform_channel_and_release_identity() -> None: + binding = _binding() + + assert binding["schema_version"] == NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION + assert binding["candidate_subject"] == "strategy" + assert binding["platform_id"] == "longbridge_sg" + assert binding["execution_channel"] == "shadow" + assert binding["strategy_release"] == _release() + assert binding["no_order"] is True + assert binding["live_authority_granted"] is False + assert non_live_execution_evidence_binding_sha256(binding) == binding["binding_sha256"] + assert canonical_non_live_execution_evidence_binding_bytes(binding) + assert ( + validate_non_live_execution_evidence_binding( + binding, + policy=_policy(), + forward_observation_receipt=_receipt(), + strategy_release=_release(), + ) + == binding + ) + serialized = json.dumps(binding, sort_keys=True) + assert "account_selector" not in serialized + assert "service_name" not in serialized + + +def test_generic_binding_accepts_the_shared_release_identity_object() -> None: + binding = _binding(strategy_release=build_strategy_release_identity(_release())) + + assert binding["strategy_release"] == _release() + + +@pytest.mark.parametrize("candidate_subject", ("strategy", "portfolio", "plugin_composite")) +def test_generic_binding_supports_each_candidate_shape(candidate_subject: str) -> None: + assert _binding(candidate_subject=candidate_subject)["candidate_subject"] == candidate_subject + + +def test_paired_shadow_helper_verifies_current_evidence_and_report_attachment() -> None: + receipt = _receipt() + evidence = _paired_evidence(receipt=receipt) + binding = build_paired_shadow_execution_evidence_binding( + policy=_policy(), + forward_observation_receipt=receipt, + paired_shadow_evidence=evidence, + candidate_subject="plugin_composite", + candidate_revision_sha256="7" * 64, + platform_id="longbridge_sg", + runtime_scope_sha256="8" * 64, + platform_adapter_sha256="6" * 64, + strategy_release=_release(), + ) + artifacts = build_non_live_execution_evidence_report_artifacts( + binding, + policy=_policy(), + forward_observation_receipt=receipt, + strategy_release=_release(), + paired_shadow_evidence=evidence, + ) + report = build_runtime_report_base( + platform="longbridge", + deploy_target="non-live", + service_name="shadow-observer", + strategy_profile="soxl_tactical", + run_id="shadow-001", + run_source="shadow", + dry_run=True, + artifacts=artifacts, + ) + + assert binding["execution_channel"] == "shadow" + assert binding["non_live_evidence_ref"]["sha256"] == evidence[ + "paired_shadow_evidence_sha256" + ] + assert json.loads(artifacts["non_live_execution_evidence_binding_json"]) == binding + assert artifacts["non_live_execution_evidence_binding_no_order"] is True + assert artifacts["non_live_execution_evidence_binding_live_authority_granted"] is False + assert report["runtime_target"] == {} + + +def test_binding_fails_closed_for_release_or_channel_mismatches() -> None: + changed_release = copy.deepcopy(_release()) + changed_release["config_sha256"] = "0" * 64 + with pytest.raises(InvalidNonLiveExecutionEvidenceBinding, match="frozen receipt"): + _binding(strategy_release=changed_release) + + shadow_only = _policy( + automatic_non_live_modes=("shadow",), + non_live_evidence_modes=("shadow_decision",), + ) + with pytest.raises(InvalidNonLiveExecutionEvidenceBinding, match="not enabled"): + _binding( + policy=shadow_only, + forward_observation_receipt=_receipt(policy=shadow_only), + execution_channel="paper", + ) + + +def test_paired_shadow_proof_rejects_an_unrelated_evidence_digest() -> None: + receipt = _receipt() + evidence = _paired_evidence(receipt=receipt) + binding = _binding( + forward_observation_receipt=receipt, + non_live_evidence_schema_version="paired_shadow_evidence.v1", + non_live_evidence_sha256="0" * 64, + ) + + with pytest.raises(InvalidNonLiveExecutionEvidenceBinding, match="does not match evidence reference"): + validate_non_live_execution_evidence_binding( + binding, + policy=_policy(), + forward_observation_receipt=receipt, + strategy_release=_release(), + paired_shadow_evidence=evidence, + )