Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/non_live_execution_evidence_binding.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@

## 账户与运行范围

记录只保存 `runtime_scope_sha256`,不保存账户号、账户 selector、服务名、部署 URL、凭证或券商原始响应。平台在其受控环境内从自己的运行时目标计算该摘要;审计/存储层只按摘要匹配。实际的账户身份核验、Paper 命令范围和订单准入仍由已有的 runtime-target、account-identity、paper-command 和 risk-gate 契约负责。
记录只保存 `runtime_scope_sha256`,不保存账户号、账户 selector、服务名、部署 URL、凭证或券商原始响应。平台应在其受控环境内调用 `non_live_runtime_scope_sha256(runtime_target=..., execution_channel=...)`:它只把 platform、部署/账户/service scope 和执行环境作为哈希材料,返回值不会包含原字段。它拒绝 Live target;Shadow 还必须使用 `dry_run` target,并且所有目标都必须有一个明确的非实盘 scope selector。审计/存储层只按摘要匹配。实际的账户身份核验、Paper 命令范围和订单准入仍由已有的 runtime-target、account-identity、paper-command 和 risk-gate 契约负责。

## 当前 Shadow 与未来 Paper

Expand Down
4 changes: 4 additions & 0 deletions src/quant_platform_kit/strategy_lifecycle/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -111,12 +111,14 @@
NON_LIVE_CANDIDATE_SUBJECTS,
NON_LIVE_EXECUTION_CHANNELS,
NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION,
NON_LIVE_RUNTIME_SCOPE_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,
non_live_runtime_scope_sha256,
validate_non_live_execution_evidence_binding,
)
from quant_platform_kit.strategy_lifecycle.live_candidate_notifications import (
Expand Down Expand Up @@ -197,6 +199,7 @@
"NON_LIVE_CANDIDATE_SUBJECTS",
"NON_LIVE_EXECUTION_CHANNELS",
"NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION",
"NON_LIVE_RUNTIME_SCOPE_SCHEMA_VERSION",
"FORWARD_RISK_SCHEMA_VERSION",
"FORWARD_RISK_TERMINAL_STATUSES",
"P4_OBSERVATION_MODES",
Expand Down Expand Up @@ -252,6 +255,7 @@
"forward_observation_receipt_sha256",
"paired_shadow_evidence_sha256",
"non_live_execution_evidence_binding_sha256",
"non_live_runtime_scope_sha256",
"validate_evidence_package",
"validate_evidence_package_file",
"validate_evidence_package_v2",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,10 @@
StrategyReleaseIdentity,
build_strategy_release_identity,
)
from quant_platform_kit.common.runtime_target import (
RuntimeExecutionEnvironment,
RuntimeTarget,
)

