From 7a3c84daf89f4feb0de4aae81686c4955c6640f4 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 5 Aug 2026 06:30:53 +0800 Subject: [PATCH] feat: add strict strategy evidence v2 gate Co-Authored-By: Codex --- docs/evidence_package_template.md | 13 + docs/evidence_package_template.zh-CN.md | 13 + .../strategy_promotion_risk_standard.zh-CN.md | 16 + scripts/validate_strategy_evidence_package.py | 279 +--- .../strategy-evidence-package.v2.schema.json | 458 ++++++ .../strategy_lifecycle/__init__.py | 10 + .../strategy_lifecycle/evidence_gate.py | 155 +- .../strategy_lifecycle/evidence_package_v2.py | 1312 +++++++++++++++++ .../live_candidate_notifications.py | 63 +- tests/test_lifecycle_evidence_gate.py | 25 + ..._lifecycle_live_candidate_notifications.py | 37 +- ...t_strategy_evidence_package_v2_contract.py | 803 ++++++++++ ...est_strategy_evidence_package_validator.py | 56 +- ...test_validate_strategy_evidence_package.py | 24 +- 14 files changed, 2955 insertions(+), 309 deletions(-) create mode 100644 src/quant_platform_kit/schemas/strategy-evidence-package.v2.schema.json create mode 100644 src/quant_platform_kit/strategy_lifecycle/evidence_package_v2.py create mode 100644 tests/test_strategy_evidence_package_v2_contract.py diff --git a/docs/evidence_package_template.md b/docs/evidence_package_template.md index a81f6b58..ee253aba 100644 --- a/docs/evidence_package_template.md +++ b/docs/evidence_package_template.md @@ -41,3 +41,16 @@ If any of the following are missing, keep the profile out of live settings: - strategy repo: produces the evidence package - platform repo: verifies runtime compatibility and gate status - operator review: makes the final live decision + +## Canonical promotion package (v2) + +Promotion reruns must produce a new `strategy_evidence_package.v2` that validates against the packaged `strategy-evidence-package.v2.schema.json` and the dependency-free Python validator. Do not relabel or implicitly migrate a v1/alias package. + +The closed v2 object binds strategy and input provenance, the exact `BacktestOrchestrator` `purged_walk_forward.v1` output, at least three ordered folds, positive purge/embargo, an independent locked OOS window of at least 12 calendar months, timing/cost/risk/metric identities, verified repo-relative artifact bytes and SHA-256 digests, and a human acceptance bound to the evidence-core digest. + +Required lifecycle claims are fail-closed: + +- learning: `learning_only=true`, `promotion_eligible=false`, `live_ready=false`, `size_zero_required=true`, `no_order=true`; +- accepted promotion evidence may set `promotion_eligible=true`, but must still keep `live_ready=false`, `size_zero_required=true`, and `no_order=true`. + +Structural validation does not invent performance thresholds. Metric quality remains a bound human promotion decision. A requested stage, CI/PR/review/health result, or notification never grants paper, shadow, live, order, or capital authority; live/runtime requests remain `HOLD`. diff --git a/docs/evidence_package_template.zh-CN.md b/docs/evidence_package_template.zh-CN.md index 87926108..eb05ddcb 100644 --- a/docs/evidence_package_template.zh-CN.md +++ b/docs/evidence_package_template.zh-CN.md @@ -41,3 +41,16 @@ requested_stage: live_candidate - 策略仓库：产出证据包 - 平台仓库：验证 runtime 兼容性和门槛状态 - 操作审批：做最终 live 决策 + +## Canonical 晋级包（v2） + +promotion rerun 必须重新生成 `strategy_evidence_package.v2`，并同时通过 packaged schema 与 dependency-free Python validator。不得把 v1/alias 静默补默认值或改标签后冒充 v2。 + +封闭的 v2 object 必须绑定 strategy/input provenance、`BacktestOrchestrator` 的 `purged_walk_forward.v1` 原始输出、至少 3 个有序 folds、正数 purge/embargo、至少 12 个日历月的锁定独立 OOS、calendar/timezone/signal/execution timing、cost/risk/全部指标，以及 repo-relative artifact 实际 bytes 与 SHA-256。human acceptance 必须用 evidence-core SHA-256 绑定当前证据。 + +生命周期真值必须 fail closed： + +- learning：`learning_only=true`、`promotion_eligible=false`、`live_ready=false`、`size_zero_required=true`、`no_order=true`； +- 完整且经绑定的人类接受的证据可以 `promotion_eligible=true`，但仍必须 `live_ready=false`、`size_zero_required=true`、`no_order=true`。 + +结构验证不臆造性能阈值；指标质量仍由绑定的人类 promotion acceptance 判断。requested stage、CI、PR、review、health 或 notification 都不能产生 paper/shadow/live、order 或 capital 权限；live/runtime 请求一律 `HOLD`。 diff --git a/docs/strategy_promotion_risk_standard.zh-CN.md b/docs/strategy_promotion_risk_standard.zh-CN.md index 4b57b6ba..791560ab 100644 --- a/docs/strategy_promotion_risk_standard.zh-CN.md +++ b/docs/strategy_promotion_risk_standard.zh-CN.md @@ -124,3 +124,19 @@ AI 自动优化必须遵守以下规则： - [ ] `position_control_allowed=true` 已绑定 `evidence_package_id` - [ ] 证据包有效期明确 - [ ] `bounded budget` 已输出且可审计 + +## `strategy_evidence_package.v2` 晋级证据门 + +晋级重跑必须由 producer 生成新的 `strategy_evidence_package.v2`；v1/alias 只保留研究与监控兼容，不自动迁移成 v2。v2 必须同时绑定： + +- strategy/source revision、input provenance/license/range/timestamp/manifest digest； +- `BacktestOrchestrator` 的 `purged_walk_forward.v1` 输出、至少 3 个有序 folds、正数 purge/embargo，以及锁定且独立的至少 12 个日历月 OOS； +- calendar/timezone/signal/execution timing、config/data-manifest/backtest/risk/IC/cost artifacts 及其实际 bytes/SHA-256； +- 上述全部风险指标及 `information_coefficient`。所有 metric/cost 必须存在、非 bool 且有限，cost/risk 状态必须为 `PASS`； +- human acceptance 的 decision/id/actor/time/authority-receipt SHA-256，并以 evidence-core SHA-256 绑定当前证据。 + +机器只判断结构、身份、有限性、日期、digest 与 PASS 状态；本文未冻结 Sharpe、return、MDD 或 IC 数值阈值，指标质量仍由绑定的人类 promotion acceptance 判断。 + +本 v2 门只产生研究晋级资格，不产生 paper/shadow/live 权限：`live_ready=false`、`size_zero_required=true`、`no_order=true` 始终成立。`requested_stage`、CI、PR、review、health 或 notification 不能改变这些真值；legacy/v2 live 或 runtime 请求都必须 `HOLD`。 + +本门完成也不改变 P3 的 `TERMINALLY_PARKED_NO_MEMBER` 状态。 diff --git a/scripts/validate_strategy_evidence_package.py b/scripts/validate_strategy_evidence_package.py index 773b8509..ce92b5ef 100644 --- a/scripts/validate_strategy_evidence_package.py +++ b/scripts/validate_strategy_evidence_package.py @@ -1,214 +1,22 @@ #!/usr/bin/env python3 -"""Validate a strategy evidence package JSON file.""" +"""Compatibility CLI for the canonical strategy evidence package validator.""" from __future__ import annotations import argparse -import hashlib -import json -import re import sys -from datetime import datetime -from pathlib import Path -from typing import Any -ALLOWED_REQUESTED_STAGES = { - "research_backtest_only", - "ai_monitored_candidate", - "shadow_candidate", - "live_candidate", - "runtime_enabled", -} -ALLOWED_KELLY_LEVELS = {"K0", "K1", "K2", "K3", "K4"} -REQUIRED_ARTIFACTS = ( - "returns", - "trades", - "positions", - "config", - "data_manifest", - "candidate_registry", - "benchmark_registry", - "cost_model", - "risk_report", - "kelly_readiness_report", +from quant_platform_kit.strategy_lifecycle.evidence_package_v2 import ( + STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION, + canonical_evidence_package_v2_bytes, + read_evidence_package_v2_json, + validate_evidence_package_v2, + validate_strategy_evidence_file, + validate_strategy_evidence_payload, ) -REQUIRED_RISK_METRICS = ( - "sharpe_ratio", - "sortino_ratio", - "max_drawdown", - "annualized_return", - "annualized_volatility", - "calmar_ratio", - "information_ratio", - "var_95", - "cvar_95", - "turnover", - "trade_count", - "win_rate", - "profit_factor", -) -REQUIRED_RISK_BENCHMARK = ("name", "alpha", "beta") -REQUIRED_RISK_COST_STRESS = ("slippage_bps", "commission_bps", "passed") -REQUIRED_RISK_OOS = ("window_start", "window_end", "locked") -SHA256_RE = re.compile(r"^[A-Fa-f0-9]{64}$") - - -def validate_payload(payload: Any, *, base_dir: Path | None = None) -> list[str]: - issues: list[str] = [] - - if not isinstance(payload, dict): - return ["top-level JSON must be an object"] - - for field in ( - "schema_version", - "profile", - "market", - "requested_stage", - "generated_at", - "evidence_package_id", - "artifacts", - "validation", - "risk", - "kelly_readiness", - "ai_optimization", - ): - if field not in payload: - issues.append(f"missing required field: {field}") - - _check_non_empty_string(payload, "schema_version", issues) - _check_non_empty_string(payload, "profile", issues) - _check_non_empty_string(payload, "market", issues) - _check_non_empty_string(payload, "evidence_package_id", issues) - - requested_stage = payload.get("requested_stage") - if not isinstance(requested_stage, str) or not requested_stage.strip(): - issues.append("requested_stage must be a non-empty string") - elif requested_stage not in ALLOWED_REQUESTED_STAGES: - issues.append(f"unsupported requested_stage: {requested_stage!r}") - - generated_at = payload.get("generated_at") - if not isinstance(generated_at, str) or not generated_at.strip(): - issues.append("generated_at must be a non-empty string") - elif not _is_datetime_string(generated_at): - issues.append(f"generated_at is not a valid date-time: {generated_at!r}") - - artifacts = payload.get("artifacts") - if not isinstance(artifacts, dict): - issues.append("artifacts must be an object") - else: - for name in REQUIRED_ARTIFACTS: - artifact = artifacts.get(name) - if not isinstance(artifact, dict): - issues.append(f"artifacts.{name} must be an object") - continue - _check_non_empty_string(artifact, "path", issues, prefix=f"artifacts.{name}") - sha256 = artifact.get("sha256") - if not isinstance(sha256, str) or not SHA256_RE.fullmatch(sha256): - issues.append(f"artifacts.{name}.sha256 must be a 64-character hex string") - continue - if base_dir is not None: - _validate_artifact_file( - name=name, - artifact=artifact, - expected_sha256=sha256, - base_dir=base_dir, - issues=issues, - ) - - validation = payload.get("validation") - if not isinstance(validation, dict): - issues.append("validation must be an object") - else: - if not isinstance(validation.get("oos_passed"), bool): - issues.append("validation.oos_passed must be a boolean") - if not isinstance(validation.get("overfit_report_present"), bool): - issues.append("validation.overfit_report_present must be a boolean") - if requested_stage in {"live_candidate", "runtime_enabled"}: - if validation.get("oos_passed") is not True: - issues.append(f"{requested_stage} requires validation.oos_passed=true") - if validation.get("overfit_report_present") is not True: - issues.append(f"{requested_stage} requires validation.overfit_report_present=true") - - risk = payload.get("risk") - if not isinstance(risk, dict): - issues.append("risk must be an object") - else: - metrics = risk.get("metrics") - if not isinstance(metrics, dict): - issues.append("risk.metrics must be an object") - else: - for field in REQUIRED_RISK_METRICS: - value = metrics.get(field) - if field == "trade_count": - if not _is_int(value): - issues.append("risk.metrics.trade_count must be an integer") - elif value < 0: - issues.append("risk.metrics.trade_count must be >= 0") - elif not _is_number(value): - issues.append(f"risk.metrics.{field} must be a number") - if _is_number(metrics.get("win_rate")): - win_rate = metrics["win_rate"] - if win_rate < 0 or win_rate > 1: - issues.append("risk.metrics.win_rate must be between 0 and 1") - - benchmark = risk.get("benchmark") - if not isinstance(benchmark, dict): - issues.append("risk.benchmark must be an object") - else: - _check_non_empty_string(benchmark, "name", issues, prefix="risk.benchmark") - if not _is_number(benchmark.get("alpha")): - issues.append("risk.benchmark.alpha must be a number") - if not _is_number(benchmark.get("beta")): - issues.append("risk.benchmark.beta must be a number") - - cost_stress = risk.get("cost_stress") - if not isinstance(cost_stress, dict): - issues.append("risk.cost_stress must be an object") - else: - if not _is_number(cost_stress.get("slippage_bps")): - issues.append("risk.cost_stress.slippage_bps must be a number") - if not _is_number(cost_stress.get("commission_bps")): - issues.append("risk.cost_stress.commission_bps must be a number") - if not isinstance(cost_stress.get("passed"), bool): - issues.append("risk.cost_stress.passed must be a boolean") - - oos = risk.get("oos") - if not isinstance(oos, dict): - issues.append("risk.oos must be an object") - else: - _check_non_empty_string(oos, "window_start", issues, prefix="risk.oos") - _check_non_empty_string(oos, "window_end", issues, prefix="risk.oos") - if not isinstance(oos.get("locked"), bool): - issues.append("risk.oos.locked must be a boolean") - - kelly_readiness = payload.get("kelly_readiness") - if not isinstance(kelly_readiness, dict): - issues.append("kelly_readiness must be an object") - else: - level = kelly_readiness.get("level") - if not isinstance(level, str) or level not in ALLOWED_KELLY_LEVELS: - issues.append("kelly_readiness.level must be one of K0, K1, K2, K3, K4") - if kelly_readiness.get("full_kelly_allowed") is not False: - issues.append("kelly_readiness.full_kelly_allowed must be false") - - ai_optimization = payload.get("ai_optimization") - if not isinstance(ai_optimization, dict): - issues.append("ai_optimization must be an object") - return issues - - -def validate_file(path: str | Path) -> list[str]: - evidence_path = Path(path) - try: - payload = json.loads(evidence_path.read_text(encoding="utf-8")) - except FileNotFoundError: - return [f"file not found: {evidence_path}"] - except json.JSONDecodeError as exc: - return [f"invalid JSON: {exc.msg} (line {exc.lineno}, column {exc.colno})"] - except OSError as exc: - return [f"failed to read file: {exc}"] - return validate_payload(payload, base_dir=evidence_path.parent) +validate_payload = validate_strategy_evidence_payload +validate_file = validate_strategy_evidence_file def main(argv: list[str] | None = None) -> int: @@ -224,72 +32,5 @@ def main(argv: list[str] | None = None) -> int: return 0 -def _check_non_empty_string( - payload: dict[str, Any], - field: str, - issues: list[str], - *, - prefix: str | None = None, -) -> None: - value = payload.get(field) - label = f"{prefix}.{field}" if prefix else field - if not isinstance(value, str) or not value.strip(): - issues.append(f"{label} must be a non-empty string") - - -def _is_number(value: Any) -> bool: - return isinstance(value, (int, float)) and not isinstance(value, bool) - - -def _is_int(value: Any) -> bool: - return isinstance(value, int) and not isinstance(value, bool) - - -def _is_datetime_string(value: str) -> bool: - candidate = value.strip() - if candidate.endswith("Z"): - candidate = f"{candidate[:-1]}+00:00" - try: - datetime.fromisoformat(candidate) - except ValueError: - return False - return True - - -def _validate_artifact_file( - *, - name: str, - artifact: dict[str, Any], - expected_sha256: str, - base_dir: Path, - issues: list[str], -) -> None: - raw_path = str(artifact.get("path") or "").strip() - label = f"artifacts.{name}" - if not raw_path: - return - path = Path(raw_path) - if path.is_absolute(): - issues.append(f"{label}.path must be repo-relative, got absolute path") - return - - resolved = (base_dir / path).resolve() - try: - resolved.relative_to(base_dir.resolve()) - except ValueError: - issues.append(f"{label}.path must stay within the evidence package directory") - return - - if not resolved.is_file(): - issues.append(f"{label}.path does not exist: {raw_path}") - return - - actual_sha256 = hashlib.sha256(resolved.read_bytes()).hexdigest() - if actual_sha256.lower() != expected_sha256.lower(): - issues.append( - f"{label}.sha256 mismatch: expected {expected_sha256.lower()}, got {actual_sha256.lower()}" - ) - - if __name__ == "__main__": raise SystemExit(main()) diff --git a/src/quant_platform_kit/schemas/strategy-evidence-package.v2.schema.json b/src/quant_platform_kit/schemas/strategy-evidence-package.v2.schema.json new file mode 100644 index 00000000..0c1ff87f --- /dev/null +++ b/src/quant_platform_kit/schemas/strategy-evidence-package.v2.schema.json @@ -0,0 +1,458 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://quantstrategylab.dev/schemas/strategy-evidence-package.v2.schema.json", + "title": "strategy_evidence_package.v2", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "evidence_package_id", + "generated_at", + "requested_stage", + "strategy", + "input_provenance", + "backtest", + "artifacts", + "metrics", + "cost_stress", + "risk_assessment", + "digests", + "human_acceptance", + "lifecycle_claims" + ], + "properties": { + "schema_version": {"const": "strategy_evidence_package.v2"}, + "evidence_package_id": {"$ref": "#/$defs/nonEmptyString"}, + "generated_at": {"$ref": "#/$defs/dateTime"}, + "requested_stage": { + "enum": [ + "research_backtest_only", + "ai_monitored_candidate", + "shadow_candidate", + "live_candidate", + "runtime_enabled" + ] + }, + "strategy": { + "type": "object", + "additionalProperties": false, + "required": ["profile", "domain", "source_revision"], + "properties": { + "profile": {"$ref": "#/$defs/nonEmptyString"}, + "domain": {"$ref": "#/$defs/nonEmptyString"}, + "source_revision": {"$ref": "#/$defs/sourceRevision"} + } + }, + "input_provenance": { + "type": "object", + "additionalProperties": false, + "required": [ + "source", + "source_revision", + "license", + "usage_scope", + "range", + "timestamp", + "manifest_sha256" + ], + "properties": { + "source": {"$ref": "#/$defs/nonEmptyString"}, + "source_revision": {"$ref": "#/$defs/nonEmptyString"}, + "license": {"$ref": "#/$defs/nonEmptyString"}, + "usage_scope": {"$ref": "#/$defs/nonEmptyString"}, + "range": { + "type": "object", + "additionalProperties": false, + "required": ["start", "end"], + "properties": { + "start": {"$ref": "#/$defs/date"}, + "end": {"$ref": "#/$defs/date"} + } + }, + "timestamp": {"$ref": "#/$defs/dateTime"}, + "manifest_sha256": {"$ref": "#/$defs/sha256"} + } + }, + "backtest": { + "type": "object", + "additionalProperties": false, + "required": [ + "orchestrator", + "protocol", + "calendar", + "timezone", + "signal_timing", + "execution_timing", + "locked_independent_oos", + "promotion_run" + ], + "properties": { + "orchestrator": {"const": "BacktestOrchestrator"}, + "protocol": {"const": "purged_walk_forward.v1"}, + "calendar": {"$ref": "#/$defs/nonEmptyString"}, + "timezone": {"$ref": "#/$defs/nonEmptyString"}, + "signal_timing": {"$ref": "#/$defs/nonEmptyString"}, + "execution_timing": {"$ref": "#/$defs/nonEmptyString"}, + "locked_independent_oos": { + "type": "object", + "additionalProperties": false, + "required": ["locked", "independent", "reused_for_selection"], + "properties": { + "locked": {"const": true}, + "independent": {"const": true}, + "reused_for_selection": {"const": false} + } + }, + "promotion_run": {"$ref": "#/$defs/promotionRun"} + } + }, + "artifacts": { + "type": "object", + "additionalProperties": false, + "required": [ + "config", + "data_manifest", + "backtest", + "risk", + "information_coefficient", + "cost_model" + ], + "properties": { + "config": {"$ref": "#/$defs/artifact"}, + "data_manifest": {"$ref": "#/$defs/artifact"}, + "backtest": {"$ref": "#/$defs/artifact"}, + "risk": {"$ref": "#/$defs/artifact"}, + "information_coefficient": {"$ref": "#/$defs/artifact"}, + "cost_model": {"$ref": "#/$defs/artifact"} + } + }, + "metrics": { + "type": "object", + "additionalProperties": false, + "required": [ + "sharpe_ratio", + "sortino_ratio", + "max_drawdown", + "annualized_return", + "annualized_volatility", + "calmar_ratio", + "information_ratio", + "information_coefficient", + "var_95", + "cvar_95", + "turnover", + "trade_count", + "win_rate", + "profit_factor" + ], + "properties": { + "sharpe_ratio": {"type": "number"}, + "sortino_ratio": {"type": "number"}, + "max_drawdown": {"type": "number"}, + "annualized_return": {"type": "number"}, + "annualized_volatility": {"type": "number"}, + "calmar_ratio": {"type": "number"}, + "information_ratio": {"type": "number"}, + "information_coefficient": {"type": "number"}, + "var_95": {"type": "number"}, + "cvar_95": {"type": "number"}, + "turnover": {"type": "number"}, + "trade_count": {"type": "integer", "minimum": 0}, + "win_rate": {"type": "number", "minimum": 0, "maximum": 1}, + "profit_factor": {"type": "number"} + } + }, + "cost_stress": { + "type": "object", + "additionalProperties": false, + "required": ["scenarios", "status"], + "properties": { + "scenarios": { + "type": "array", + "minItems": 3, + "maxItems": 3, + "prefixItems": [ + {"$ref": "#/$defs/costScenario1"}, + {"$ref": "#/$defs/costScenario2"}, + {"$ref": "#/$defs/costScenario3"} + ], + "items": false + }, + "status": {"const": "PASS"} + } + }, + "risk_assessment": { + "type": "object", + "additionalProperties": false, + "required": ["status", "standard_id", "standard_sha256"], + "properties": { + "status": {"const": "PASS"}, + "standard_id": {"$ref": "#/$defs/nonEmptyString"}, + "standard_sha256": {"$ref": "#/$defs/sha256"} + } + }, + "digests": { + "type": "object", + "additionalProperties": false, + "required": [ + "config_sha256", + "data_manifest_sha256", + "backtest_sha256", + "risk_sha256", + "information_coefficient_sha256", + "cost_model_sha256", + "evidence_core_sha256", + "package_sha256" + ], + "properties": { + "config_sha256": {"$ref": "#/$defs/sha256"}, + "data_manifest_sha256": {"$ref": "#/$defs/sha256"}, + "backtest_sha256": {"$ref": "#/$defs/sha256"}, + "risk_sha256": {"$ref": "#/$defs/sha256"}, + "information_coefficient_sha256": {"$ref": "#/$defs/sha256"}, + "cost_model_sha256": {"$ref": "#/$defs/sha256"}, + "evidence_core_sha256": {"$ref": "#/$defs/sha256"}, + "package_sha256": {"$ref": "#/$defs/sha256"} + } + }, + "human_acceptance": { + "oneOf": [ + {"type": "null"}, + { + "type": "object", + "additionalProperties": false, + "required": [ + "decision", + "acceptance_id", + "actor", + "accepted_at", + "authority_receipt_sha256", + "evidence_core_sha256" + ], + "properties": { + "decision": {"enum": ["ACCEPTED", "REJECTED"]}, + "acceptance_id": {"$ref": "#/$defs/nonEmptyString"}, + "actor": {"$ref": "#/$defs/nonEmptyString"}, + "accepted_at": {"$ref": "#/$defs/dateTime"}, + "authority_receipt_sha256": {"$ref": "#/$defs/sha256"}, + "evidence_core_sha256": {"$ref": "#/$defs/sha256"} + } + } + ] + }, + "lifecycle_claims": { + "type": "object", + "additionalProperties": false, + "required": [ + "learning_only", + "promotion_eligible", + "live_ready", + "size_zero_required", + "no_order" + ], + "properties": { + "learning_only": {"type": "boolean"}, + "promotion_eligible": {"type": "boolean"}, + "live_ready": {"const": false}, + "size_zero_required": {"const": true}, + "no_order": {"const": true} + }, + "allOf": [ + { + "if": {"properties": {"learning_only": {"const": true}}}, + "then": {"properties": {"promotion_eligible": {"const": false}}} + } + ] + } + }, + "$defs": { + "nonEmptyString": {"type": "string", "minLength": 1, "pattern": ".*\\S.*"}, + "sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "sourceRevision": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "date": {"type": "string", "format": "date", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}$"}, + "dateTime": { + "type": "string", + "format": "date-time", + "pattern": "(?:Z|[+-][0-9]{2}:[0-9]{2})$" + }, + "artifactPath": { + "type": "string", + "minLength": 1, + "pattern": "^(?!/)(?!.*(?:^|/)\\.\\.(?:/|$))(?!.*(?:^|/)\\.(?:/|$))(?!.*[\\u0000-\\u001f\\u007f])(?!.*\\\\)(?!.*//).+$" + }, + "artifact": { + "type": "object", + "additionalProperties": false, + "required": ["path", "sha256"], + "properties": { + "path": {"$ref": "#/$defs/artifactPath"}, + "sha256": {"$ref": "#/$defs/sha256"} + } + }, + "fold": { + "type": "object", + "additionalProperties": false, + "required": ["train_start", "train_end", "test_start", "test_end"], + "properties": { + "train_start": {"$ref": "#/$defs/date"}, + "train_end": {"$ref": "#/$defs/date"}, + "test_start": {"$ref": "#/$defs/date"}, + "test_end": {"$ref": "#/$defs/date"} + } + }, + "costInputs": { + "type": "object", + "additionalProperties": false, + "required": ["commission_bps", "slippage_bps", "market_impact_bps"], + "properties": { + "commission_bps": {"type": "number", "minimum": 0}, + "slippage_bps": {"type": "number", "minimum": 0}, + "market_impact_bps": {"type": "number", "minimum": 0} + } + }, + "validationIdentity": { + "type": "object", + "additionalProperties": false, + "required": [ + "protocol", + "fold_id", + "fold_role", + "train_start", + "train_end", + "test_start", + "test_end", + "locked_oos_start", + "locked_oos_end", + "purge_days", + "embargo_days" + ], + "properties": { + "protocol": {"const": "purged_walk_forward.v1"}, + "fold_id": {"$ref": "#/$defs/nonEmptyString"}, + "fold_role": {"enum": ["test", "locked_oos"]}, + "train_start": {"oneOf": [{"$ref": "#/$defs/date"}, {"type": "null"}]}, + "train_end": {"oneOf": [{"$ref": "#/$defs/date"}, {"type": "null"}]}, + "test_start": {"$ref": "#/$defs/date"}, + "test_end": {"$ref": "#/$defs/date"}, + "locked_oos_start": {"$ref": "#/$defs/date"}, + "locked_oos_end": {"$ref": "#/$defs/date"}, + "purge_days": {"type": "integer", "minimum": 1}, + "embargo_days": {"type": "integer", "minimum": 1} + } + }, + "backtestResult": { + "type": "object", + "additionalProperties": false, + "required": [ + "strategy_profile", + "domain", + "start_date", + "end_date", + "source_revision", + "cost_model", + "cost_inputs", + "validation_identity", + "sharpe_ratio", + "max_drawdown", + "cagr", + "observation_count", + "run_duration_seconds" + ], + "properties": { + "strategy_profile": {"$ref": "#/$defs/nonEmptyString"}, + "domain": {"$ref": "#/$defs/nonEmptyString"}, + "param_set_id": {"type": "string"}, + "params": {"type": "object"}, + "param_version": {"type": "integer", "minimum": 1}, + "sharpe_ratio": {"type": "number"}, + "calmar_ratio": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "sortino_ratio": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "max_drawdown": {"type": "number"}, + "cagr": {"type": "number"}, + "volatility": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "win_rate": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "total_return": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "start_date": {"$ref": "#/$defs/date"}, + "end_date": {"$ref": "#/$defs/date"}, + "observation_count": {"type": "integer", "minimum": 1}, + "benchmark_symbol": {"type": "string"}, + "benchmark_cagr": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "benchmark_max_drawdown": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "excess_cagr": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "oos_sharpe": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "oos_calmar": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "oos_max_drawdown": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "walk_forward_stability": {"oneOf": [{"type": "number"}, {"type": "null"}]}, + "run_id": {"type": "string"}, + "run_duration_seconds": {"type": "number", "minimum": 0}, + "source_script": {"type": "string"}, + "computed_at": {"type": "string"}, + "source_revision": {"$ref": "#/$defs/sourceRevision"}, + "cost_model": {"$ref": "#/$defs/nonEmptyString"}, + "validation_identity": {"$ref": "#/$defs/validationIdentity"}, + "cost_inputs": {"$ref": "#/$defs/costInputs"} + } + }, + "promotionRun": { + "type": "object", + "additionalProperties": false, + "required": [ + "strategy_profile", + "domain", + "folds", + "fold_results", + "locked_oos_result", + "locked_oos_start", + "locked_oos_end", + "purge_days", + "embargo_days", + "source_revision", + "cost_model" + ], + "properties": { + "strategy_profile": {"$ref": "#/$defs/nonEmptyString"}, + "domain": {"$ref": "#/$defs/nonEmptyString"}, + "folds": {"type": "array", "minItems": 3, "items": {"$ref": "#/$defs/fold"}}, + "fold_results": {"type": "array", "minItems": 3, "items": {"$ref": "#/$defs/backtestResult"}}, + "locked_oos_result": {"$ref": "#/$defs/backtestResult"}, + "locked_oos_start": {"$ref": "#/$defs/date"}, + "locked_oos_end": {"$ref": "#/$defs/date"}, + "purge_days": {"type": "integer", "minimum": 1}, + "embargo_days": {"type": "integer", "minimum": 1}, + "source_revision": {"$ref": "#/$defs/sourceRevision"}, + "cost_model": { + "type": "object", + "additionalProperties": false, + "required": ["model_id", "commission_bps", "slippage_bps", "market_impact_bps"], + "properties": { + "model_id": {"$ref": "#/$defs/nonEmptyString"}, + "commission_bps": {"type": "number", "minimum": 0}, + "slippage_bps": {"type": "number", "minimum": 0}, + "market_impact_bps": {"type": "number", "minimum": 0} + } + } + } + }, + "costScenario1": {"$ref": "#/$defs/costScenarioBase1"}, + "costScenario2": {"$ref": "#/$defs/costScenarioBase2"}, + "costScenario3": {"$ref": "#/$defs/costScenarioBase3"}, + "costScenarioBase1": { + "type": "object", + "additionalProperties": false, + "required": ["multiplier", "total_cost_bps"], + "properties": {"multiplier": {"const": 1}, "total_cost_bps": {"type": "number", "minimum": 0}} + }, + "costScenarioBase2": { + "type": "object", + "additionalProperties": false, + "required": ["multiplier", "total_cost_bps"], + "properties": {"multiplier": {"const": 2}, "total_cost_bps": {"type": "number", "minimum": 0}} + }, + "costScenarioBase3": { + "type": "object", + "additionalProperties": false, + "required": ["multiplier", "total_cost_bps"], + "properties": {"multiplier": {"const": 3}, "total_cost_bps": {"type": "number", "minimum": 0}} + } + } +} diff --git a/src/quant_platform_kit/strategy_lifecycle/__init__.py b/src/quant_platform_kit/strategy_lifecycle/__init__.py index 14e4bc5d..981d3935 100644 --- a/src/quant_platform_kit/strategy_lifecycle/__init__.py +++ b/src/quant_platform_kit/strategy_lifecycle/__init__.py @@ -29,6 +29,12 @@ validate_evidence_package, validate_evidence_package_file, ) +from quant_platform_kit.strategy_lifecycle.evidence_package_v2 import ( + STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION, + canonical_evidence_package_v2_bytes, + read_evidence_package_v2_json, + validate_evidence_package_v2, +) from quant_platform_kit.strategy_lifecycle.live_candidate_notifications import ( LiveCandidateNotificationEvent, build_live_candidate_notification, @@ -45,6 +51,7 @@ "RESEARCH_SPEC_SCHEMA_VERSION", "OPTIMIZATION_SPEC_SCHEMA_VERSION", "StrategyHealthScore", + "STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION", "StrategyPerformanceSnapshot", "UpdateLogEntry", "UpdateStage", @@ -53,9 +60,12 @@ "EvidencePackage", "LiveCandidateNotificationEvent", "load_evidence_package", + "canonical_evidence_package_v2_bytes", + "read_evidence_package_v2_json", "build_live_candidate_notification", "validate_evidence_package", "validate_evidence_package_file", + "validate_evidence_package_v2", "validate_optimization_spec", "validate_research_spec", "validate_strategy_spec", diff --git a/src/quant_platform_kit/strategy_lifecycle/evidence_gate.py b/src/quant_platform_kit/strategy_lifecycle/evidence_gate.py index 37f2b07f..da0fd281 100644 --- a/src/quant_platform_kit/strategy_lifecycle/evidence_gate.py +++ b/src/quant_platform_kit/strategy_lifecycle/evidence_gate.py @@ -2,12 +2,17 @@ from __future__ import annotations -import json import tomllib from dataclasses import dataclass, field from pathlib import Path from typing import Any, Mapping +from .evidence_package_v2 import ( + STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION, + read_evidence_package_v2_json, + validate_evidence_package_v2, +) + ALLOWED_EVIDENCE_STAGES = ( "research_backtest_only", "ai_monitored_candidate", @@ -40,7 +45,21 @@ class EvidencePackage: evidence_version: str = "evidence_package.v1" submitted_at: str = "" + # Appended defaults preserve every legacy positional argument above. + schema_version: str = "" + promotion_eligible: bool = False + live_ready: bool = False + size_zero_required: bool = True + no_order: bool = True + promotion_status: str = "LEGACY_RESEARCH_ONLY" + canonical_payload: Mapping[str, Any] = field(default_factory=dict, repr=False) + def to_dict(self) -> dict[str, object]: + if ( + self.schema_version == STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION + and self.canonical_payload + ): + return dict(self.canonical_payload) return { "strategy_profile": self.strategy_profile, "domain": self.domain, @@ -64,31 +83,56 @@ class EvidenceGateResult: issues: tuple[str, ...] = () warnings: tuple[str, ...] = () + # Appended defaults preserve every legacy positional argument above. + promotion_eligible: bool = False + live_ready: bool = False + size_zero_required: bool = True + no_order: bool = True + promotion_status: str = "LEGACY_RESEARCH_ONLY" + def to_dict(self) -> dict[str, object]: - return { + payload: dict[str, object] = { "valid": self.valid, "issues": list(self.issues), "warnings": list(self.warnings), "package": self.package.to_dict(), } + if self.package.schema_version == STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION: + payload.update( + { + "promotion_eligible": self.promotion_eligible, + "live_ready": self.live_ready, + "size_zero_required": self.size_zero_required, + "no_order": self.no_order, + "promotion_status": self.promotion_status, + } + ) + return payload def load_evidence_package(path: str | Path) -> dict[str, Any]: evidence_path = Path(path) - raw = evidence_path.read_text(encoding="utf-8") suffix = evidence_path.suffix.lower() if suffix == ".json": - payload = json.loads(raw) + payload = read_evidence_package_v2_json(evidence_path) elif suffix == ".toml": + raw = evidence_path.read_text(encoding="utf-8") payload = tomllib.loads(raw) else: - raise ValueError(f"Unsupported evidence package format: {evidence_path.suffix or ''}") + raise ValueError( + f"Unsupported evidence package format: {evidence_path.suffix or ''}" + ) if not isinstance(payload, dict): raise ValueError("Evidence package must decode to a mapping") return payload -def validate_evidence_package(raw: Mapping[str, Any]) -> EvidenceGateResult: +def validate_evidence_package( + raw: Mapping[str, Any], *, base_dir: str | Path | None = None +) -> EvidenceGateResult: + if raw.get("schema_version") == STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION: + return _validate_v2_evidence_package(raw, base_dir=base_dir) + issues: list[str] = [] warnings: list[str] = [] @@ -96,7 +140,9 @@ def validate_evidence_package(raw: Mapping[str, Any]) -> EvidenceGateResult: domain = _first_str(raw, "domain", "market") requested_stage = _first_str(raw, "requested_stage", "stage") target_platforms = _normalize_platforms( - _first_value(raw, "target_platforms", "platforms", "target_platform", "runtime_targets") + _first_value( + raw, "target_platforms", "platforms", "target_platform", "runtime_targets" + ) ) backtest_summary = _first_mapping(raw, "backtest_summary", "backtest", "evidence") drift_notes = _first_value(raw, "drift_notes", "drift", "regime_notes") @@ -108,8 +154,14 @@ def validate_evidence_package(raw: Mapping[str, Any]) -> EvidenceGateResult: ) plugin_gate = _first_value(raw, "plugin_gate", "plugin_gates", "plugins") rollout_notes = _first_value(raw, "rollout_notes", "rollout", "operator_notes") - evidence_version = str(_first_value(raw, "evidence_version", "schema", default="evidence_package.v1") or "evidence_package.v1") - submitted_at = str(_first_value(raw, "submitted_at", "generated_at", "created_at", default="") or "") + evidence_version = str( + _first_value(raw, "evidence_version", "schema", default="evidence_package.v1") + or "evidence_package.v1" + ) + submitted_at = str( + _first_value(raw, "submitted_at", "generated_at", "created_at", default="") + or "" + ) if not strategy_profile: issues.append("missing strategy_profile/profile") @@ -134,7 +186,10 @@ def validate_evidence_package(raw: Mapping[str, Any]) -> EvidenceGateResult: if plugin_gate is not None and not _plugin_gate_is_usable(plugin_gate): issues.append("plugin_gate evidence is incomplete or unsupported") - if requested_stage in {"research_backtest_only", "ai_monitored_candidate"} and not _is_non_empty(rollout_notes): + if requested_stage in { + "research_backtest_only", + "ai_monitored_candidate", + } and not _is_non_empty(rollout_notes): warnings.append("rollout_notes is recommended for candidates") package = EvidencePackage( @@ -151,11 +206,76 @@ def validate_evidence_package(raw: Mapping[str, Any]) -> EvidenceGateResult: evidence_version=evidence_version, submitted_at=submitted_at, ) - return EvidenceGateResult(valid=not issues, package=package, issues=tuple(issues), warnings=tuple(warnings)) + return EvidenceGateResult( + valid=not issues, + package=package, + issues=tuple(issues), + warnings=tuple(warnings), + promotion_eligible=False, + live_ready=False, + size_zero_required=True, + no_order=True, + promotion_status="LEGACY_RESEARCH_ONLY", + ) def validate_evidence_package_file(path: str | Path) -> EvidenceGateResult: - return validate_evidence_package(load_evidence_package(path)) + evidence_path = Path(path) + return validate_evidence_package( + load_evidence_package(evidence_path), base_dir=evidence_path.parent + ) + + +def _validate_v2_evidence_package( + raw: Mapping[str, Any], *, base_dir: str | Path | None +) -> EvidenceGateResult: + issues = validate_evidence_package_v2(raw, base_dir=base_dir) + strategy = raw.get("strategy") + claims = raw.get("lifecycle_claims") + acceptance = raw.get("human_acceptance") + strategy_mapping = strategy if isinstance(strategy, Mapping) else {} + claims_mapping = claims if isinstance(claims, Mapping) else {} + acceptance_mapping = acceptance if isinstance(acceptance, Mapping) else {} + promotion_eligible = claims_mapping.get("promotion_eligible") is True and not issues + learning_only = claims_mapping.get("learning_only") is True + if issues: + promotion_status = "INVALID" + elif learning_only: + promotion_status = "LEARNING_ONLY" + elif promotion_eligible: + promotion_status = "PROMOTION_ELIGIBLE" + elif acceptance_mapping.get("decision") != "ACCEPTED": + promotion_status = "HUMAN_REQUIRED" + else: + promotion_status = "STRUCTURALLY_COMPLETE" + package = EvidencePackage( + strategy_profile=str(strategy_mapping.get("profile") or ""), + domain=str(strategy_mapping.get("domain") or ""), + requested_stage=str(raw.get("requested_stage") or ""), + backtest_summary=raw.get("metrics") + if isinstance(raw.get("metrics"), Mapping) + else {}, + operator_notes=acceptance, + evidence_version=STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION, + submitted_at=str(raw.get("generated_at") or ""), + schema_version=STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION, + promotion_eligible=promotion_eligible, + live_ready=False, + size_zero_required=True, + no_order=True, + promotion_status=promotion_status, + canonical_payload=dict(raw), + ) + return EvidenceGateResult( + valid=not issues, + package=package, + issues=tuple(issues), + promotion_eligible=promotion_eligible, + live_ready=False, + size_zero_required=True, + no_order=True, + promotion_status=promotion_status, + ) def _first_value(raw: Mapping[str, Any], *keys: str, default: Any = None) -> Any: @@ -215,7 +335,10 @@ def _backtest_summary_has_evidence(summary: Mapping[str, Any]) -> bool: def _platform_compatibility_is_usable(compatibility: Mapping[str, Any]) -> bool: if not compatibility: return False - if any(bool(compatibility.get(key)) for key in ("verified", "compatible", "supported", "enabled")): + if any( + bool(compatibility.get(key)) + for key in ("verified", "compatible", "supported", "enabled") + ): return True if _is_non_empty(compatibility.get("runtime_enabled_profiles")): return True @@ -226,7 +349,11 @@ def _platform_compatibility_is_usable(compatibility: Mapping[str, Any]) -> bool: def _plugin_gate_is_usable(plugin_gate: Any) -> bool: if isinstance(plugin_gate, Mapping): - status = str(plugin_gate.get("status") or plugin_gate.get("evidence_status") or "").strip().lower() + status = ( + str(plugin_gate.get("status") or plugin_gate.get("evidence_status") or "") + .strip() + .lower() + ) if status and status not in ALLOWED_PLUGIN_GATE_STATUSES: return False if status == "automation_approved": diff --git a/src/quant_platform_kit/strategy_lifecycle/evidence_package_v2.py b/src/quant_platform_kit/strategy_lifecycle/evidence_package_v2.py new file mode 100644 index 00000000..7b7b81bd --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/evidence_package_v2.py @@ -0,0 +1,1312 @@ +"""Canonical, dependency-free strategy evidence package v2 validation.""" + +from __future__ import annotations + +import copy +import hashlib +import json +import math +import re +from calendar import monthrange +from collections.abc import Mapping +from datetime import date, datetime, timedelta +from pathlib import Path, PurePosixPath +from typing import Any + + +STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION = "strategy_evidence_package.v2" + +_SOURCE_REVISION_RE = re.compile(r"^[0-9a-f]{40}$") +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +_CORE_FIELDS = ( + "schema_version", + "evidence_package_id", + "generated_at", + "requested_stage", + "strategy", + "input_provenance", + "backtest", + "artifacts", + "metrics", + "cost_stress", + "risk_assessment", +) +_TOP_LEVEL_FIELDS = frozenset( + (*_CORE_FIELDS, "digests", "human_acceptance", "lifecycle_claims") +) +_ARTIFACT_DIGEST_FIELDS = { + "config": "config_sha256", + "data_manifest": "data_manifest_sha256", + "backtest": "backtest_sha256", + "risk": "risk_sha256", + "information_coefficient": "information_coefficient_sha256", + "cost_model": "cost_model_sha256", +} +_METRIC_FIELDS = ( + "sharpe_ratio", + "sortino_ratio", + "max_drawdown", + "annualized_return", + "annualized_volatility", + "calmar_ratio", + "information_ratio", + "information_coefficient", + "var_95", + "cvar_95", + "turnover", + "trade_count", + "win_rate", + "profit_factor", +) +_RESULT_NUMBER_FIELDS = ( + "sharpe_ratio", + "calmar_ratio", + "sortino_ratio", + "max_drawdown", + "cagr", + "volatility", + "win_rate", + "total_return", + "observation_count", + "benchmark_cagr", + "benchmark_max_drawdown", + "excess_cagr", + "oos_sharpe", + "oos_calmar", + "oos_max_drawdown", + "walk_forward_stability", + "run_duration_seconds", +) +_RESULT_FIELDS = frozenset( + { + "strategy_profile", + "domain", + "param_set_id", + "params", + "param_version", + "start_date", + "end_date", + "benchmark_symbol", + "run_id", + "source_script", + "computed_at", + "source_revision", + "cost_model", + "validation_identity", + "cost_inputs", + *_RESULT_NUMBER_FIELDS, + } +) +_LEGACY_ALLOWED_REQUESTED_STAGES = { + "research_backtest_only", + "ai_monitored_candidate", + "shadow_candidate", + "live_candidate", + "runtime_enabled", +} +_LEGACY_KELLY_LEVELS = {"K0", "K1", "K2", "K3", "K4"} +_LEGACY_ARTIFACTS = ( + "returns", + "trades", + "positions", + "config", + "data_manifest", + "candidate_registry", + "benchmark_registry", + "cost_model", + "risk_report", + "kelly_readiness_report", +) +_LEGACY_METRICS = ( + "sharpe_ratio", + "sortino_ratio", + "max_drawdown", + "annualized_return", + "annualized_volatility", + "calmar_ratio", + "information_ratio", + "var_95", + "cvar_95", + "turnover", + "trade_count", + "win_rate", + "profit_factor", +) + + +def canonical_evidence_package_v2_bytes(payload: Mapping[str, Any]) -> bytes: + """Return deterministic UTF-8 JSON bytes and reject non-finite values.""" + + if not isinstance(payload, Mapping): + raise TypeError("evidence package must be a mapping") + return json.dumps( + dict(payload), + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + + +def read_evidence_package_v2_json(path: str | Path) -> dict[str, Any]: + """Read strict JSON: UTF-8 only, no duplicate keys or non-finite numbers.""" + + evidence_path = Path(path) + try: + raw = evidence_path.read_bytes() + except FileNotFoundError: + raise ValueError(f"file not found: {evidence_path}") from None + except OSError as exc: + raise ValueError(f"failed to read file: {exc}") from exc + try: + text = raw.decode("utf-8", errors="strict") + except UnicodeDecodeError as exc: + raise ValueError(f"invalid UTF-8: byte {exc.start}") from exc + + def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError(f"invalid JSON: duplicate key: {key}") + result[key] = value + return result + + def reject_constant(value: str) -> Any: + raise ValueError(f"invalid JSON: non-finite number: {value}") + + try: + payload = json.loads( + text, + object_pairs_hook=reject_duplicates, + parse_constant=reject_constant, + ) + except json.JSONDecodeError as exc: + raise ValueError( + f"invalid JSON: {exc.msg} (line {exc.lineno}, column {exc.colno})" + ) from exc + if not isinstance(payload, dict): + raise ValueError("top-level JSON must be an object") + return payload + + +def validate_evidence_package_v2( + payload: Mapping[str, Any], *, base_dir: str | Path | None = None +) -> tuple[str, ...]: + """Validate the closed v2 contract without optional runtime dependencies.""" + + issues: list[str] = [] + if not isinstance(payload, Mapping): + return ("top-level evidence package must be an object",) + _closed_object(payload, "top-level", _TOP_LEVEL_FIELDS, _TOP_LEVEL_FIELDS, issues) + + if payload.get("schema_version") != STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION: + issues.append("schema_version must equal strategy_evidence_package.v2") + _non_empty_string(payload.get("evidence_package_id"), "evidence_package_id", issues) + generated_at = _timezone_datetime( + payload.get("generated_at"), "generated_at", issues + ) + if payload.get("requested_stage") not in _LEGACY_ALLOWED_REQUESTED_STAGES: + issues.append("requested_stage is unsupported") + + strategy = _object(payload.get("strategy"), "strategy", issues) + if strategy is not None: + _closed_object( + strategy, + "strategy", + {"profile", "domain", "source_revision"}, + {"profile", "domain", "source_revision"}, + issues, + ) + _non_empty_string(strategy.get("profile"), "strategy.profile", issues) + _non_empty_string(strategy.get("domain"), "strategy.domain", issues) + _source_revision( + strategy.get("source_revision"), "strategy.source_revision", issues + ) + + input_provenance = _object( + payload.get("input_provenance"), "input_provenance", issues + ) + if input_provenance is not None: + allowed = { + "source", + "source_revision", + "license", + "usage_scope", + "range", + "timestamp", + "manifest_sha256", + } + _closed_object(input_provenance, "input_provenance", allowed, allowed, issues) + for field in ("source", "source_revision", "license", "usage_scope"): + _non_empty_string( + input_provenance.get(field), f"input_provenance.{field}", issues + ) + input_range = _object( + input_provenance.get("range"), "input_provenance.range", issues + ) + if input_range is not None: + _closed_object( + input_range, + "input_provenance.range", + {"start", "end"}, + {"start", "end"}, + issues, + ) + start = _calendar_date( + input_range.get("start"), "input_provenance.range.start", issues + ) + end = _calendar_date( + input_range.get("end"), "input_provenance.range.end", issues + ) + if start is not None and end is not None and start > end: + issues.append("input_provenance.range boundaries are reversed") + input_timestamp = _timezone_datetime( + input_provenance.get("timestamp"), "input_provenance.timestamp", issues + ) + if ( + generated_at is not None + and input_timestamp is not None + and input_timestamp > generated_at + ): + issues.append("input_provenance.timestamp cannot be after generated_at") + _sha256( + input_provenance.get("manifest_sha256"), + "input_provenance.manifest_sha256", + issues, + ) + + artifacts = _validate_artifacts(payload.get("artifacts"), base_dir, issues) + metrics = _object(payload.get("metrics"), "metrics", issues) + if metrics is not None: + _closed_object( + metrics, "metrics", set(_METRIC_FIELDS), set(_METRIC_FIELDS), issues + ) + for field in _METRIC_FIELDS: + if field == "trade_count": + _finite_integer( + metrics.get(field), f"metrics.{field}", issues, minimum=0 + ) + else: + _finite_number(metrics.get(field), f"metrics.{field}", issues) + win_rate = metrics.get("win_rate") + if _is_finite_number(win_rate) and not 0 <= float(win_rate) <= 1: + issues.append("metrics.win_rate must be between 0 and 1") + + cost_stress = _validate_cost_stress(payload.get("cost_stress"), issues) + _validate_risk_assessment(payload.get("risk_assessment"), issues) + _validate_backtest( + payload.get("backtest"), + strategy, + input_provenance, + cost_stress, + issues, + ) + + digests = _object(payload.get("digests"), "digests", issues) + digest_fields = { + *_ARTIFACT_DIGEST_FIELDS.values(), + "evidence_core_sha256", + "package_sha256", + } + if digests is not None: + _closed_object(digests, "digests", digest_fields, digest_fields, issues) + for field in sorted(digest_fields): + _sha256(digests.get(field), f"digests.{field}", issues) + for artifact_name, digest_field in _ARTIFACT_DIGEST_FIELDS.items(): + artifact = artifacts.get(artifact_name) if artifacts is not None else None + if isinstance(artifact, Mapping) and digests.get( + digest_field + ) != artifact.get("sha256"): + issues.append( + f"digests.{digest_field} must match artifacts.{artifact_name}.sha256" + ) + if input_provenance is not None and digests.get( + "data_manifest_sha256" + ) != input_provenance.get("manifest_sha256"): + issues.append( + "input_provenance.manifest_sha256 must match digests.data_manifest_sha256" + ) + + expected_core_sha256: str | None = None + try: + core = {field: payload[field] for field in _CORE_FIELDS} + expected_core_sha256 = hashlib.sha256( + canonical_evidence_package_v2_bytes(core) + ).hexdigest() + except (KeyError, TypeError, ValueError) as exc: + issues.append(f"evidence core cannot be canonicalized: {exc}") + if ( + expected_core_sha256 is not None + and digests is not None + and digests.get("evidence_core_sha256") != expected_core_sha256 + ): + issues.append("digests.evidence_core_sha256 mismatch") + + acceptance_ok = _validate_human_acceptance( + payload.get("human_acceptance"), + generated_at=generated_at, + evidence_core_sha256=expected_core_sha256, + issues=issues, + ) + _validate_lifecycle_claims( + payload.get("lifecycle_claims"), acceptance_ok=acceptance_ok, issues=issues + ) + + if digests is not None and _SHA256_RE.fullmatch( + str(digests.get("package_sha256") or "") + ): + try: + package_projection = copy.deepcopy(dict(payload)) + package_projection["digests"].pop("package_sha256", None) + expected_package_sha256 = hashlib.sha256( + canonical_evidence_package_v2_bytes(package_projection) + ).hexdigest() + if digests.get("package_sha256") != expected_package_sha256: + issues.append("digests.package_sha256 mismatch") + except (TypeError, ValueError) as exc: + issues.append(f"package cannot be canonicalized: {exc}") + + return tuple(dict.fromkeys(issues)) + + +def validate_strategy_evidence_payload( + payload: Any, *, base_dir: Path | None = None +) -> list[str]: + """Compatibility dispatcher; v2 and legacy lanes share one implementation.""" + + if ( + isinstance(payload, Mapping) + and payload.get("schema_version") == STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION + ): + return list(validate_evidence_package_v2(payload, base_dir=base_dir)) + return _validate_legacy_payload(payload, base_dir=base_dir) + + +def validate_strategy_evidence_file(path: str | Path) -> list[str]: + evidence_path = Path(path) + try: + payload = read_evidence_package_v2_json(evidence_path) + except ValueError as exc: + return [str(exc)] + return validate_strategy_evidence_payload(payload, base_dir=evidence_path.parent) + + +def _validate_artifacts( + value: Any, base_dir: str | Path | None, issues: list[str] +) -> Mapping[str, Any] | None: + artifacts = _object(value, "artifacts", issues) + required = set(_ARTIFACT_DIGEST_FIELDS) + if artifacts is None: + return None + _closed_object(artifacts, "artifacts", required, required, issues) + root = Path(base_dir).resolve() if base_dir is not None else None + for name in sorted(required): + label = f"artifacts.{name}" + artifact = _object(artifacts.get(name), label, issues) + if artifact is None: + continue + _closed_object(artifact, label, {"path", "sha256"}, {"path", "sha256"}, issues) + raw_path = artifact.get("path") + expected_sha256 = artifact.get("sha256") + _artifact_path(raw_path, f"{label}.path", issues) + _sha256(expected_sha256, f"{label}.sha256", issues) + if ( + root is None + or not isinstance(raw_path, str) + or not _valid_artifact_path(raw_path) + ): + continue + candidate = root / raw_path + try: + resolved = candidate.resolve(strict=False) + resolved.relative_to(root) + except (OSError, ValueError): + issues.append( + f"{label}.path must stay within the evidence package directory" + ) + continue + if not resolved.is_file(): + issues.append(f"{label}.path does not exist: {raw_path}") + continue + try: + actual_sha256 = hashlib.sha256(resolved.read_bytes()).hexdigest() + except OSError as exc: + issues.append(f"{label}.path cannot be read: {exc}") + continue + if isinstance(expected_sha256, str) and actual_sha256 != expected_sha256: + issues.append( + f"{label}.sha256 mismatch: expected {expected_sha256}, got {actual_sha256}" + ) + return artifacts + + +def _validate_cost_stress(value: Any, issues: list[str]) -> Mapping[str, Any] | None: + cost_stress = _object(value, "cost_stress", issues) + if cost_stress is None: + return None + _closed_object( + cost_stress, + "cost_stress", + {"scenarios", "status"}, + {"scenarios", "status"}, + issues, + ) + if cost_stress.get("status") != "PASS": + issues.append("cost_stress.status must equal PASS") + scenarios = cost_stress.get("scenarios") + if not isinstance(scenarios, list) or len(scenarios) != 3: + issues.append( + "cost_stress.scenarios must contain ordered 1x, 2x, and 3x scenarios" + ) + return cost_stress + for index, (scenario, multiplier) in enumerate(zip(scenarios, (1, 2, 3))): + label = f"cost_stress.scenarios[{index}]" + item = _object(scenario, label, issues) + if item is None: + continue + _closed_object( + item, + label, + {"multiplier", "total_cost_bps"}, + {"multiplier", "total_cost_bps"}, + issues, + ) + if item.get("multiplier") != multiplier or isinstance( + item.get("multiplier"), bool + ): + issues.append(f"{label}.multiplier must equal {multiplier}") + _finite_number( + item.get("total_cost_bps"), f"{label}.total_cost_bps", issues, minimum=0 + ) + return cost_stress + + +def _validate_risk_assessment(value: Any, issues: list[str]) -> None: + risk = _object(value, "risk_assessment", issues) + if risk is None: + return + allowed = {"status", "standard_id", "standard_sha256"} + _closed_object(risk, "risk_assessment", allowed, allowed, issues) + if risk.get("status") != "PASS": + issues.append("risk_assessment.status must equal PASS") + _non_empty_string(risk.get("standard_id"), "risk_assessment.standard_id", issues) + _sha256(risk.get("standard_sha256"), "risk_assessment.standard_sha256", issues) + + +def _validate_backtest( + value: Any, + strategy: Mapping[str, Any] | None, + input_provenance: Mapping[str, Any] | None, + cost_stress: Mapping[str, Any] | None, + issues: list[str], +) -> None: + backtest = _object(value, "backtest", issues) + if backtest is None: + return + allowed = { + "orchestrator", + "protocol", + "calendar", + "timezone", + "signal_timing", + "execution_timing", + "locked_independent_oos", + "promotion_run", + } + _closed_object(backtest, "backtest", allowed, allowed, issues) + if backtest.get("orchestrator") != "BacktestOrchestrator": + issues.append("backtest.orchestrator must equal BacktestOrchestrator") + if backtest.get("protocol") != "purged_walk_forward.v1": + issues.append("backtest.protocol must equal purged_walk_forward.v1") + for field in ("calendar", "timezone", "signal_timing", "execution_timing"): + _non_empty_string(backtest.get(field), f"backtest.{field}", issues) + locked = _object( + backtest.get("locked_independent_oos"), + "backtest.locked_independent_oos", + issues, + ) + if locked is not None: + fields = {"locked", "independent", "reused_for_selection"} + _closed_object( + locked, "backtest.locked_independent_oos", fields, fields, issues + ) + if locked.get("locked") is not True: + issues.append("backtest.locked_independent_oos.locked must be true") + if locked.get("independent") is not True: + issues.append("backtest.locked_independent_oos.independent must be true") + if locked.get("reused_for_selection") is not False: + issues.append( + "backtest.locked_independent_oos.reused_for_selection must be false" + ) + run = _object(backtest.get("promotion_run"), "backtest.promotion_run", issues) + if run is None: + return + run_fields = { + "strategy_profile", + "domain", + "folds", + "fold_results", + "locked_oos_result", + "locked_oos_start", + "locked_oos_end", + "purge_days", + "embargo_days", + "source_revision", + "cost_model", + } + _closed_object(run, "backtest.promotion_run", run_fields, run_fields, issues) + if strategy is not None: + if run.get("strategy_profile") != strategy.get("profile"): + issues.append( + "backtest.promotion_run.strategy_profile must match strategy.profile" + ) + if run.get("domain") != strategy.get("domain"): + issues.append("backtest.promotion_run.domain must match strategy.domain") + if run.get("source_revision") != strategy.get("source_revision"): + issues.append( + "backtest.promotion_run.source_revision must match strategy.source_revision" + ) + _source_revision( + run.get("source_revision"), "backtest.promotion_run.source_revision", issues + ) + purge_days = _finite_integer( + run.get("purge_days"), "backtest.promotion_run.purge_days", issues, minimum=1 + ) + embargo_days = _finite_integer( + run.get("embargo_days"), + "backtest.promotion_run.embargo_days", + issues, + minimum=1, + ) + locked_start = _calendar_date( + run.get("locked_oos_start"), "backtest.promotion_run.locked_oos_start", issues + ) + locked_end = _calendar_date( + run.get("locked_oos_end"), "backtest.promotion_run.locked_oos_end", issues + ) + if ( + locked_start is not None + and locked_end is not None + and locked_end < _add_calendar_months(locked_start, 12) + ): + issues.append( + "backtest.promotion_run locked OOS must span at least 12 calendar months" + ) + + cost_model = _object( + run.get("cost_model"), "backtest.promotion_run.cost_model", issues + ) + cost_values: dict[str, float] = {} + if cost_model is not None: + cost_fields = { + "model_id", + "commission_bps", + "slippage_bps", + "market_impact_bps", + } + _closed_object( + cost_model, + "backtest.promotion_run.cost_model", + cost_fields, + cost_fields, + issues, + ) + _non_empty_string( + cost_model.get("model_id"), + "backtest.promotion_run.cost_model.model_id", + issues, + ) + for field in ("commission_bps", "slippage_bps", "market_impact_bps"): + number = _finite_number( + cost_model.get(field), + f"backtest.promotion_run.cost_model.{field}", + issues, + minimum=0, + ) + if number is not None: + cost_values[field] = number + if len(cost_values) == 3 and cost_stress is not None: + scenarios = cost_stress.get("scenarios") + if isinstance(scenarios, list) and len(scenarios) == 3: + base_cost = sum(cost_values.values()) + for index, multiplier in enumerate((1, 2, 3)): + scenario = scenarios[index] + if isinstance(scenario, Mapping) and _is_finite_number( + scenario.get("total_cost_bps") + ): + if not math.isclose( + float(scenario["total_cost_bps"]), + base_cost * multiplier, + rel_tol=0, + abs_tol=1e-12, + ): + issues.append( + f"cost_stress.scenarios[{index}].total_cost_bps must match {multiplier}x declared costs" + ) + + folds = run.get("folds") + fold_results = run.get("fold_results") + if not isinstance(folds, list) or len(folds) < 3: + issues.append( + "backtest.promotion_run.folds must contain at least three ordered folds" + ) + folds = [] + if not isinstance(fold_results, list) or len(fold_results) < 3: + issues.append( + "backtest.promotion_run.fold_results must contain at least three results" + ) + fold_results = [] + if len(folds) != len(fold_results): + issues.append("backtest.promotion_run fold/result counts must match") + + parsed_folds: list[tuple[date, date, date, date]] = [] + previous_test_end: date | None = None + for index, fold_value in enumerate(folds): + fold = _object(fold_value, f"backtest.promotion_run.folds[{index}]", issues) + if fold is None: + continue + fold_fields = {"train_start", "train_end", "test_start", "test_end"} + _closed_object( + fold, + f"backtest.promotion_run.folds[{index}]", + fold_fields, + fold_fields, + issues, + ) + boundaries = tuple( + _calendar_date( + fold.get(field), + f"backtest.promotion_run.folds[{index}].{field}", + issues, + ) + for field in ("train_start", "train_end", "test_start", "test_end") + ) + if any(boundary is None for boundary in boundaries): + continue + train_start, train_end, test_start, test_end = boundaries + assert train_start and train_end and test_start and test_end + if train_start > train_end or test_start > test_end: + issues.append( + f"backtest.promotion_run.folds[{index}] boundaries are reversed" + ) + if ( + purge_days is not None + and train_end + timedelta(days=purge_days) >= test_start + ): + issues.append(f"backtest.promotion_run.folds[{index}] violates purge") + if ( + previous_test_end is not None + and embargo_days is not None + and previous_test_end + timedelta(days=embargo_days) >= train_start + ): + issues.append( + f"backtest.promotion_run.folds[{index}] violates ordered embargo" + ) + parsed_folds.append((train_start, train_end, test_start, test_end)) + previous_test_end = test_end + if ( + previous_test_end is not None + and locked_start is not None + and embargo_days is not None + and previous_test_end + timedelta(days=embargo_days) >= locked_start + ): + issues.append("backtest.promotion_run locked OOS overlaps folds or embargo") + if input_provenance is not None and parsed_folds and locked_end is not None: + input_range = input_provenance.get("range") + if isinstance(input_range, Mapping): + try: + input_start = date.fromisoformat(str(input_range.get("start"))) + input_end = date.fromisoformat(str(input_range.get("end"))) + except ValueError: + pass + else: + if input_start > parsed_folds[0][0] or input_end < locked_end: + issues.append( + "input_provenance.range must cover every fold and locked OOS window" + ) + + for index, result in enumerate(fold_results): + fold = parsed_folds[index] if index < len(parsed_folds) else None + _validate_backtest_result( + result, + label=f"backtest.promotion_run.fold_results[{index}]", + strategy=strategy, + run=run, + cost_values=cost_values, + fold_role="test", + fold=fold, + locked_start=locked_start, + locked_end=locked_end, + issues=issues, + ) + _validate_backtest_result( + run.get("locked_oos_result"), + label="backtest.promotion_run.locked_oos_result", + strategy=strategy, + run=run, + cost_values=cost_values, + fold_role="locked_oos", + fold=(None, None, locked_start, locked_end) + if locked_start is not None and locked_end is not None + else None, + locked_start=locked_start, + locked_end=locked_end, + issues=issues, + ) + + +def _validate_backtest_result( + value: Any, + *, + label: str, + strategy: Mapping[str, Any] | None, + run: Mapping[str, Any], + cost_values: Mapping[str, float], + fold_role: str, + fold: tuple[date | None, date | None, date | None, date | None] | None, + locked_start: date | None, + locked_end: date | None, + issues: list[str], +) -> None: + result = _object(value, label, issues) + if result is None: + return + required = { + "strategy_profile", + "domain", + "start_date", + "end_date", + "source_revision", + "cost_model", + "cost_inputs", + "validation_identity", + "sharpe_ratio", + "max_drawdown", + "cagr", + "observation_count", + "run_duration_seconds", + } + _closed_object(result, label, required, _RESULT_FIELDS, issues) + if strategy is not None: + if result.get("strategy_profile") != strategy.get("profile"): + issues.append(f"{label}.strategy_profile mismatch") + if result.get("domain") != strategy.get("domain"): + issues.append(f"{label}.domain mismatch") + if result.get("source_revision") != strategy.get("source_revision"): + issues.append(f"{label}.source_revision mismatch") + cost_model = run.get("cost_model") + model_id = cost_model.get("model_id") if isinstance(cost_model, Mapping) else None + if result.get("cost_model") != model_id: + issues.append(f"{label}.cost_model mismatch") + for field in _RESULT_NUMBER_FIELDS: + if field not in result: + continue + if field in {"observation_count", "param_version"}: + _finite_integer(result.get(field), f"{label}.{field}", issues, minimum=0) + elif result.get(field) is not None: + _finite_number(result.get(field), f"{label}.{field}", issues) + if ( + _is_finite_number(result.get("observation_count")) + and int(result["observation_count"]) <= 0 + ): + issues.append(f"{label}.observation_count must be positive") + if ( + _is_finite_number(result.get("run_duration_seconds")) + and float(result["run_duration_seconds"]) < 0 + ): + issues.append(f"{label}.run_duration_seconds must be non-negative") + start_date = _calendar_date(result.get("start_date"), f"{label}.start_date", issues) + end_date = _calendar_date(result.get("end_date"), f"{label}.end_date", issues) + if fold is not None and (start_date, end_date) != (fold[2], fold[3]): + issues.append(f"{label} dates must match the orchestrator window") + inputs = _object(result.get("cost_inputs"), f"{label}.cost_inputs", issues) + if inputs is not None: + fields = {"commission_bps", "slippage_bps", "market_impact_bps"} + _closed_object(inputs, f"{label}.cost_inputs", fields, fields, issues) + for field in fields: + number = _finite_number( + inputs.get(field), f"{label}.cost_inputs.{field}", issues, minimum=0 + ) + if ( + number is not None + and field in cost_values + and number != cost_values[field] + ): + issues.append(f"{label}.cost_inputs.{field} mismatch") + identity = _object( + result.get("validation_identity"), f"{label}.validation_identity", issues + ) + if identity is None: + return + identity_fields = { + "protocol", + "fold_id", + "fold_role", + "train_start", + "train_end", + "test_start", + "test_end", + "locked_oos_start", + "locked_oos_end", + "purge_days", + "embargo_days", + } + _closed_object( + identity, + f"{label}.validation_identity", + identity_fields, + identity_fields, + issues, + ) + if identity.get("protocol") != "purged_walk_forward.v1": + issues.append(f"{label}.validation_identity.protocol mismatch") + _non_empty_string( + identity.get("fold_id"), f"{label}.validation_identity.fold_id", issues + ) + if identity.get("fold_role") != fold_role: + issues.append(f"{label}.validation_identity.fold_role mismatch") + identity_dates = { + name: _calendar_date( + identity.get(name), f"{label}.validation_identity.{name}", issues + ) + for name in ("test_start", "test_end", "locked_oos_start", "locked_oos_end") + } + for name in ("train_start", "train_end"): + if fold_role == "locked_oos": + if identity.get(name) is not None: + issues.append(f"{label}.validation_identity.{name} must be null") + else: + identity_dates[name] = _calendar_date( + identity.get(name), f"{label}.validation_identity.{name}", issues + ) + if fold is not None: + expected = { + "train_start": fold[0], + "train_end": fold[1], + "test_start": fold[2], + "test_end": fold[3], + } + for name, expected_value in expected.items(): + if fold_role != "locked_oos" or name.startswith("test"): + if identity_dates.get(name) != expected_value: + issues.append(f"{label}.validation_identity.{name} mismatch") + if ( + identity_dates.get("locked_oos_start") != locked_start + or identity_dates.get("locked_oos_end") != locked_end + ): + issues.append(f"{label}.validation_identity locked OOS mismatch") + for name in ("purge_days", "embargo_days"): + value_number = _finite_integer( + identity.get(name), f"{label}.validation_identity.{name}", issues, minimum=1 + ) + if value_number is not None and value_number != run.get(name): + issues.append(f"{label}.validation_identity.{name} mismatch") + + +def _validate_human_acceptance( + value: Any, + *, + generated_at: datetime | None, + evidence_core_sha256: str | None, + issues: list[str], +) -> bool: + if value is None: + return False + acceptance = _object(value, "human_acceptance", issues) + if acceptance is None: + return False + fields = { + "decision", + "acceptance_id", + "actor", + "accepted_at", + "authority_receipt_sha256", + "evidence_core_sha256", + } + _closed_object(acceptance, "human_acceptance", fields, fields, issues) + if acceptance.get("decision") not in {"ACCEPTED", "REJECTED"}: + issues.append("human_acceptance.decision must be ACCEPTED or REJECTED") + _non_empty_string( + acceptance.get("acceptance_id"), "human_acceptance.acceptance_id", issues + ) + _non_empty_string(acceptance.get("actor"), "human_acceptance.actor", issues) + accepted_at = _timezone_datetime( + acceptance.get("accepted_at"), "human_acceptance.accepted_at", issues + ) + _sha256( + acceptance.get("authority_receipt_sha256"), + "human_acceptance.authority_receipt_sha256", + issues, + ) + _sha256( + acceptance.get("evidence_core_sha256"), + "human_acceptance.evidence_core_sha256", + issues, + ) + matches = ( + evidence_core_sha256 is not None + and acceptance.get("evidence_core_sha256") == evidence_core_sha256 + ) + if not matches: + issues.append("human_acceptance.evidence_core_sha256 mismatch") + current = ( + generated_at is not None + and accepted_at is not None + and accepted_at >= generated_at + ) + return acceptance.get("decision") == "ACCEPTED" and matches and current + + +def _validate_lifecycle_claims( + value: Any, *, acceptance_ok: bool, issues: list[str] +) -> None: + claims = _object(value, "lifecycle_claims", issues) + if claims is None: + return + fields = { + "learning_only", + "promotion_eligible", + "live_ready", + "size_zero_required", + "no_order", + } + _closed_object(claims, "lifecycle_claims", fields, fields, issues) + for field in fields: + if not isinstance(claims.get(field), bool): + issues.append(f"lifecycle_claims.{field} must be a boolean") + if claims.get("live_ready") is not False: + issues.append("lifecycle_claims.live_ready must remain false") + if claims.get("size_zero_required") is not True: + issues.append("lifecycle_claims.size_zero_required must remain true") + if claims.get("no_order") is not True: + issues.append("lifecycle_claims.no_order must remain true") + if ( + claims.get("learning_only") is True + and claims.get("promotion_eligible") is not False + ): + issues.append("learning_only requires promotion_eligible=false") + if claims.get("promotion_eligible") is True: + if claims.get("learning_only") is True: + issues.append("learning_only evidence cannot be promotion_eligible") + if not acceptance_ok: + issues.append( + "promotion_eligible=true requires current bound human_acceptance" + ) + + +def _validate_legacy_payload(payload: Any, *, base_dir: Path | None) -> list[str]: + issues: list[str] = [] + if not isinstance(payload, dict): + return ["top-level JSON must be an object"] + required = ( + "schema_version", + "profile", + "market", + "requested_stage", + "generated_at", + "evidence_package_id", + "artifacts", + "validation", + "risk", + "kelly_readiness", + "ai_optimization", + ) + for field in required: + if field not in payload: + issues.append(f"missing required field: {field}") + for field in ("schema_version", "profile", "market", "evidence_package_id"): + _legacy_non_empty(payload, field, issues) + requested_stage = payload.get("requested_stage") + if not isinstance(requested_stage, str) or not requested_stage.strip(): + issues.append("requested_stage must be a non-empty string") + elif requested_stage not in _LEGACY_ALLOWED_REQUESTED_STAGES: + issues.append(f"unsupported requested_stage: {requested_stage!r}") + generated_at = payload.get("generated_at") + if not isinstance(generated_at, str) or not generated_at.strip(): + issues.append("generated_at must be a non-empty string") + elif _parse_legacy_datetime(generated_at) is None: + issues.append(f"generated_at is not a valid date-time: {generated_at!r}") + artifacts = payload.get("artifacts") + if not isinstance(artifacts, dict): + issues.append("artifacts must be an object") + else: + for name in _LEGACY_ARTIFACTS: + artifact = artifacts.get(name) + if not isinstance(artifact, dict): + issues.append(f"artifacts.{name} must be an object") + continue + _legacy_non_empty(artifact, "path", issues, prefix=f"artifacts.{name}") + sha = artifact.get("sha256") + if not isinstance(sha, str) or not re.fullmatch(r"[A-Fa-f0-9]{64}", sha): + issues.append( + f"artifacts.{name}.sha256 must be a 64-character hex string" + ) + elif base_dir is not None: + _legacy_artifact_file(name, artifact, sha, base_dir, issues) + validation = payload.get("validation") + if not isinstance(validation, dict): + issues.append("validation must be an object") + else: + for flag in ("oos_passed", "overfit_report_present"): + if not isinstance(validation.get(flag), bool): + issues.append(f"validation.{flag} must be a boolean") + if ( + requested_stage in {"live_candidate", "runtime_enabled"} + and validation.get(flag) is not True + ): + issues.append(f"{requested_stage} requires validation.{flag}=true") + risk = payload.get("risk") + if not isinstance(risk, dict): + issues.append("risk must be an object") + else: + metrics = risk.get("metrics") + if not isinstance(metrics, dict): + issues.append("risk.metrics must be an object") + else: + for field in _LEGACY_METRICS: + value = metrics.get(field) + if field == "trade_count": + if not _is_integer(value): + issues.append("risk.metrics.trade_count must be an integer") + elif value < 0: + issues.append("risk.metrics.trade_count must be >= 0") + elif not _is_finite_number(value): + issues.append(f"risk.metrics.{field} must be a number") + if ( + _is_finite_number(metrics.get("win_rate")) + and not 0 <= metrics["win_rate"] <= 1 + ): + issues.append("risk.metrics.win_rate must be between 0 and 1") + benchmark = risk.get("benchmark") + if not isinstance(benchmark, dict): + issues.append("risk.benchmark must be an object") + else: + _legacy_non_empty(benchmark, "name", issues, prefix="risk.benchmark") + for field in ("alpha", "beta"): + if not _is_finite_number(benchmark.get(field)): + issues.append(f"risk.benchmark.{field} must be a number") + cost = risk.get("cost_stress") + if not isinstance(cost, dict): + issues.append("risk.cost_stress must be an object") + else: + for field in ("slippage_bps", "commission_bps"): + if not _is_finite_number(cost.get(field)): + issues.append(f"risk.cost_stress.{field} must be a number") + if not isinstance(cost.get("passed"), bool): + issues.append("risk.cost_stress.passed must be a boolean") + oos = risk.get("oos") + if not isinstance(oos, dict): + issues.append("risk.oos must be an object") + else: + for field in ("window_start", "window_end"): + _legacy_non_empty(oos, field, issues, prefix="risk.oos") + if not isinstance(oos.get("locked"), bool): + issues.append("risk.oos.locked must be a boolean") + kelly = payload.get("kelly_readiness") + if not isinstance(kelly, dict): + issues.append("kelly_readiness must be an object") + else: + if kelly.get("level") not in _LEGACY_KELLY_LEVELS: + issues.append("kelly_readiness.level must be one of K0, K1, K2, K3, K4") + if kelly.get("full_kelly_allowed") is not False: + issues.append("kelly_readiness.full_kelly_allowed must be false") + if not isinstance(payload.get("ai_optimization"), dict): + issues.append("ai_optimization must be an object") + return issues + + +def _legacy_artifact_file( + name: str, + artifact: Mapping[str, Any], + expected_sha256: str, + base_dir: Path, + issues: list[str], +) -> None: + raw_path = str(artifact.get("path") or "").strip() + label = f"artifacts.{name}" + if not raw_path: + return + if not _valid_artifact_path(raw_path): + if Path(raw_path).is_absolute(): + issues.append(f"{label}.path must be repo-relative, got absolute path") + else: + issues.append( + f"{label}.path must stay within the evidence package directory" + ) + return + root = base_dir.resolve() + resolved = (root / raw_path).resolve(strict=False) + try: + resolved.relative_to(root) + except ValueError: + issues.append(f"{label}.path must stay within the evidence package directory") + return + if not resolved.is_file(): + issues.append(f"{label}.path does not exist: {raw_path}") + return + actual = hashlib.sha256(resolved.read_bytes()).hexdigest() + if actual.lower() != expected_sha256.lower(): + issues.append( + f"{label}.sha256 mismatch: expected {expected_sha256.lower()}, got {actual.lower()}" + ) + + +def _closed_object( + value: Mapping[str, Any], + label: str, + required: set[str] | frozenset[str], + allowed: set[str] | frozenset[str], + issues: list[str], +) -> None: + for field in sorted(required - set(value)): + issues.append(f"{label} missing required field: {field}") + for field in sorted(set(value) - allowed): + issues.append(f"{label} contains unexpected field: {field}") + + +def _object(value: Any, label: str, issues: list[str]) -> Mapping[str, Any] | None: + if not isinstance(value, Mapping): + issues.append(f"{label} must be an object") + return None + return value + + +def _non_empty_string(value: Any, label: str, issues: list[str]) -> str | None: + if not isinstance(value, str) or not value.strip(): + issues.append(f"{label} must be a non-empty string") + return None + return value + + +def _source_revision(value: Any, label: str, issues: list[str]) -> str | None: + if not isinstance(value, str) or not _SOURCE_REVISION_RE.fullmatch(value): + issues.append(f"{label} must be a lowercase 40-character Git revision") + return None + return value + + +def _sha256(value: Any, label: str, issues: list[str]) -> str | None: + if not isinstance(value, str) or not _SHA256_RE.fullmatch(value): + issues.append(f"{label} must be a lowercase SHA-256") + return None + return value + + +def _finite_number( + value: Any, label: str, issues: list[str], *, minimum: float | None = None +) -> float | None: + if not _is_finite_number(value): + issues.append(f"{label} must be a finite non-boolean number") + return None + number = float(value) + if minimum is not None and number < minimum: + issues.append(f"{label} must be >= {minimum:g}") + return None + return number + + +def _finite_integer( + value: Any, label: str, issues: list[str], *, minimum: int | None = None +) -> int | None: + if not _is_integer(value): + issues.append(f"{label} must be an integer") + return None + if minimum is not None and value < minimum: + issues.append(f"{label} must be >= {minimum}") + return None + return value + + +def _is_finite_number(value: Any) -> bool: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(value) + ) + + +def _is_integer(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _calendar_date(value: Any, label: str, issues: list[str]) -> date | None: + if not isinstance(value, str): + issues.append(f"{label} must be an ISO calendar date") + return None + try: + parsed = date.fromisoformat(value) + except ValueError: + issues.append(f"{label} must be an ISO calendar date") + return None + if parsed.isoformat() != value: + issues.append(f"{label} must use canonical YYYY-MM-DD form") + return None + return parsed + + +def _parse_timezone_datetime(value: str) -> datetime | None: + candidate = value[:-1] + "+00:00" if value.endswith("Z") else value + try: + parsed = datetime.fromisoformat(candidate) + except ValueError: + return None + if parsed.tzinfo is None or parsed.utcoffset() is None: + return None + return parsed + + +def _parse_legacy_datetime(value: str) -> datetime | None: + candidate = value[:-1] + "+00:00" if value.endswith("Z") else value + try: + return datetime.fromisoformat(candidate) + except ValueError: + return None + + +def _timezone_datetime(value: Any, label: str, issues: list[str]) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + issues.append(f"{label} must be a timezone-qualified date-time") + return None + parsed = _parse_timezone_datetime(value) + if parsed is None: + issues.append(f"{label} must be a timezone-qualified date-time") + return parsed + + +def _artifact_path(value: Any, label: str, issues: list[str]) -> None: + if not isinstance(value, str) or not _valid_artifact_path(value): + issues.append( + f"{label} must be a confined repo-relative path without aliases or control characters" + ) + + +def _valid_artifact_path(value: str) -> bool: + if not value or Path(value).is_absolute() or "\\" in value: + return False + if any(ord(character) < 32 or ord(character) == 127 for character in value): + return False + parts = value.split("/") + if any(part in {"", ".", ".."} for part in parts): + return False + return PurePosixPath(value).as_posix() == value + + +def _add_calendar_months(value: date, months: int) -> date: + month_index = value.month - 1 + months + year = value.year + month_index // 12 + month = month_index % 12 + 1 + day = min(value.day, monthrange(year, month)[1]) + return date(year, month, day) + + +def _legacy_non_empty( + payload: Mapping[str, Any], + field: str, + issues: list[str], + *, + prefix: str | None = None, +) -> None: + value = payload.get(field) + label = f"{prefix}.{field}" if prefix else field + if not isinstance(value, str) or not value.strip(): + issues.append(f"{label} must be a non-empty string") diff --git a/src/quant_platform_kit/strategy_lifecycle/live_candidate_notifications.py b/src/quant_platform_kit/strategy_lifecycle/live_candidate_notifications.py index efbd7e37..57472442 100644 --- a/src/quant_platform_kit/strategy_lifecycle/live_candidate_notifications.py +++ b/src/quant_platform_kit/strategy_lifecycle/live_candidate_notifications.py @@ -43,13 +43,19 @@ def to_dict(self) -> dict[str, Any]: "body": self.body, **dict(self.metadata or {}), } - return {key: value for key, value in payload.items() if value not in (None, "", (), [])} + return { + key: value + for key, value in payload.items() + if value not in (None, "", (), []) + } def to_rendered_notification(self) -> RenderedNotification: return RenderedNotification(detailed_text=self.body, compact_text=self.subject) -def build_live_candidate_notification(result: EvidenceGateResult) -> LiveCandidateNotificationEvent | None: +def build_live_candidate_notification( + result: EvidenceGateResult, +) -> LiveCandidateNotificationEvent | None: """Build a notification event from an evidence gate result. Returns None when the requested stage is outside the live-candidate path. @@ -62,13 +68,19 @@ def build_live_candidate_notification(result: EvidenceGateResult) -> LiveCandida strategy_profile = package.strategy_profile domain = package.domain valid = bool(result.valid) - approval_action = "approve" if valid else "hold" + # Evidence validation is never paper/shadow/live authority. + approval_action = "hold" severity = _severity_for(stage, valid) reason = _reason_for(result) evidence_summary = _build_evidence_summary(result) alert_key = _build_alert_key(strategy_profile, domain, stage, approval_action) subject = _build_subject(strategy_profile, domain, stage, approval_action, valid) - body = _build_body(result, reason=reason, evidence_summary=evidence_summary, approval_action=approval_action) + body = _build_body( + result, + reason=reason, + evidence_summary=evidence_summary, + approval_action=approval_action, + ) return LiveCandidateNotificationEvent( strategy_profile=strategy_profile, @@ -91,8 +103,6 @@ def build_live_candidate_notification(result: EvidenceGateResult) -> LiveCandida def _severity_for(stage: str, valid: bool) -> str: - if valid: - return "info" if stage == "runtime_enabled": return "critical" return "warning" @@ -106,7 +116,7 @@ def _reason_for(result: EvidenceGateResult) -> str: parts.append("warnings: " + "; ".join(result.warnings)) if parts: return " | ".join(parts) - return "evidence gate passed" + return "promotion evidence validated; live authority is not granted" def _build_evidence_summary(result: EvidenceGateResult) -> str: @@ -114,23 +124,38 @@ def _build_evidence_summary(result: EvidenceGateResult) -> str: summary_parts = [ _format_summary_item("backtest", _summarize_mapping(package.backtest_summary)), _format_summary_item("drift_notes", _summarize_value(package.drift_notes)), - _format_summary_item("platform_compatibility", _summarize_mapping(package.platform_compatibility or {})), + _format_summary_item( + "platform_compatibility", + _summarize_mapping(package.platform_compatibility or {}), + ), _format_summary_item("plugin_gate", _summarize_value(package.plugin_gate)), - _format_summary_item("target_platforms", ", ".join(package.target_platforms) or "none"), + _format_summary_item( + "target_platforms", ", ".join(package.target_platforms) or "none" + ), ] if package.rollout_notes: - summary_parts.append(_format_summary_item("rollout_notes", _summarize_value(package.rollout_notes))) + summary_parts.append( + _format_summary_item( + "rollout_notes", _summarize_value(package.rollout_notes) + ) + ) if result.warnings: - summary_parts.append(_format_summary_item("warnings", "; ".join(result.warnings))) + summary_parts.append( + _format_summary_item("warnings", "; ".join(result.warnings)) + ) return " | ".join(part for part in summary_parts if part) -def _build_alert_key(strategy_profile: str, domain: str, stage: str, approval_action: str) -> str: +def _build_alert_key( + strategy_profile: str, domain: str, stage: str, approval_action: str +) -> str: return f"lifecycle/live_candidate/{domain}/{strategy_profile}/{stage}/{approval_action}" -def _build_subject(strategy_profile: str, domain: str, stage: str, approval_action: str, valid: bool) -> str: - outcome = "APPROVED" if valid else "HOLD" +def _build_subject( + strategy_profile: str, domain: str, stage: str, approval_action: str, valid: bool +) -> str: + outcome = "HOLD" return f"[{domain}] {outcome} {stage}: {strategy_profile} ({approval_action})" @@ -166,7 +191,15 @@ def _summarize_mapping(value: Mapping[str, Any]) -> str: if not value: return "empty" parts: list[str] = [] - for key in ("observation_count", "sharpe_ratio", "cagr", "max_drawdown", "total_return", "status", "verified"): + for key in ( + "observation_count", + "sharpe_ratio", + "cagr", + "max_drawdown", + "total_return", + "status", + "verified", + ): if key in value and value[key] not in (None, ""): parts.append(f"{key}={value[key]}") if not parts: diff --git a/tests/test_lifecycle_evidence_gate.py b/tests/test_lifecycle_evidence_gate.py index 04c25a55..dc2ed86b 100644 --- a/tests/test_lifecycle_evidence_gate.py +++ b/tests/test_lifecycle_evidence_gate.py @@ -58,6 +58,11 @@ def test_live_package_valid_when_required_sections_present() -> None: ) assert result.valid + assert result.promotion_eligible is False + assert result.live_ready is False + assert result.size_zero_required is True + assert result.no_order is True + assert result.promotion_status == "LEGACY_RESEARCH_ONLY" def test_file_loader_and_cli(tmp_path: Path) -> None: @@ -78,3 +83,23 @@ def test_file_loader_and_cli(tmp_path: Path) -> None: exit_code = main(["evidence", "--file", str(path)]) assert exit_code == 0 + + +def test_legacy_live_and_runtime_requests_are_research_only_holds() -> None: + for stage in ("live_candidate", "runtime_enabled"): + result = validate_evidence_package( + { + "strategy_profile": "legacy_profile", + "domain": "us_equity", + "requested_stage": stage, + "target_platforms": ["ibkr"], + "backtest_summary": {"observation_count": 252}, + "drift_notes": {"status": "stable"}, + "platform_compatibility": {"verified": True}, + } + ) + + assert result.valid + assert result.promotion_eligible is False + assert result.live_ready is False + assert result.promotion_status == "LEGACY_RESEARCH_ONLY" diff --git a/tests/test_lifecycle_live_candidate_notifications.py b/tests/test_lifecycle_live_candidate_notifications.py index 560182a1..5db99374 100644 --- a/tests/test_lifecycle_live_candidate_notifications.py +++ b/tests/test_lifecycle_live_candidate_notifications.py @@ -2,13 +2,15 @@ from __future__ import annotations -from quant_platform_kit.strategy_lifecycle.evidence_gate import validate_evidence_package +from quant_platform_kit.strategy_lifecycle.evidence_gate import ( + validate_evidence_package, +) from quant_platform_kit.strategy_lifecycle.live_candidate_notifications import ( build_live_candidate_notification, ) -def test_builds_live_candidate_notification_for_valid_package() -> None: +def test_valid_legacy_live_candidate_notification_is_still_hold_only() -> None: result = validate_evidence_package( { "strategy_profile": "cn_chinext_growth_momentum_quality", @@ -28,14 +30,37 @@ def test_builds_live_candidate_notification_for_valid_package() -> None: assert event.strategy_profile == "cn_chinext_growth_momentum_quality" assert event.domain == "cn_equity" assert event.stage == "live_candidate" - assert event.approval_action == "approve" - assert event.severity == "info" - assert event.alert_key == "lifecycle/live_candidate/cn_equity/cn_chinext_growth_momentum_quality/live_candidate/approve" + assert event.approval_action == "hold" + assert event.severity == "warning" + assert ( + event.alert_key + == "lifecycle/live_candidate/cn_equity/cn_chinext_growth_momentum_quality/live_candidate/hold" + ) assert "backtest=observation_count=252" in event.evidence_summary assert "platform_compatibility=verified=True" in event.evidence_summary rendered = event.to_rendered_notification() assert rendered.compact_text == event.subject - assert "Approval Action: approve" in rendered.detailed_text + assert "Approval Action: hold" in rendered.detailed_text + + +def test_runtime_enabled_notification_can_never_approve() -> None: + result = validate_evidence_package( + { + "strategy_profile": "legacy_profile", + "domain": "us_equity", + "requested_stage": "runtime_enabled", + "target_platforms": ["ibkr"], + "backtest_summary": {"observation_count": 252}, + "drift_notes": {"status": "stable"}, + "platform_compatibility": {"verified": True}, + } + ) + + event = build_live_candidate_notification(result) + + assert event is not None + assert event.approval_action == "hold" + assert event.severity == "critical" def test_builds_hold_notification_for_invalid_live_package() -> None: diff --git a/tests/test_strategy_evidence_package_v2_contract.py b/tests/test_strategy_evidence_package_v2_contract.py new file mode 100644 index 00000000..041823d8 --- /dev/null +++ b/tests/test_strategy_evidence_package_v2_contract.py @@ -0,0 +1,803 @@ +from __future__ import annotations + +import copy +import hashlib +import importlib +import json +import math +from dataclasses import fields +from datetime import date +from importlib import resources +from pathlib import Path +from typing import Any + +import pytest +from jsonschema import Draft202012Validator, FormatChecker + + +SCHEMA_VERSION = "strategy_evidence_package.v2" +SOURCE_REVISION = "a" * 40 +CORE_FIELDS = ( + "schema_version", + "evidence_package_id", + "generated_at", + "requested_stage", + "strategy", + "input_provenance", + "backtest", + "artifacts", + "metrics", + "cost_stress", + "risk_assessment", +) +ARTIFACT_NAMES = ( + "config", + "data_manifest", + "backtest", + "risk", + "information_coefficient", + "cost_model", +) + + +def _canonical(value: Any) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + + +def _sha256(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _artifact_records(tmp_path: Path) -> dict[str, dict[str, str]]: + artifact_dir = tmp_path / "artifacts" + artifact_dir.mkdir(exist_ok=True) + records: dict[str, dict[str, str]] = {} + for name in ARTIFACT_NAMES: + path = artifact_dir / f"{name}.json" + path.write_bytes(_canonical({"artifact": name})) + records[name] = { + "path": path.relative_to(tmp_path).as_posix(), + "sha256": _sha256(path.read_bytes()), + } + return records + + +def _folds() -> list[dict[str, str]]: + return [ + { + "train_start": "2015-01-01", + "train_end": "2015-12-31", + "test_start": "2016-01-03", + "test_end": "2016-06-30", + }, + { + "train_start": "2016-07-03", + "train_end": "2017-06-30", + "test_start": "2017-07-03", + "test_end": "2017-12-31", + }, + { + "train_start": "2018-01-03", + "train_end": "2018-12-31", + "test_start": "2019-01-03", + "test_end": "2019-06-30", + }, + ] + + +def _result( + *, + fold_id: str, + fold_role: str, + test_start: str, + test_end: str, + train_start: str | None, + train_end: str | None, +) -> dict[str, Any]: + return { + "strategy_profile": "alpha_momentum", + "domain": "us_equity", + "sharpe_ratio": 1.2, + "max_drawdown": -0.1, + "cagr": 0.15, + "observation_count": 126, + "run_duration_seconds": 1.0, + "start_date": test_start, + "end_date": test_end, + "source_revision": SOURCE_REVISION, + "cost_model": "retail_us_equity_v1", + "cost_inputs": { + "commission_bps": 1.0, + "slippage_bps": 2.0, + "market_impact_bps": 0.5, + }, + "validation_identity": { + "protocol": "purged_walk_forward.v1", + "fold_id": fold_id, + "fold_role": fold_role, + "train_start": train_start, + "train_end": train_end, + "test_start": test_start, + "test_end": test_end, + "locked_oos_start": "2019-07-03", + "locked_oos_end": "2020-07-03", + "purge_days": 1, + "embargo_days": 1, + }, + } + + +def _promotion_run() -> dict[str, Any]: + folds = _folds() + fold_results = [ + _result( + fold_id=f"promotion_wf{index}", + fold_role="test", + test_start=fold["test_start"], + test_end=fold["test_end"], + train_start=fold["train_start"], + train_end=fold["train_end"], + ) + for index, fold in enumerate(folds) + ] + return { + "strategy_profile": "alpha_momentum", + "domain": "us_equity", + "folds": folds, + "fold_results": fold_results, + "locked_oos_result": _result( + fold_id="promotion_locked_oos", + fold_role="locked_oos", + test_start="2019-07-03", + test_end="2020-07-03", + train_start=None, + train_end=None, + ), + "locked_oos_start": "2019-07-03", + "locked_oos_end": "2020-07-03", + "purge_days": 1, + "embargo_days": 1, + "source_revision": SOURCE_REVISION, + "cost_model": { + "model_id": "retail_us_equity_v1", + "commission_bps": 1.0, + "slippage_bps": 2.0, + "market_impact_bps": 0.5, + }, + } + + +def _refresh_digests(payload: dict[str, Any], *, bind_acceptance: bool = True) -> None: + core = {key: payload[key] for key in CORE_FIELDS} + core_sha256 = _sha256(_canonical(core)) + payload["digests"]["evidence_core_sha256"] = core_sha256 + acceptance = payload.get("human_acceptance") + if bind_acceptance and isinstance(acceptance, dict): + acceptance["evidence_core_sha256"] = core_sha256 + package_projection = copy.deepcopy(payload) + package_projection["digests"].pop("package_sha256", None) + payload["digests"]["package_sha256"] = _sha256(_canonical(package_projection)) + + +def _payload( + tmp_path: Path, + *, + learning_only: bool = False, + accepted: bool = True, + promotion_eligible: bool | None = None, + requested_stage: str = "ai_monitored_candidate", +) -> dict[str, Any]: + artifacts = _artifact_records(tmp_path) + if promotion_eligible is None: + promotion_eligible = accepted and not learning_only + payload: dict[str, Any] = { + "schema_version": SCHEMA_VERSION, + "evidence_package_id": "pkg_20260805_001", + "generated_at": "2026-08-05T00:00:00Z", + "requested_stage": requested_stage, + "strategy": { + "profile": "alpha_momentum", + "domain": "us_equity", + "source_revision": SOURCE_REVISION, + }, + "input_provenance": { + "source": "licensed_fixture_provider", + "source_revision": "dataset-2026-08-01", + "license": "personal_internal_research", + "usage_scope": "non-commercial internal research", + "range": {"start": "2015-01-01", "end": "2020-07-03"}, + "timestamp": "2026-08-04T23:00:00Z", + "manifest_sha256": artifacts["data_manifest"]["sha256"], + }, + "backtest": { + "orchestrator": "BacktestOrchestrator", + "protocol": "purged_walk_forward.v1", + "calendar": "XNYS", + "timezone": "America/New_York", + "signal_timing": "close_t", + "execution_timing": "open_t_plus_1", + "locked_independent_oos": { + "locked": True, + "independent": True, + "reused_for_selection": False, + }, + "promotion_run": _promotion_run(), + }, + "artifacts": artifacts, + "metrics": { + "sharpe_ratio": 1.42, + "sortino_ratio": 2.15, + "max_drawdown": -0.12, + "annualized_return": 0.18, + "annualized_volatility": 0.22, + "calmar_ratio": 1.5, + "information_ratio": 0.83, + "information_coefficient": 0.07, + "var_95": -0.03, + "cvar_95": -0.05, + "turnover": 1.8, + "trade_count": 128, + "win_rate": 0.57, + "profit_factor": 1.34, + }, + "cost_stress": { + "scenarios": [ + {"multiplier": 1, "total_cost_bps": 3.5}, + {"multiplier": 2, "total_cost_bps": 7.0}, + {"multiplier": 3, "total_cost_bps": 10.5}, + ], + "status": "PASS", + }, + "risk_assessment": { + "status": "PASS", + "standard_id": "docs/strategy_promotion_risk_standard.zh-CN.md", + "standard_sha256": "f" * 64, + }, + "digests": { + "config_sha256": artifacts["config"]["sha256"], + "data_manifest_sha256": artifacts["data_manifest"]["sha256"], + "backtest_sha256": artifacts["backtest"]["sha256"], + "risk_sha256": artifacts["risk"]["sha256"], + "information_coefficient_sha256": artifacts["information_coefficient"][ + "sha256" + ], + "cost_model_sha256": artifacts["cost_model"]["sha256"], + "evidence_core_sha256": "0" * 64, + "package_sha256": "0" * 64, + }, + "human_acceptance": None, + "lifecycle_claims": { + "learning_only": learning_only, + "promotion_eligible": promotion_eligible, + "live_ready": False, + "size_zero_required": True, + "no_order": True, + }, + } + if accepted and not learning_only: + payload["human_acceptance"] = { + "decision": "ACCEPTED", + "acceptance_id": "promotion_acceptance_20260805", + "actor": "human:operator", + "accepted_at": "2026-08-05T00:05:00Z", + "authority_receipt_sha256": "e" * 64, + "evidence_core_sha256": "0" * 64, + } + _refresh_digests(payload) + return payload + + +def _module(): + return importlib.import_module( + "quant_platform_kit.strategy_lifecycle.evidence_package_v2" + ) + + +def _issues( + payload: dict[str, Any], *, base_dir: Path | None = None +) -> tuple[str, ...]: + return tuple(_module().validate_evidence_package_v2(payload, base_dir=base_dir)) + + +def _set_path(payload: dict[str, Any], dotted_path: str, value: Any) -> None: + target: Any = payload + parts = dotted_path.split(".") + for part in parts[:-1]: + target = target[int(part)] if part.isdigit() else target[part] + if parts[-1].isdigit(): + target[int(parts[-1])] = value + else: + target[parts[-1]] = value + + +def test_v2_public_api_and_packaged_schema_exist() -> None: + module = _module() + + assert module.STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION == SCHEMA_VERSION + assert callable(module.read_evidence_package_v2_json) + assert callable(module.validate_evidence_package_v2) + assert callable(module.canonical_evidence_package_v2_bytes) + schema = resources.files("quant_platform_kit.schemas").joinpath( + "strategy-evidence-package.v2.schema.json" + ) + assert schema.is_file() + + +def test_valid_learning_package_enforces_non_live_truth_vector(tmp_path: Path) -> None: + payload = _payload(tmp_path, learning_only=True, accepted=False) + + assert _issues(payload, base_dir=tmp_path) == () + assert payload["lifecycle_claims"] == { + "learning_only": True, + "promotion_eligible": False, + "live_ready": False, + "size_zero_required": True, + "no_order": True, + } + + +def test_complete_package_without_acceptance_is_human_required(tmp_path: Path) -> None: + from quant_platform_kit.strategy_lifecycle.evidence_gate import ( + validate_evidence_package, + ) + + payload = _payload(tmp_path, accepted=False, promotion_eligible=False) + result = validate_evidence_package(payload, base_dir=tmp_path) + + assert result.valid + assert result.promotion_status == "HUMAN_REQUIRED" + assert result.promotion_eligible is False + assert result.live_ready is False + assert result.size_zero_required is True + assert result.no_order is True + + +def test_bound_acceptance_can_promote_but_never_authorizes_live(tmp_path: Path) -> None: + from quant_platform_kit.strategy_lifecycle.evidence_gate import ( + validate_evidence_package, + ) + + result = validate_evidence_package(_payload(tmp_path), base_dir=tmp_path) + + assert result.valid + assert result.promotion_status == "PROMOTION_ELIGIBLE" + assert result.promotion_eligible is True + assert result.live_ready is False + assert result.size_zero_required is True + assert result.no_order is True + + +def test_accepts_exact_backtest_orchestrator_promotion_output(tmp_path: Path) -> None: + from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import ( + BacktestOrchestrator, + ) + from quant_platform_kit.strategy_lifecycle.contracts import ( + BacktestResult, + PromotionCostModel, + PurgedWalkForwardFold, + ) + from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore + + class Runner: + @staticmethod + def _result(start_date: date, end_date: date) -> BacktestResult: + return BacktestResult( + strategy_profile="alpha_momentum", + domain="us_equity", + param_set_id="candidate", + params={}, + sharpe_ratio=1.2, + max_drawdown=-0.1, + cagr=0.15, + start_date=start_date, + end_date=end_date, + observation_count=126, + run_duration_seconds=1.0, + ) + + def run_purged_fold( + self, + strategy_profile: str, + params: dict[str, Any], + *, + fold: PurgedWalkForwardFold, + purge_days: int, + embargo_days: int, + cost_model: PromotionCostModel, + ) -> BacktestResult: + return self._result(fold.test_start, fold.test_end) + + def run_locked_oos( + self, + strategy_profile: str, + params: dict[str, Any], + *, + start_date: date, + end_date: date, + cost_model: PromotionCostModel, + ) -> BacktestResult: + return self._result(start_date, end_date) + + orchestrator = BacktestOrchestrator( + store=PerformanceStore(local_root=tmp_path / "store") + ) + orchestrator.register_runner("us_equity", Runner()) + folds = [ + PurgedWalkForwardFold( + date.fromisoformat(fold["train_start"]), + date.fromisoformat(fold["train_end"]), + date.fromisoformat(fold["test_start"]), + date.fromisoformat(fold["test_end"]), + ) + for fold in _folds() + ] + run = orchestrator.run_promotion( + "alpha_momentum", + domain="us_equity", + params={}, + folds=folds, + locked_oos_start=date(2019, 7, 3), + locked_oos_end=date(2020, 7, 3), + purge_days=1, + embargo_days=1, + source_revision=SOURCE_REVISION, + cost_model=PromotionCostModel( + model_id="retail_us_equity_v1", + commission_bps=1.0, + slippage_bps=2.0, + market_impact_bps=0.5, + ), + ) + payload = _payload(tmp_path) + payload["backtest"]["promotion_run"] = run.to_dict() + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) == () + + +@pytest.mark.parametrize( + ("path", "value"), + [ + ("metrics.sharpe_ratio", None), + ("metrics.max_drawdown", True), + ("metrics.var_95", math.nan), + ("metrics.information_coefficient", math.inf), + ("cost_stress.scenarios.0.total_cost_bps", False), + ("backtest.promotion_run.cost_model.slippage_bps", math.nan), + ("backtest.promotion_run.fold_results.0.cost_inputs.commission_bps", math.inf), + ], +) +def test_metrics_and_costs_are_present_non_bool_and_finite( + tmp_path: Path, path: str, value: Any +) -> None: + payload = _payload(tmp_path) + if value is None: + target: Any = payload + parts = path.split(".") + for part in parts[:-1]: + target = target[int(part)] if part.isdigit() else target[part] + target.pop(parts[-1]) + else: + _set_path(payload, path, value) + if not (isinstance(value, float) and not math.isfinite(value)): + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) + + +@pytest.mark.parametrize( + "raw", + [ + b'{"schema_version":"strategy_evidence_package.v2","schema_version":"duplicate"}', + b'{"value":NaN}', + b'{"value":Infinity}', + b'{"value":-Infinity}', + b'{"value":1} trailing', + b'{"value":"\xff"}', + ], +) +def test_strict_json_reader_rejects_ambiguous_or_invalid_json( + tmp_path: Path, raw: bytes +) -> None: + path = tmp_path / "invalid.json" + path.write_bytes(raw) + + with pytest.raises(ValueError): + _module().read_evidence_package_v2_json(path) + + +def test_schema_and_python_parity_for_expressible_constraints(tmp_path: Path) -> None: + module = _module() + payload = _payload(tmp_path) + schema = json.loads( + resources.files("quant_platform_kit.schemas") + .joinpath("strategy-evidence-package.v2.schema.json") + .read_text(encoding="utf-8") + ) + validator = Draft202012Validator(schema, format_checker=FormatChecker()) + + assert list(validator.iter_errors(payload)) == [] + assert tuple(module.validate_evidence_package_v2(payload, base_dir=tmp_path)) == () + + for dotted_path, value in ( + ("artifacts.config.path", "artifacts/config\u0000.json"), + ("metrics.sharpe_ratio", True), + ("backtest.promotion_run.purge_days", 0), + ("input_provenance.timestamp", "2026-08-04T23:00:00"), + ): + candidate = copy.deepcopy(payload) + _set_path(candidate, dotted_path, value) + _refresh_digests(candidate) + assert list(validator.iter_errors(candidate)), dotted_path + assert tuple( + module.validate_evidence_package_v2(candidate, base_dir=tmp_path) + ), dotted_path + + +@pytest.mark.parametrize( + "bad_path", + [ + "../outside.json", + "/tmp/absolute.json", + "artifacts/../outside.json", + "artifacts/config\n.json", + ], +) +def test_artifact_paths_are_repo_relative_confined_and_unaliased( + tmp_path: Path, bad_path: str +) -> None: + payload = _payload(tmp_path) + payload["artifacts"]["config"]["path"] = bad_path + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) + + +def test_artifact_symlink_escape_is_rejected(tmp_path: Path) -> None: + payload = _payload(tmp_path) + outside = tmp_path.parent / "qpk-evidence-outside.json" + outside.write_text("outside", encoding="utf-8") + link = tmp_path / "artifacts" / "escape.json" + link.symlink_to(outside) + payload["artifacts"]["config"] = { + "path": link.relative_to(tmp_path).as_posix(), + "sha256": _sha256(outside.read_bytes()), + } + payload["digests"]["config_sha256"] = payload["artifacts"]["config"]["sha256"] + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) + + +def test_artifact_bytes_and_all_digest_bindings_are_verified(tmp_path: Path) -> None: + payload = _payload(tmp_path) + config_path = tmp_path / payload["artifacts"]["config"]["path"] + config_path.write_text("changed", encoding="utf-8") + + assert any( + "config" in issue and "sha256" in issue + for issue in _issues(payload, base_dir=tmp_path) + ) + + payload = _payload(tmp_path) + payload["digests"]["risk_sha256"] = "1" * 64 + _refresh_digests(payload) + assert any("risk_sha256" in issue for issue in _issues(payload, base_dir=tmp_path)) + + payload = _payload(tmp_path) + payload["human_acceptance"]["evidence_core_sha256"] = "2" * 64 + _refresh_digests(payload, bind_acceptance=False) + assert any( + "evidence_core_sha256" in issue for issue in _issues(payload, base_dir=tmp_path) + ) + + payload = _payload(tmp_path) + payload["digests"]["package_sha256"] = "3" * 64 + assert any( + "package_sha256" in issue for issue in _issues(payload, base_dir=tmp_path) + ) + + +@pytest.mark.parametrize( + "mutation", + [ + "few_folds", + "misordered", + "zero_purge", + "zero_embargo", + "unlocked", + "reused", + "short_oos", + ], +) +def test_backtest_orchestrator_identity_fails_closed( + tmp_path: Path, mutation: str +) -> None: + payload = _payload(tmp_path) + run = payload["backtest"]["promotion_run"] + if mutation == "few_folds": + run["folds"] = run["folds"][:2] + run["fold_results"] = run["fold_results"][:2] + elif mutation == "misordered": + run["folds"][0], run["folds"][1] = run["folds"][1], run["folds"][0] + elif mutation == "zero_purge": + run["purge_days"] = 0 + elif mutation == "zero_embargo": + run["embargo_days"] = 0 + elif mutation == "unlocked": + payload["backtest"]["locked_independent_oos"]["locked"] = False + elif mutation == "reused": + payload["backtest"]["locked_independent_oos"]["reused_for_selection"] = True + else: + run["locked_oos_end"] = "2020-07-02" + run["locked_oos_result"]["end_date"] = "2020-07-02" + run["locked_oos_result"]["validation_identity"]["test_end"] = "2020-07-02" + run["locked_oos_result"]["validation_identity"]["locked_oos_end"] = "2020-07-02" + for result in run["fold_results"]: + result["validation_identity"]["locked_oos_end"] = "2020-07-02" + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) + + +def test_locked_oos_uses_calendar_months_across_leap_day(tmp_path: Path) -> None: + payload = _payload(tmp_path) + run = payload["backtest"]["promotion_run"] + run["locked_oos_start"] = "2020-02-29" + run["locked_oos_end"] = "2021-02-28" + payload["input_provenance"]["range"]["end"] = "2021-02-28" + run["locked_oos_result"]["start_date"] = "2020-02-29" + run["locked_oos_result"]["end_date"] = "2021-02-28" + run["locked_oos_result"]["validation_identity"]["test_start"] = "2020-02-29" + run["locked_oos_result"]["validation_identity"]["test_end"] = "2021-02-28" + for result in [*run["fold_results"], run["locked_oos_result"]]: + identity = result["validation_identity"] + identity["locked_oos_start"] = "2020-02-29" + identity["locked_oos_end"] = "2021-02-28" + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) == () + + +@pytest.mark.parametrize( + ("path", "value"), + [ + ("strategy.source_revision", "b" * 40), + ("input_provenance.source_revision", ""), + ("input_provenance.license", ""), + ("input_provenance.range.start", "2016-01-01"), + ("input_provenance.timestamp", "2026-08-04T23:00:00"), + ("input_provenance.manifest_sha256", "b" * 64), + ("backtest.orchestrator", "CallerLoop"), + ("backtest.protocol", "walk_forward.v0"), + ("backtest.calendar", ""), + ("backtest.timezone", ""), + ("backtest.signal_timing", ""), + ("backtest.execution_timing", ""), + ("backtest.promotion_run.strategy_profile", "other"), + ("backtest.promotion_run.cost_model.model_id", "other_costs"), + ("cost_stress.scenarios.1.total_cost_bps", 8.0), + ("digests.config_sha256", "b" * 64), + ("risk_assessment.status", "WARN"), + ("cost_stress.status", "WARN"), + ], +) +def test_strategy_input_backtest_cost_and_risk_identities_are_bound( + tmp_path: Path, path: str, value: Any +) -> None: + payload = _payload(tmp_path) + _set_path(payload, path, value) + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) + + +@pytest.mark.parametrize( + "condition", ["missing", "rejected", "stale", "digest_mismatch"] +) +def test_human_acceptance_must_be_current_accepted_and_bound( + tmp_path: Path, condition: str +) -> None: + payload = _payload(tmp_path) + if condition == "missing": + payload["human_acceptance"] = None + elif condition == "rejected": + payload["human_acceptance"]["decision"] = "REJECTED" + elif condition == "stale": + payload["human_acceptance"]["accepted_at"] = "2026-08-04T23:59:59Z" + else: + payload["human_acceptance"]["evidence_core_sha256"] = "d" * 64 + _refresh_digests(payload, bind_acceptance=False) + + issues = _issues(payload, base_dir=tmp_path) + assert issues + assert any( + "promotion_eligible" in issue or "human_acceptance" in issue for issue in issues + ) + + +def test_requested_stage_and_caller_flags_cannot_create_live_truth( + tmp_path: Path, +) -> None: + payload = _payload(tmp_path, requested_stage="runtime_enabled") + payload["lifecycle_claims"]["live_ready"] = True + payload["ci_passed"] = True + _refresh_digests(payload) + + issues = _issues(payload, base_dir=tmp_path) + assert issues + assert any("live_ready" in issue for issue in issues) + assert any("unexpected" in issue or "additional" in issue for issue in issues) + + +def test_no_unfrozen_performance_threshold_is_invented(tmp_path: Path) -> None: + payload = _payload(tmp_path) + payload["metrics"]["sharpe_ratio"] = -7.0 + payload["metrics"]["annualized_return"] = -0.9 + payload["metrics"]["information_coefficient"] = -0.8 + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) == () + + +def test_backtest_result_identity_is_cross_checked(tmp_path: Path) -> None: + payload = _payload(tmp_path) + result = payload["backtest"]["promotion_run"]["fold_results"][0] + result["source_revision"] = "b" * 40 + result["validation_identity"]["protocol"] = "caller_label.v1" + _refresh_digests(payload) + + assert _issues(payload, base_dir=tmp_path) + + +def test_canonical_bytes_are_stable_and_reject_non_finite_values( + tmp_path: Path, +) -> None: + module = _module() + payload = _payload(tmp_path) + reordered = dict(reversed(list(payload.items()))) + + assert module.canonical_evidence_package_v2_bytes( + payload + ) == module.canonical_evidence_package_v2_bytes(reordered) + payload["metrics"]["sharpe_ratio"] = math.nan + with pytest.raises((TypeError, ValueError)): + module.canonical_evidence_package_v2_bytes(payload) + + +def test_legacy_dataclass_positional_order_is_unchanged() -> None: + from quant_platform_kit.strategy_lifecycle.evidence_gate import ( + EvidenceGateResult, + EvidencePackage, + ) + + assert [field.name for field in fields(EvidencePackage)][:12] == [ + "strategy_profile", + "domain", + "requested_stage", + "target_platforms", + "backtest_summary", + "drift_notes", + "platform_compatibility", + "plugin_gate", + "rollout_notes", + "operator_notes", + "evidence_version", + "submitted_at", + ] + assert [field.name for field in fields(EvidenceGateResult)][:4] == [ + "valid", + "package", + "issues", + "warnings", + ] + legacy = EvidenceGateResult( + True, EvidencePackage("legacy_profile", "us_equity", "research_backtest_only") + ) + assert set(legacy.to_dict()) == {"valid", "issues", "warnings", "package"} diff --git a/tests/test_strategy_evidence_package_validator.py b/tests/test_strategy_evidence_package_validator.py index 780d4759..5775b0b4 100644 --- a/tests/test_strategy_evidence_package_validator.py +++ b/tests/test_strategy_evidence_package_validator.py @@ -1,6 +1,7 @@ from __future__ import annotations import hashlib +import importlib import json import subprocess import sys @@ -78,7 +79,16 @@ def test_validate_payload_accepts_minimal_valid_package() -> None: assert validate_payload(_valid_payload()) == [] -def test_validate_payload_rejects_live_package_without_required_validation_flags() -> None: +def test_legacy_validator_keeps_naive_generated_at_compatibility() -> None: + payload = _valid_payload() + payload["generated_at"] = "2026-07-07T00:00:00" + + assert validate_payload(payload) == [] + + +def test_validate_payload_rejects_live_package_without_required_validation_flags() -> ( + None +): payload = _valid_payload() payload["validation"] = { "oos_passed": False, @@ -91,7 +101,9 @@ def test_validate_payload_rejects_live_package_without_required_validation_flags assert "live_candidate requires validation.overfit_report_present=true" in issues -def test_validate_payload_rejects_runtime_enabled_package_without_required_validation_flags() -> None: +def test_validate_payload_rejects_runtime_enabled_package_without_required_validation_flags() -> ( + None +): payload = _valid_payload() payload["requested_stage"] = "runtime_enabled" payload["validation"] = { @@ -130,7 +142,9 @@ def test_cli_exit_code_for_invalid_json(tmp_path: Path) -> None: path = tmp_path / "evidence.json" path.write_text("{", encoding="utf-8") - assert validate_file(path) == ["invalid JSON: Expecting property name enclosed in double quotes (line 1, column 2)"] + assert validate_file(path) == [ + "invalid JSON: Expecting property name enclosed in double quotes (line 1, column 2)" + ] proc = subprocess.run( [sys.executable, "scripts/validate_strategy_evidence_package.py", str(path)], @@ -191,3 +205,39 @@ def test_validate_file_rejects_sha256_mismatch(tmp_path: Path) -> None: issues = validate_file(path) assert any("artifacts.returns.sha256 mismatch" in issue for issue in issues) + + +def test_script_reexports_the_canonical_v2_api() -> None: + canonical = importlib.import_module( + "quant_platform_kit.strategy_lifecycle.evidence_package_v2" + ) + wrapper = importlib.import_module("scripts.validate_strategy_evidence_package") + + assert ( + wrapper.STRATEGY_EVIDENCE_PACKAGE_SCHEMA_VERSION + == "strategy_evidence_package.v2" + ) + assert ( + wrapper.read_evidence_package_v2_json is canonical.read_evidence_package_v2_json + ) + assert ( + wrapper.validate_evidence_package_v2 is canonical.validate_evidence_package_v2 + ) + assert ( + wrapper.canonical_evidence_package_v2_bytes + is canonical.canonical_evidence_package_v2_bytes + ) + + +def test_validate_file_rejects_duplicate_keys_and_non_finite_json( + tmp_path: Path, +) -> None: + duplicate = tmp_path / "duplicate.json" + duplicate.write_text( + '{"schema_version":"one","schema_version":"two"}', encoding="utf-8" + ) + non_finite = tmp_path / "non-finite.json" + non_finite.write_text('{"value":NaN}', encoding="utf-8") + + assert any("duplicate" in issue for issue in validate_file(duplicate)) + assert any("non-finite" in issue for issue in validate_file(non_finite)) diff --git a/tests/test_validate_strategy_evidence_package.py b/tests/test_validate_strategy_evidence_package.py index 1e270cc1..01f3a531 100644 --- a/tests/test_validate_strategy_evidence_package.py +++ b/tests/test_validate_strategy_evidence_package.py @@ -78,7 +78,9 @@ def _valid_payload() -> dict[str, object]: } -def _run_validator(tmp_path: Path, payload: dict[str, object]) -> subprocess.CompletedProcess[str]: +def _run_validator( + tmp_path: Path, payload: dict[str, object] +) -> subprocess.CompletedProcess[str]: path = tmp_path / "evidence.json" path.write_text(json.dumps(payload), encoding="utf-8") return subprocess.run( @@ -123,7 +125,9 @@ def test_missing_required_artifact_fails(tmp_path: Path) -> None: @pytest.mark.parametrize("flag", ["oos_passed", "overfit_report_present"]) -def test_live_candidate_rejects_false_validation_flags(tmp_path: Path, flag: str) -> None: +def test_live_candidate_rejects_false_validation_flags( + tmp_path: Path, flag: str +) -> None: payload = _valid_payload() payload["validation"] = dict(payload["validation"]) payload["validation"][flag] = False @@ -153,3 +157,19 @@ def test_missing_risk_metric_fails(tmp_path: Path) -> None: assert result.returncode == 1 assert "risk.metrics.win_rate must be a number" in result.stderr + + +def test_compatibility_cli_rejects_duplicate_keys(tmp_path: Path) -> None: + path = tmp_path / "duplicate.json" + path.write_text('{"schema_version":"one","schema_version":"two"}', encoding="utf-8") + + result = subprocess.run( + [sys.executable, str(SCRIPT), str(path)], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 1 + assert "duplicate" in result.stderr