diff --git a/.github/workflows/reusable-drift-check.yml b/.github/workflows/reusable-drift-check.yml index 006fd57..fb8c0c1 100644 --- a/.github/workflows/reusable-drift-check.yml +++ b/.github/workflows/reusable-drift-check.yml @@ -203,6 +203,9 @@ jobs: mkdir -p "$LIFECYCLE_LOCAL_ROOT" "$GITHUB_WORKSPACE/$SNAPSHOT_CHECKOUT_PATH/data/output" cp -a "$lifecycle_source/." "$LIFECYCLE_LOCAL_ROOT/" cp -a "$snapshot_source/." "$snapshot_target/" + if [ "$STRATEGY_DOMAIN" = "crypto" ]; then + echo "CRYPTO_LIFECYCLE_PREFLIGHT_ROOT=$PREFLIGHT_BUNDLE_ROOT" >> "$GITHUB_ENV" + fi - name: Build lifecycle performance snapshots run: quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} diff --git a/schemas/strategy-review.v1.schema.json b/schemas/strategy-review.v1.schema.json new file mode 100644 index 0000000..2483551 --- /dev/null +++ b/schemas/strategy-review.v1.schema.json @@ -0,0 +1,132 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://quantstrategylab.dev/schemas/strategy-review.v1.schema.json", + "title": "Strategy Review Result v1", + "type": "object", + "required": ["schema_version", "profile", "decision", "promotion_allowed", "score", "hard_gates", "scorecard", "blocking_reason_codes", "evidence", "decision_packet"], + "properties": { + "schema_version": { "const": "strategy_review.v1" }, + "profile": { "type": "string", "minLength": 1 }, + "decision": { "enum": ["pass", "fail", "insufficient_evidence"] }, + "promotion_allowed": { "type": "boolean", "const": false }, + "score": { "type": "number", "minimum": 0, "maximum": 100 }, + "hard_gates": { + "type": "array", + "minItems": 12, + "maxItems": 12, + "items": { + "type": "object", + "required": ["id", "status", "reason_codes", "evidence_refs"], + "properties": { + "id": { "type": "string", "pattern": "^H([1-9]|1[0-2])$" }, + "name": { "type": "string" }, + "status": { "enum": ["pass", "fail", "insufficient_evidence"] }, + "reason_codes": { "type": "array", "items": { "type": "string" } }, + "evidence_refs": { "type": "array", "items": { "type": "string", "minLength": 1 } } + }, + "additionalProperties": false + }, + "allOf": [ + { "contains": { "type": "object", "properties": { "id": { "const": "H1" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H2" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H3" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H4" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H5" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H6" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H7" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H8" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H9" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H10" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H11" } }, "required": ["id"] } }, + { "contains": { "type": "object", "properties": { "id": { "const": "H12" } }, "required": ["id"] } } + ] + }, + "scorecard": { "type": "object", "required": ["total", "max", "scored_gates"], "properties": { "total": { "type": "number" }, "max": { "type": "number" }, "scored_gates": { "type": "integer" } }, "additionalProperties": false }, + "blocking_reason_codes": { "type": "array", "items": { "type": "string" } }, + "evidence": { + "type": "object", + "required": ["metrics_kind", "data_source", "sample_count", "oos_folds", "placeholder_metrics", "provenance"], + "properties": { + "metrics_kind": { "const": "performance" }, + "data_source": { "type": "string", "minLength": 1 }, + "sample_count": { "type": "integer", "minimum": 0 }, + "oos_folds": { "type": "integer", "minimum": 0 }, + "placeholder_metrics": { "const": false }, + "provenance": { + "type": "object", + "required": ["snapshot", "backtest"], + "properties": { + "snapshot": { "$ref": "#/$defs/provenance" }, + "backtest": { "$ref": "#/$defs/provenance" } + }, + "additionalProperties": false + }, + "generated_at": { "type": "string" }, + "missing_artifacts": { "type": "array", "items": { "type": "string" } } + }, + "additionalProperties": false + }, + "decision_packet": { + "type": "object", + "required": ["strategy_what", "return_source", "loss_scenarios", "max_risk", "evidence_sufficiency", "version_change", "system_recommendation", "technical_evidence_refs", "automation_boundary", "allowed_human_decisions"], + "properties": { + "strategy_what": { "type": "string", "minLength": 1 }, + "return_source": { "type": "string", "minLength": 1 }, + "loss_scenarios": { "type": "string", "minLength": 1 }, + "max_risk": { "type": "string", "minLength": 1 }, + "evidence_sufficiency": { "enum": ["sufficient", "insufficient_evidence"] }, + "version_change": { "type": "string", "minLength": 1 }, + "system_recommendation": { "enum": ["approve_research", "approve_shadow", "approve_canary", "approve_live", "reject_rollback", "insufficient_evidence"] }, + "technical_evidence_refs": { "type": "array", "items": { "type": "string", "minLength": 1 } }, + "automation_boundary": { + "type": "object", + "required": ["research_auto_after_hard_gates", "shadow_auto_after_hard_gates", "canary_mode", "canary_limits", "auto_scale_allowed", "normal_live_requires_human", "funding_leverage_risk_override_requires_human", "hard_risk_auto_pause_rollback"], + "properties": { + "research_auto_after_hard_gates": { "const": true }, + "shadow_auto_after_hard_gates": { "const": true }, + "canary_mode": { "const": "bounded_preapproved_only" }, + "canary_limits": { + "type": "object", + "required": ["max_capital", "capital_currency", "max_duration_days", "max_drawdown_fraction", "max_leverage", "max_concurrency"], + "properties": { + "max_capital": { "type": "number", "exclusiveMinimum": 0 }, + "capital_currency": { "type": "string", "minLength": 3 }, + "max_duration_days": { "type": "integer", "minimum": 1 }, + "max_drawdown_fraction": { "type": "number", "exclusiveMinimum": 0, "exclusiveMaximum": 1 }, + "max_leverage": { "type": "number", "exclusiveMinimum": 0 }, + "max_concurrency": { "type": "integer", "minimum": 1 } + }, + "additionalProperties": false + }, + "auto_scale_allowed": { "const": false }, + "normal_live_requires_human": { "const": true }, + "funding_leverage_risk_override_requires_human": { "const": true }, + "hard_risk_auto_pause_rollback": { "const": true } + }, + "additionalProperties": false + }, + "allowed_human_decisions": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { "enum": ["approve_research", "approve_shadow", "approve_canary", "approve_live", "reject_rollback"] } + } + }, + "additionalProperties": false + } + }, + "$defs": { + "provenance": { + "type": "object", + "required": ["source_revision", "cost_model", "data_timestamp", "status"], + "properties": { + "source_revision": { "type": "string", "minLength": 1 }, + "cost_model": { "type": "string", "minLength": 1 }, + "data_timestamp": { "type": "string", "minLength": 1 }, + "status": { "enum": ["verified", "legacy_missing", "unavailable"] } + }, + "additionalProperties": false + } + }, + "additionalProperties": false +} diff --git a/scripts/validate_strategy_review.py b/scripts/validate_strategy_review.py new file mode 100644 index 0000000..5e44e7a --- /dev/null +++ b/scripts/validate_strategy_review.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +"""Fail-closed validator for strategy_review.v1 results.""" + +from __future__ import annotations + +import argparse +import json +import math +from pathlib import Path +from typing import Any + +STATUSES = {"pass", "fail", "insufficient_evidence"} +DECISIONS = {"pass", "fail", "insufficient_evidence"} +SENTINELS = {"", "unavailable", "not_available", "legacy_missing", "unknown", "none", "null"} + + +def validate_review(payload: Any) -> list[str]: + issues: list[str] = [] + if not isinstance(payload, dict): + return ["top-level JSON must be an object"] + def check_keys(value: Any, allowed: set[str], label: str) -> None: + if isinstance(value, dict): + unknown = sorted(set(value) - allowed) + if unknown: + issues.append(f"{label} contains unsupported fields: {', '.join(unknown)}") + + check_keys(payload, {"schema_version", "profile", "decision", "promotion_allowed", "score", "hard_gates", "scorecard", "blocking_reason_codes", "evidence", "decision_packet"}, "review") + if payload.get("schema_version") != "strategy_review.v1": + issues.append("schema_version must be strategy_review.v1") + decision = payload.get("decision") + if not isinstance(decision, str) or decision not in DECISIONS: + issues.append("decision must be pass, fail, or insufficient_evidence") + if not isinstance(payload.get("profile"), str) or not payload["profile"].strip(): + issues.append("profile must be a non-empty string") + gates = payload.get("hard_gates") + if not isinstance(gates, list) or len(gates) != 12: + issues.append("hard_gates must contain exactly H1-H12") + gates = gates if isinstance(gates, list) else [] + ids = [gate.get("id") for gate in gates if isinstance(gate, dict)] + if set(ids) != {f"H{i}" for i in range(1, 13)}: + issues.append("hard_gates ids must be unique H1-H12") + for gate in gates: + if not isinstance(gate, dict): + issues.append("each hard gate must be an object") + continue + check_keys(gate, {"id", "name", "status", "reason_codes", "evidence_refs"}, f"{gate.get('id', '')}") + status = gate.get("status") + if not isinstance(status, str) or status not in STATUSES: + issues.append(f"{gate.get('id', '')} has invalid status") + for field in ("reason_codes", "evidence_refs"): + if not isinstance(gate.get(field), list): + issues.append(f"{gate.get('id', '')}.{field} must be an array") + elif any(not isinstance(item, str) or not item.strip() for item in gate[field]): + issues.append(f"{gate.get('id', '')}.{field} must contain strings") + + score = payload.get("score") + if not isinstance(score, (int, float)) or isinstance(score, bool) or not 0 <= score <= 100: + issues.append("score must be a number in [0, 100]") + if not isinstance(payload.get("scorecard"), dict): + issues.append("scorecard must be an object") + else: + check_keys(payload["scorecard"], {"total", "max", "scored_gates"}, "scorecard") + for field in ("total", "max"): + value = payload["scorecard"].get(field) + if not isinstance(value, (int, float)) or isinstance(value, bool) or not math.isfinite(value): + issues.append(f"scorecard.{field} must be a number") + scored_gates = payload["scorecard"].get("scored_gates") + if not isinstance(scored_gates, int) or isinstance(scored_gates, bool): + issues.append("scorecard.scored_gates must be an integer") + elif not 0 <= scored_gates <= 12: + issues.append("scorecard.scored_gates must be in [0, 12]") + total = payload["scorecard"].get("total") + maximum = payload["scorecard"].get("max") + if isinstance(total, (int, float)) and isinstance(maximum, (int, float)) and not isinstance(total, bool) and not isinstance(maximum, bool) and (not math.isfinite(total) or not math.isfinite(maximum) or not 0 <= total <= maximum): + issues.append("scorecard.total must be between 0 and scorecard.max") + evidence = payload.get("evidence") + provenance: dict[str, Any] = {} + missing_artifacts: list[Any] | None = None + if not isinstance(evidence, dict): + issues.append("evidence must be an object") + else: + check_keys(evidence, {"metrics_kind", "data_source", "sample_count", "oos_folds", "placeholder_metrics", "provenance", "generated_at", "missing_artifacts"}, "evidence") + if evidence.get("metrics_kind") != "performance": + issues.append("evidence.metrics_kind must be performance") + for field in ("data_source",): + if not isinstance(evidence.get(field), str) or not evidence[field].strip(): + issues.append(f"evidence.{field} must be a non-empty string") + provenance = evidence.get("provenance") + if not isinstance(provenance, dict): + issues.append("evidence.provenance must be an object") + provenance = {} + else: + check_keys(provenance, {"snapshot", "backtest"}, "evidence.provenance") + for source in ("snapshot", "backtest"): + item = provenance.get(source) + if not isinstance(item, dict): + issues.append(f"evidence.provenance.{source} must be an object") + provenance[source] = {} + continue + check_keys(item, {"source_revision", "cost_model", "data_timestamp", "status"}, f"evidence.provenance.{source}") + for field in ("source_revision", "cost_model", "data_timestamp"): + if not isinstance(item.get(field), str) or not item[field].strip(): + issues.append(f"evidence.provenance.{source}.{field} must be non-empty") + item_status = item.get("status") + if not isinstance(item_status, str) or item_status not in {"verified", "legacy_missing", "unavailable"}: + issues.append(f"evidence.provenance.{source}.status is invalid") + if item.get("status") == "verified": + if str(item.get("source_revision", "")).strip().lower() in SENTINELS or str(item.get("cost_model", "")).strip().lower() in SENTINELS: + issues.append(f"evidence.provenance.{source}.verified cannot use sentinel provenance") + try: + from datetime import datetime + timestamp = datetime.fromisoformat(item["data_timestamp"].replace("Z", "+00:00")) + if timestamp.tzinfo is None: + raise ValueError + except (AttributeError, TypeError, ValueError): + issues.append(f"evidence.provenance.{source}.verified requires a real timestamp") + if evidence.get("placeholder_metrics") is not False: + issues.append("placeholder metrics are not admissible evidence") + missing_artifacts = evidence.get("missing_artifacts", []) + if not isinstance(missing_artifacts, list) or any(not isinstance(item, str) for item in missing_artifacts): + issues.append("evidence.missing_artifacts must be a string array") + if not isinstance(evidence.get("sample_count"), int) or isinstance(evidence.get("sample_count"), bool) or evidence["sample_count"] < 0: + issues.append("evidence.sample_count must be a non-negative integer") + if not isinstance(evidence.get("oos_folds"), int) or isinstance(evidence.get("oos_folds"), bool) or evidence["oos_folds"] < 0: + issues.append("evidence.oos_folds must be a non-negative integer") + + packet = payload.get("decision_packet") + packet_fields = ("strategy_what", "return_source", "loss_scenarios", "max_risk", "version_change") + if not isinstance(packet, dict): + issues.append("decision_packet must be an object") + else: + check_keys(packet, {"strategy_what", "return_source", "loss_scenarios", "max_risk", "evidence_sufficiency", "version_change", "system_recommendation", "technical_evidence_refs", "automation_boundary", "allowed_human_decisions"}, "decision_packet") + for field in packet_fields: + if not isinstance(packet.get(field), str) or not packet[field].strip(): + issues.append(f"decision_packet.{field} must be a non-empty string") + evidence_sufficiency = packet.get("evidence_sufficiency") + if not isinstance(evidence_sufficiency, str) or evidence_sufficiency not in {"sufficient", "insufficient_evidence"}: + issues.append("decision_packet.evidence_sufficiency is invalid") + recommendations = {"approve_research", "approve_shadow", "approve_canary", "approve_live", "reject_rollback", "insufficient_evidence"} + recommendation = packet.get("system_recommendation") + if not isinstance(recommendation, str) or recommendation not in recommendations: + issues.append("decision_packet.system_recommendation is invalid") + if not isinstance(packet.get("technical_evidence_refs"), list) or any(not isinstance(item, str) or not item.strip() for item in packet.get("technical_evidence_refs", [])): + issues.append("decision_packet.technical_evidence_refs must be a string array") + boundary = packet.get("automation_boundary") + if not isinstance(boundary, dict): + issues.append("decision_packet.automation_boundary must be an object") + else: + check_keys(boundary, {"research_auto_after_hard_gates", "shadow_auto_after_hard_gates", "canary_mode", "canary_limits", "auto_scale_allowed", "normal_live_requires_human", "funding_leverage_risk_override_requires_human", "hard_risk_auto_pause_rollback"}, "automation_boundary") + required_boundary = { + "research_auto_after_hard_gates": True, + "shadow_auto_after_hard_gates": True, + "canary_mode": "bounded_preapproved_only", + "auto_scale_allowed": False, + "normal_live_requires_human": True, + "funding_leverage_risk_override_requires_human": True, + "hard_risk_auto_pause_rollback": True, + } + for field, expected in required_boundary.items(): + if boundary.get(field) != expected: + issues.append(f"decision_packet.automation_boundary.{field} is unsafe") + if not isinstance(boundary.get("canary_limits"), dict): + issues.append("decision_packet.automation_boundary.canary_limits must be an object") + else: + limits = boundary["canary_limits"] + checks = { + "max_capital": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v) and v > 0, + "capital_currency": lambda v: isinstance(v, str) and len(v.strip()) >= 3, + "max_duration_days": lambda v: isinstance(v, int) and not isinstance(v, bool) and v >= 1, + "max_drawdown_fraction": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v) and 0 < v < 1, + "max_leverage": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v) and v > 0, + "max_concurrency": lambda v: isinstance(v, int) and not isinstance(v, bool) and v >= 1, + } + if set(limits) != set(checks) or any(not check(limits.get(key)) for key, check in checks.items()): + issues.append("decision_packet.automation_boundary.canary_limits is incomplete or invalid") + allowed = packet.get("allowed_human_decisions") + if not isinstance(allowed, list) or not allowed or any(not isinstance(item, str) or item not in {"approve_research", "approve_shadow", "approve_canary", "approve_live", "reject_rollback"} for item in allowed): + issues.append("decision_packet.allowed_human_decisions is invalid") + elif len(allowed) != len(set(allowed)): + issues.append("decision_packet.allowed_human_decisions must be unique") + expected_allowed = { + "approve_research": {"approve_research", "reject_rollback"}, + "approve_shadow": {"approve_shadow", "reject_rollback"}, + "approve_canary": {"approve_canary", "reject_rollback"}, + "approve_live": {"approve_live", "reject_rollback"}, + "reject_rollback": {"reject_rollback"}, + "insufficient_evidence": {"approve_research", "reject_rollback"}, + }.get(recommendation) + if expected_allowed is not None and set(allowed or []) != expected_allowed: + issues.append("decision_packet.allowed_human_decisions does not match system recommendation") + if packet.get("evidence_sufficiency") == "insufficient_evidence" and packet.get("system_recommendation") != "insufficient_evidence": + issues.append("insufficient evidence requires an insufficient_evidence recommendation") + if payload.get("decision") == "insufficient_evidence" and ( + packet.get("evidence_sufficiency") != "insufficient_evidence" + or packet.get("system_recommendation") != "insufficient_evidence" + ): + issues.append("decision=insufficient_evidence requires matching packet state") + if payload.get("decision") == "pass" and packet.get("system_recommendation") not in { + "approve_research", "approve_shadow", "approve_canary", "approve_live" + }: + issues.append("decision=pass requires an approval recommendation") + if payload.get("decision") == "fail" and packet.get("system_recommendation") != "reject_rollback": + issues.append("decision=fail requires a reject_rollback recommendation") + if packet.get("evidence_sufficiency") == "insufficient_evidence" and any(item != "approve_research" and item != "reject_rollback" for item in allowed or []): + issues.append("insufficient evidence cannot allow shadow, canary, or live approval") + promotive = {"approve_shadow", "approve_canary", "approve_live"} + approval_recommendation = packet.get("system_recommendation") in {"approve_research", *promotive} + if approval_recommendation and payload.get("decision") != "pass": + issues.append("approval recommendation requires decision=pass") + if payload.get("decision") != "pass" and any(item in promotive for item in allowed or []): + issues.append("failed or insufficient review cannot allow shadow, canary, or live approval") + if payload.get("decision") == "pass" and any(isinstance(gate, dict) and gate.get("status") != "pass" for gate in gates): + issues.append("decision=pass requires every hard gate to pass") + if payload.get("decision") == "pass" and packet.get("system_recommendation") == "insufficient_evidence": + issues.append("decision=pass cannot recommend insufficient_evidence") + if packet.get("system_recommendation") in {"approve_research", "approve_shadow", "approve_canary", "approve_live"} and any( + isinstance(gate, dict) and gate.get("status") != "pass" for gate in gates + ): + issues.append("approval recommendation requires every hard gate to pass") + promotive = packet.get("system_recommendation") in {"approve_shadow", "approve_canary", "approve_live"} + if payload.get("decision") == "pass" or promotive: + if not isinstance(evidence, dict) or evidence.get("sample_count", 0) <= 0 or evidence.get("oos_folds", 0) < 3: + issues.append("passing review requires positive sample_count and at least 3 oos_folds") + if not isinstance(packet.get("technical_evidence_refs"), list) or not packet["technical_evidence_refs"]: + issues.append("passing review requires technical evidence references") + if isinstance(missing_artifacts, list) and missing_artifacts: + issues.append("passing review cannot declare missing artifacts") + for gate in gates: + if isinstance(gate, dict) and gate.get("status") == "pass" and not gate.get("evidence_refs"): + issues.append(f"{gate.get('id', '')} pass requires evidence_refs") + if any(provenance.get(source, {}).get("status") != "verified" for source in ("snapshot", "backtest")): + issues.append("passing review requires verified snapshot and backtest provenance") + + blocking = payload.get("blocking_reason_codes") + if not isinstance(blocking, list): + issues.append("blocking_reason_codes must be an array") + blocking = [] + elif any(not isinstance(item, str) for item in blocking): + issues.append("blocking_reason_codes must contain strings") + failed = [gate for gate in gates if isinstance(gate, dict) and gate.get("status") in {"fail", "insufficient_evidence"}] + if failed and not blocking: + issues.append("failed or insufficient hard gates require blocking_reason_codes") + if payload.get("promotion_allowed") is not False: + issues.append("promotion_allowed must be false in v1 review output") + if isinstance(evidence, dict) and isinstance(packet, dict) and ( + payload.get("decision") == "pass" + or packet.get("system_recommendation") in {"approve_research", "approve_shadow", "approve_canary", "approve_live"} + ) and str(evidence.get("data_source", "")).strip().lower() in SENTINELS: + issues.append("approved review cannot use a provenance sentinel as evidence.data_source") + if payload.get("decision") == "pass" and failed: + issues.append("decision cannot be pass when a hard gate failed or lacks evidence") + if payload.get("decision") == "pass" and isinstance(payload.get("score"), (int, float)) and isinstance(payload.get("scorecard"), dict): + total = payload["scorecard"].get("total") + maximum = payload["scorecard"].get("max") + scored_gates = payload["scorecard"].get("scored_gates") + if not isinstance(maximum, (int, float)) or maximum <= 0 or not isinstance(scored_gates, int) or scored_gates != 12: + issues.append("passing review requires a positive scorecard.max and all hard gates scored") + elif isinstance(total, (int, float)) and abs(payload["score"] - (100 * total / maximum)) > 1e-9: + issues.append("passing review score must match scorecard.total/max") + if payload.get("decision") == "pass" and blocking: + issues.append("decision cannot be pass with blocking_reason_codes") + return issues + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser() + parser.add_argument("review", type=Path) + args = parser.parse_args(argv) + issues = validate_review(json.loads(args.review.read_text(encoding="utf-8"), parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON constant: {value}")))) + if issues: + for issue in issues: + print(f"ERROR: {issue}") + return 1 + print("strategy_review.v1: valid") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index 89e81a5..dfb7921 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -100,6 +100,8 @@ class StrategyPerformanceSnapshot: data_freshness_days: int = 0 source_artifact_path: str = "" computed_at: str = "" + source_revision: str = "" + cost_model: str = "" def to_dict(self) -> dict[str, object]: return { @@ -115,6 +117,8 @@ def to_dict(self) -> dict[str, object]: "data_freshness_days": self.data_freshness_days, "source_artifact_path": self.source_artifact_path, "computed_at": self.computed_at, + "source_revision": self.source_revision, + "cost_model": self.cost_model, } @@ -256,6 +260,8 @@ class BacktestResult: run_duration_seconds: float = 0.0 source_script: str = "" computed_at: str = "" + source_revision: str = "" + cost_model: str = "" def to_dict(self) -> dict[str, object]: return { @@ -287,6 +293,8 @@ def to_dict(self) -> dict[str, object]: "run_duration_seconds": self.run_duration_seconds, "source_script": self.source_script, "computed_at": self.computed_at, + "source_revision": self.source_revision, + "cost_model": self.cost_model, } diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_export.py b/src/quant_platform_kit/strategy_lifecycle/performance_export.py index 5672a4e..c9f9564 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_export.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_export.py @@ -15,6 +15,7 @@ METRICS_KIND = "performance" DEFAULT_WINDOWS: tuple[int, ...] = (126, 252, 63, 21) REQUIRED_METRICS = ("sharpe", "cagr", "calmar", "win_rate", "max_dd") +PROVENANCE_SENTINELS = {"", "unavailable", "not_available", "legacy_missing", "unknown", "none", "null"} def _now_iso() -> str: @@ -86,6 +87,16 @@ def _assert_required_metrics(metrics: Mapping[str, Any], *, label: str) -> None: raise ValueError(f"{label} missing required metrics: {', '.join(missing)}") +def _date_timestamp(value: date | None) -> str: + if value is None: + return "unavailable" + return datetime(value.year, value.month, value.day, tzinfo=timezone.utc).isoformat().replace("+00:00", "Z") + + +def _usable_provenance(value: str) -> bool: + return isinstance(value, str) and value.strip().lower() not in PROVENANCE_SENTINELS + + def export_strategy_performance( domain: str, *, @@ -113,6 +124,7 @@ def export_strategy_performance( raise ValueError(f"Missing latest lifecycle backtest for domain={domain!r}, profile={profile!r}") window = _window_for_snapshot(snapshot, preferred_windows) + snapshot_timestamp = _date_timestamp(snapshot.as_of) snapshots.append( { "repo": repo, @@ -131,6 +143,26 @@ def export_strategy_performance( "window_end": window.end_date.isoformat(), "snapshot_computed_at": snapshot.computed_at, "backtest_computed_at": backtest.computed_at, + "snapshot_source_revision": snapshot.source_revision, + "backtest_source_revision": backtest.source_revision, + "snapshot_cost_model": snapshot.cost_model, + "backtest_cost_model": backtest.cost_model, + "provenance": { + "snapshot": { + "source_revision": snapshot.source_revision or "legacy_missing", + "cost_model": snapshot.cost_model or "legacy_missing", + "data_timestamp": snapshot_timestamp, + "status": "verified" if snapshot.as_of and _usable_provenance(snapshot.source_revision) and _usable_provenance(snapshot.cost_model) else "legacy_missing", + }, + "backtest": { + "source_revision": backtest.source_revision or "legacy_missing", + "cost_model": backtest.cost_model or "legacy_missing", + "data_timestamp": _date_timestamp(backtest.end_date), + "status": "verified" if _usable_provenance(backtest.source_revision) and _usable_provenance(backtest.cost_model) and backtest.end_date else "legacy_missing", + }, + }, + "snapshot_data_timestamp": snapshot_timestamp, + "backtest_data_timestamp": _date_timestamp(backtest.end_date), }, } ) diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index b009811..4fbe735 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -505,6 +505,8 @@ def _snapshot_from_dict(data: Mapping[str, Any]) -> StrategyPerformanceSnapshot data_freshness_days=int(data.get("data_freshness_days", 0)), source_artifact_path=str(data.get("source_artifact_path", "")), computed_at=str(data.get("computed_at", "")), + source_revision=data.get("source_revision") if isinstance(data.get("source_revision"), str) else "", + cost_model=data.get("cost_model") if isinstance(data.get("cost_model"), str) else "", ) except Exception: return None @@ -588,6 +590,8 @@ def _backtest_from_dict(data: Mapping[str, Any]) -> BacktestResult | None: run_duration_seconds=float(data.get("run_duration_seconds", 0.0) or 0.0), source_script=str(data.get("source_script", "")), computed_at=str(data.get("computed_at", "")), + source_revision=data.get("source_revision") if isinstance(data.get("source_revision"), str) else "", + cost_model=data.get("cost_model") if isinstance(data.get("cost_model"), str) else "", ) except Exception: return None diff --git a/tests/test_strategy_performance_export.py b/tests/test_strategy_performance_export.py index 8a0b556..25b1aa0 100644 --- a/tests/test_strategy_performance_export.py +++ b/tests/test_strategy_performance_export.py @@ -52,6 +52,8 @@ def test_export_strategy_performance_writes_canonical_payload(self) -> None: data_freshness_days=1, source_artifact_path="data/output/live_returns.csv", computed_at="2026-06-30T00:00:00Z", + source_revision="fixture-rev", + cost_model="fee_10bps_slippage_5bps", ) backtest = BacktestResult( strategy_profile="crypto_live_pool_rotation", @@ -76,6 +78,9 @@ def test_export_strategy_performance_writes_canonical_payload(self) -> None: oos_max_drawdown=-0.11, walk_forward_stability=0.93, computed_at="2026-06-29T00:00:00Z", + end_date=date(2026, 6, 30), + source_revision="fixture-rev", + cost_model="fee_10bps_slippage_5bps", ) store.save_snapshot(snapshot) store.save_backtest_result(backtest) @@ -113,6 +118,23 @@ def test_export_strategy_performance_fails_without_backtest(self) -> None: with self.assertRaisesRegex(ValueError, "Missing latest lifecycle backtest"): export_strategy_performance("crypto", repo="QuantStrategyLab/CryptoLivePoolPipelines", store=store) + def test_export_preserves_conflicting_provenance_separately(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_snapshot(StrategyPerformanceSnapshot( + strategy_profile="crypto_live_pool_rotation", domain="crypto", platform="binance", + as_of=date(2026, 6, 30), windows={126: _window()}, source_revision="snapshot-rev", + )) + store.save_backtest_result(BacktestResult( + strategy_profile="crypto_live_pool_rotation", domain="crypto", param_set_id="baseline", + params={}, sharpe_ratio=1.0, calmar_ratio=1.0, max_drawdown=-0.1, cagr=0.2, + win_rate=0.55, end_date=date(2026, 6, 30), source_revision="backtest-rev", + )) + payload = export_strategy_performance("crypto", repo="QuantStrategyLab/CryptoLivePoolPipelines", store=store) + metadata = payload["snapshots"][0]["metadata"] + self.assertEqual(metadata["snapshot_source_revision"], "snapshot-rev") + self.assertEqual(metadata["backtest_source_revision"], "backtest-rev") + def test_performance_store_load_latest_backtest_falls_back_to_local(self) -> None: with tempfile.TemporaryDirectory() as tmp: store = PerformanceStore(local_root=Path(tmp)) diff --git a/tests/test_strategy_review_validator.py b/tests/test_strategy_review_validator.py new file mode 100644 index 0000000..80a17b1 --- /dev/null +++ b/tests/test_strategy_review_validator.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import importlib.util +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location("review_validator", ROOT / "scripts/validate_strategy_review.py") +MODULE = importlib.util.module_from_spec(SPEC) +assert SPEC and SPEC.loader +SPEC.loader.exec_module(MODULE) + + +def _review(**overrides: object) -> dict[str, object]: + payload: dict[str, object] = { + "schema_version": "strategy_review.v1", + "profile": "crypto_live_pool_rotation", + "decision": "insufficient_evidence", + "promotion_allowed": False, + "score": 0, + "hard_gates": [ + {"id": f"H{i}", "status": "insufficient_evidence", "reason_codes": ["INSUFFICIENT_DATA"], "evidence_refs": []} + for i in range(1, 13) + ], + "scorecard": {"total": 0, "max": 100, "scored_gates": 0}, + "blocking_reason_codes": ["INSUFFICIENT_DATA"], + "evidence": { + "metrics_kind": "performance", + "data_source": "not_available", + "sample_count": 0, + "oos_folds": 0, + "placeholder_metrics": False, + "provenance": { + "snapshot": {"source_revision": "fixture-rev", "cost_model": "not_available", "data_timestamp": "2026-07-11T00:00:00Z", "status": "unavailable"}, + "backtest": {"source_revision": "fixture-rev", "cost_model": "not_available", "data_timestamp": "2026-07-11T00:00:00Z", "status": "unavailable"}, + }, + }, + "decision_packet": { + "strategy_what": "策略做什么尚待真实证据确认", + "return_source": "真实 performance artifacts 未提供", + "loss_scenarios": "未完成回测，主要亏损场景不可确认", + "max_risk": "最大风险不可确认", + "evidence_sufficiency": "insufficient_evidence", + "version_change": "仅生成 fail-closed 评审结果", + "system_recommendation": "insufficient_evidence", + "technical_evidence_refs": [], + "automation_boundary": { + "research_auto_after_hard_gates": True, + "shadow_auto_after_hard_gates": True, + "canary_mode": "bounded_preapproved_only", + "canary_limits": {"max_capital": 1000.0, "capital_currency": "USD", "max_duration_days": 14, "max_drawdown_fraction": 0.05, "max_leverage": 1.0, "max_concurrency": 1}, + "auto_scale_allowed": False, + "normal_live_requires_human": True, + "funding_leverage_risk_override_requires_human": True, + "hard_risk_auto_pause_rollback": True, + }, + "allowed_human_decisions": ["approve_research", "reject_rollback"], + }, + } + payload.update(overrides) + return payload + + +class StrategyReviewValidatorTests(unittest.TestCase): + def test_insufficient_evidence_is_valid_and_fail_closed(self) -> None: + self.assertEqual(MODULE.validate_review(_review()), []) + + def test_operational_metrics_are_rejected(self) -> None: + payload = _review(evidence={**_review()["evidence"], "metrics_kind": "operational"}) + self.assertTrue(any("metrics_kind" in issue for issue in MODULE.validate_review(payload))) + + def test_placeholder_metrics_are_rejected(self) -> None: + payload = _review(evidence={**_review()["evidence"], "placeholder_metrics": True}) + self.assertTrue(any("placeholder" in issue for issue in MODULE.validate_review(payload))) + + def test_pass_cannot_override_failed_gate(self) -> None: + payload = _review(decision="pass") + self.assertTrue(any("cannot be pass" in issue for issue in MODULE.validate_review(payload))) + + def test_unknown_decision_is_rejected(self) -> None: + self.assertTrue(any("decision must be" in issue for issue in MODULE.validate_review(_review(decision="unknown")))) + + def test_empty_provenance_is_rejected(self) -> None: + evidence = {**_review()["evidence"], "data_source": ""} + evidence["provenance"] = {**evidence["provenance"], "snapshot": {**evidence["provenance"]["snapshot"], "cost_model": " "}} + payload = _review(evidence=evidence) + issues = MODULE.validate_review(payload) + self.assertTrue(any("data_source" in issue for issue in issues)) + self.assertTrue(any("cost_model" in issue for issue in issues)) + + def test_missing_scorecard_and_boolean_counts_are_rejected(self) -> None: + payload = _review() + payload.pop("scorecard") + payload["evidence"] = {**_review()["evidence"], "sample_count": True, "oos_folds": False} + issues = MODULE.validate_review(payload) + self.assertTrue(any("scorecard" in issue for issue in issues)) + self.assertTrue(any("sample_count" in issue for issue in issues)) + self.assertTrue(any("oos_folds" in issue for issue in issues)) + + def test_decision_and_recommendation_must_be_consistent(self) -> None: + payload = _review() + payload["decision"] = "pass" + payload["decision_packet"] = {**payload["decision_packet"], "evidence_sufficiency": "sufficient", "system_recommendation": "reject_rollback", "allowed_human_decisions": ["reject_rollback"]} + self.assertTrue(any("approval recommendation" in issue for issue in MODULE.validate_review(payload))) + + def test_reason_code_arrays_must_contain_strings(self) -> None: + payload = _review() + payload["hard_gates"][0]["reason_codes"] = [123] # type: ignore[index] + payload["blocking_reason_codes"] = [False] + issues = MODULE.validate_review(payload) + self.assertTrue(any("must contain strings" in issue for issue in issues)) + + def test_approval_recommendation_requires_all_gates(self) -> None: + payload = _review() + payload["decision_packet"]["system_recommendation"] = "approve_research" # type: ignore[index] + self.assertTrue(any("every hard gate" in issue for issue in MODULE.validate_review(payload))) + + def test_pass_requires_real_samples_folds_and_evidence_refs(self) -> None: + payload = _review(decision="pass") + payload["hard_gates"] = [{**gate, "status": "pass"} for gate in payload["hard_gates"]] + payload["decision_packet"]["system_recommendation"] = "approve_research" # type: ignore[index] + issues = MODULE.validate_review(payload) + self.assertTrue(any("positive sample_count" in issue for issue in issues)) + + def test_duplicate_human_actions_are_rejected(self) -> None: + payload = _review() + payload["decision_packet"]["allowed_human_decisions"] = ["approve_research", "approve_research"] # type: ignore[index] + self.assertTrue(any("must be unique" in issue for issue in MODULE.validate_review(payload))) + + def test_verified_provenance_rejects_sentinel(self) -> None: + payload = _review() + payload["decision_packet"]["evidence_sufficiency"] = "sufficient" # type: ignore[index] + payload["decision"] = "pass" + payload["evidence"]["sample_count"] = 10 # type: ignore[index] + payload["evidence"]["oos_folds"] = 3 # type: ignore[index] + for gate in payload["hard_gates"]: + gate["status"] = "pass" + gate["evidence_refs"] = ["artifact"] + payload["decision_packet"]["system_recommendation"] = "approve_research" # type: ignore[index] + payload["decision_packet"]["technical_evidence_refs"] = ["artifact"] # type: ignore[index] + payload["evidence"]["provenance"]["backtest"]["status"] = "verified" # type: ignore[index] + payload["evidence"]["provenance"]["backtest"]["source_revision"] = "legacy_missing" # type: ignore[index] + self.assertTrue(any("verified" in issue for issue in MODULE.validate_review(payload))) + + def test_insufficient_decision_requires_matching_packet(self) -> None: + payload = _review(decision="insufficient_evidence") + payload["decision_packet"]["evidence_sufficiency"] = "sufficient" # type: ignore[index] + payload["decision_packet"]["system_recommendation"] = "reject_rollback" # type: ignore[index] + self.assertTrue(any("matching packet state" in issue for issue in MODULE.validate_review(payload))) + + def test_canary_limits_reject_non_finite_numbers(self) -> None: + payload = _review() + payload["decision_packet"]["automation_boundary"]["canary_limits"]["max_capital"] = float("inf") # type: ignore[index] + self.assertTrue(any("canary_limits" in issue for issue in MODULE.validate_review(payload))) + + +if __name__ == "__main__": + unittest.main()