from .forward_observation import ForwardObservationPolicy
from .forward_observation_receipt import (
Expand All @@ -40,6 +44,7 @@
NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION = (
"non_live_execution_evidence_binding.v1"
)
NON_LIVE_RUNTIME_SCOPE_SCHEMA_VERSION = "non_live_runtime_scope.v1"
NON_LIVE_CANDIDATE_SUBJECTS = frozenset(
{"strategy", "portfolio", "plugin_composite"}
)
Expand Down Expand Up @@ -128,6 +133,69 @@ def _channel(value: object) -> str:
return channel


def _scope_component(value: object, field: str) -> str | None:
if value is None:
return None
text = _text(value, field)
return text


def non_live_runtime_scope_sha256(
*, runtime_target: RuntimeTarget, execution_channel: str
) -> str:
"""Derive an opaque scope identity from one non-live runtime target.

The returned digest is the only value suitable for a durable non-live
evidence binding. Selectors are deliberately used only as hash material:
the function never returns them, logs them, or attaches them to a report.
A funded ``live`` target cannot be converted into non-live evidence.
"""

if not isinstance(runtime_target, RuntimeTarget):
_invalid("runtime_target must be a RuntimeTarget")
channel = _channel(execution_channel)
environment = runtime_target.execution_environment
if environment is RuntimeExecutionEnvironment.LIVE:
_invalid("non-live runtime scope cannot use a live execution target")
if (
channel == "shadow"
and environment is not RuntimeExecutionEnvironment.DRY_RUN
):
_invalid("shadow runtime scope requires a dry_run execution target")
platform = _platform(runtime_target.platform_id)
account_selector = tuple(
_scope_component(value, "runtime_target.account_selector[]")
for value in runtime_target.account_selector
)
material = {
"schema_version": NON_LIVE_RUNTIME_SCOPE_SCHEMA_VERSION,
"platform_id": platform,
"execution_channel": channel,
"execution_environment": environment.value,
"deployment_selector": _scope_component(
runtime_target.deployment_selector,
"runtime_target.deployment_selector",
),
"account_scope": _scope_component(
runtime_target.account_scope, "runtime_target.account_scope"
),
"account_selector": list(account_selector),
"service_name": _scope_component(
runtime_target.service_name, "runtime_target.service_name"
),
}
if not any(
(
material["deployment_selector"],
material["account_scope"],
material["account_selector"],
material["service_name"],
)
):
_invalid("runtime_target needs an explicit non-live scope selector")
return sha256(_canonical_bytes(material)).hexdigest()


def _strategy_release(value: object) -> StrategyReleaseIdentity:
if isinstance(value, StrategyReleaseIdentity):
return value
Expand Down Expand Up @@ -455,11 +523,13 @@ def build_non_live_execution_evidence_report_artifacts(
"NON_LIVE_CANDIDATE_SUBJECTS",
"NON_LIVE_EXECUTION_CHANNELS",
"NON_LIVE_EXECUTION_EVIDENCE_BINDING_SCHEMA_VERSION",
"NON_LIVE_RUNTIME_SCOPE_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",
"non_live_runtime_scope_sha256",
"validate_non_live_execution_evidence_binding",
]
77 changes: 77 additions & 0 deletions tests/test_non_live_execution_evidence.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,10 @@
import pytest

from quant_platform_kit.common.runtime_reports import build_runtime_report_base
from quant_platform_kit.common.runtime_target import (
RuntimeExecutionEnvironment,
build_runtime_target,
)
from quant_platform_kit.common.strategy_release import build_strategy_release_identity
from quant_platform_kit.strategy_lifecycle.forward_observation import (
ForwardObservationPolicy,
Expand All @@ -22,6 +26,7 @@
build_paired_shadow_execution_evidence_binding,
canonical_non_live_execution_evidence_binding_bytes,
non_live_execution_evidence_binding_sha256,
non_live_runtime_scope_sha256,
validate_non_live_execution_evidence_binding,
)
from quant_platform_kit.strategy_lifecycle.paired_shadow_evidence import (
Expand Down Expand Up @@ -161,6 +166,78 @@ def test_generic_binding_accepts_the_shared_release_identity_object() -> None:
assert binding["strategy_release"] == _release()


def test_runtime_scope_digest_is_opaque_stable_and_target_bound() -> None:
target = build_runtime_target(
platform_id="longbridge_sg",
strategy_profile="soxl_tactical",
dry_run_only=True,
deployment_selector="sg",
account_selector=("paper_scope",),
account_scope="sg_paper",
service_name="longbridge-shadow-observer",
)
changed_target = build_runtime_target(
platform_id="longbridge_sg",
strategy_profile="soxl_tactical",
dry_run_only=True,
deployment_selector="sg",
account_selector=("different_scope",),
account_scope="sg_paper",
service_name="longbridge-shadow-observer",
)

first = non_live_runtime_scope_sha256(
runtime_target=target, execution_channel="shadow"
)
assert first == non_live_runtime_scope_sha256(
runtime_target=target, execution_channel="shadow"
)
assert first != non_live_runtime_scope_sha256(
runtime_target=changed_target, execution_channel="shadow"
)
assert len(first) == 64
assert "paper_scope" not in first


def test_runtime_scope_rejects_live_or_ambiguous_shadow_targets() -> None:
live_target = build_runtime_target(
platform_id="longbridge_sg",
strategy_profile="soxl_tactical",
dry_run_only=False,
deployment_selector="sg",
)
paper_target = build_runtime_target(
platform_id="longbridge_sg",
strategy_profile="soxl_tactical",
dry_run_only=False,
deployment_selector="sg",
execution_environment=RuntimeExecutionEnvironment.PAPER,
)
unscoped_shadow_target = build_runtime_target(
platform_id="longbridge_sg",
strategy_profile="soxl_tactical",
dry_run_only=True,
)

with pytest.raises(InvalidNonLiveExecutionEvidenceBinding, match="live execution"):
non_live_runtime_scope_sha256(
runtime_target=live_target, execution_channel="paper"
)
with pytest.raises(InvalidNonLiveExecutionEvidenceBinding, match="dry_run"):
non_live_runtime_scope_sha256(
runtime_target=paper_target, execution_channel="shadow"
)
assert len(
non_live_runtime_scope_sha256(
runtime_target=paper_target, execution_channel="paper"
)
) == 64
with pytest.raises(InvalidNonLiveExecutionEvidenceBinding, match="scope selector"):
non_live_runtime_scope_sha256(
runtime_target=unscoped_shadow_target, execution_channel="shadow"
)


@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
Expand Down