|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Build one comparable watcher payload from two sanitized run artifacts.""" |
| 3 | + |
| 4 | +from __future__ import annotations |
| 5 | + |
| 6 | +import argparse |
| 7 | +import json |
| 8 | +import math |
| 9 | +import re |
| 10 | +from datetime import datetime |
| 11 | +from pathlib import Path |
| 12 | +from typing import Any, Mapping |
| 13 | + |
| 14 | + |
| 15 | +SCHEMA_VERSION = "strategy_performance.v2" |
| 16 | +METRICS_KIND = "performance" |
| 17 | +_REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") |
| 18 | +_SHA256 = re.compile(r"^[0-9a-f]{64}$") |
| 19 | +_REVISION = re.compile(r"^[0-9a-f]{40}$") |
| 20 | +_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$") |
| 21 | +_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") |
| 22 | +_METRICS = frozenset({"sharpe", "cagr", "calmar", "win_rate", "max_dd"}) |
| 23 | +_FORBIDDEN_KEYS = re.compile(r"(?:secret|token|password|credential|api[_-]?key|order|fill|capital|account|broker|path)", re.IGNORECASE) |
| 24 | +_SAFE_RESEARCH_KEYS = frozenset({"no_order"}) |
| 25 | + |
| 26 | + |
| 27 | +class StrategyWatcherArtifactError(ValueError): |
| 28 | + """Raised when two artifacts cannot form a safe metric comparison.""" |
| 29 | + |
| 30 | + |
| 31 | +def _exact_mapping(value: object, fields: frozenset[str], label: str) -> dict[str, Any]: |
| 32 | + if not isinstance(value, Mapping) or set(value) != fields: |
| 33 | + raise StrategyWatcherArtifactError(f"invalid {label}") |
| 34 | + return dict(value) |
| 35 | + |
| 36 | + |
| 37 | +def _timestamp(value: object, label: str) -> str: |
| 38 | + if not isinstance(value, str) or not _TIMESTAMP.fullmatch(value): |
| 39 | + raise StrategyWatcherArtifactError(f"invalid {label}") |
| 40 | + try: |
| 41 | + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) |
| 42 | + except ValueError as exc: |
| 43 | + raise StrategyWatcherArtifactError(f"invalid {label}") from exc |
| 44 | + if parsed.tzinfo is None or parsed.utcoffset() is None: |
| 45 | + raise StrategyWatcherArtifactError(f"invalid {label}") |
| 46 | + return value |
| 47 | + |
| 48 | + |
| 49 | +def _finite_metrics(value: object, label: str) -> dict[str, float]: |
| 50 | + metrics = _exact_mapping(value, _METRICS, label) |
| 51 | + result: dict[str, float] = {} |
| 52 | + for key, raw in metrics.items(): |
| 53 | + if isinstance(raw, bool) or not isinstance(raw, (int, float)) or not math.isfinite(float(raw)): |
| 54 | + raise StrategyWatcherArtifactError(f"invalid {label}") |
| 55 | + result[key] = float(raw) |
| 56 | + return result |
| 57 | + |
| 58 | + |
| 59 | +def _sha256(value: object, label: str) -> str: |
| 60 | + if not isinstance(value, str) or not _SHA256.fullmatch(value): |
| 61 | + raise StrategyWatcherArtifactError(f"invalid {label}") |
| 62 | + return value |
| 63 | + |
| 64 | + |
| 65 | +def _revision(value: object, label: str) -> str: |
| 66 | + if not isinstance(value, str) or not _REVISION.fullmatch(value): |
| 67 | + raise StrategyWatcherArtifactError(f"invalid {label}") |
| 68 | + return value |
| 69 | + |
| 70 | + |
| 71 | +def _forbid_unsafe_keys(value: object) -> None: |
| 72 | + if isinstance(value, Mapping): |
| 73 | + for key, nested in value.items(): |
| 74 | + if str(key) not in _SAFE_RESEARCH_KEYS and _FORBIDDEN_KEYS.search(str(key)): |
| 75 | + raise StrategyWatcherArtifactError("unsafe artifact field") |
| 76 | + _forbid_unsafe_keys(nested) |
| 77 | + elif isinstance(value, list): |
| 78 | + for nested in value: |
| 79 | + _forbid_unsafe_keys(nested) |
| 80 | + elif isinstance(value, float) and not math.isfinite(value): |
| 81 | + raise StrategyWatcherArtifactError("non-finite artifact value") |
| 82 | + |
| 83 | + |
| 84 | +def _performance_artifact(value: object, *, expected_repository: str) -> dict[str, Any]: |
| 85 | + artifact = _exact_mapping( |
| 86 | + value, |
| 87 | + frozenset( |
| 88 | + { |
| 89 | + "schema_version", |
| 90 | + "metrics_kind", |
| 91 | + "repository", |
| 92 | + "strategy_profile", |
| 93 | + "candidate_kind", |
| 94 | + "domain", |
| 95 | + "generated_at", |
| 96 | + "as_of", |
| 97 | + "current_metrics", |
| 98 | + "evidence", |
| 99 | + "lifecycle", |
| 100 | + "authority", |
| 101 | + } |
| 102 | + ), |
| 103 | + "strategy performance artifact", |
| 104 | + ) |
| 105 | + _forbid_unsafe_keys(artifact) |
| 106 | + if artifact["schema_version"] != SCHEMA_VERSION or artifact["metrics_kind"] != METRICS_KIND: |
| 107 | + raise StrategyWatcherArtifactError("unsupported strategy performance artifact") |
| 108 | + if artifact["repository"] != expected_repository: |
| 109 | + raise StrategyWatcherArtifactError("artifact repository does not match validated source repository") |
| 110 | + if not isinstance(artifact["strategy_profile"], str) or not re.fullmatch(r"[A-Za-z0-9._=-]{1,120}", artifact["strategy_profile"]): |
| 111 | + raise StrategyWatcherArtifactError("invalid strategy profile") |
| 112 | + if artifact["candidate_kind"] not in {"individual", "portfolio", "plugin"}: |
| 113 | + raise StrategyWatcherArtifactError("invalid candidate kind") |
| 114 | + if artifact["domain"] not in {"us_equity", "hk_equity", "cn_equity", "crypto"}: |
| 115 | + raise StrategyWatcherArtifactError("invalid domain") |
| 116 | + _timestamp(artifact["generated_at"], "generated_at") |
| 117 | + if not isinstance(artifact["as_of"], str) or not _DATE.fullmatch(artifact["as_of"]): |
| 118 | + raise StrategyWatcherArtifactError("invalid as_of") |
| 119 | + try: |
| 120 | + datetime.fromisoformat(artifact["as_of"]) |
| 121 | + except ValueError as exc: |
| 122 | + raise StrategyWatcherArtifactError("invalid as_of") from exc |
| 123 | + artifact["current_metrics"] = _finite_metrics(artifact["current_metrics"], "current_metrics") |
| 124 | + evidence = _exact_mapping( |
| 125 | + artifact["evidence"], |
| 126 | + frozenset({"p1_input_digest", "p2_config_digest", "p3_evidence_id", "strategy_revision", "producer_revision"}), |
| 127 | + "evidence", |
| 128 | + ) |
| 129 | + for key in ("p1_input_digest", "p2_config_digest", "p3_evidence_id"): |
| 130 | + _sha256(evidence[key], f"evidence.{key}") |
| 131 | + _revision(evidence["strategy_revision"], "evidence.strategy_revision") |
| 132 | + _revision(evidence["producer_revision"], "evidence.producer_revision") |
| 133 | + if artifact["lifecycle"] != {"stage": "P3", "status": "verified"}: |
| 134 | + raise StrategyWatcherArtifactError("artifact is not verified P3 evidence") |
| 135 | + if artifact["authority"] != {"research_only": True, "no_order": True, "p4_p5_p6_authorized": False}: |
| 136 | + raise StrategyWatcherArtifactError("artifact authority is not research-only") |
| 137 | + return artifact |
| 138 | + |
| 139 | + |
| 140 | +def build_strategy_watcher_artifact_payload( |
| 141 | + *, |
| 142 | + current_artifact: object, |
| 143 | + baseline_artifact: object, |
| 144 | + source_repository: object, |
| 145 | + workflow_file: object, |
| 146 | + current_run_id: object, |
| 147 | + baseline_run_id: object, |
| 148 | +) -> dict[str, object]: |
| 149 | + """Join two completed P3 performance observations for an issue-only watcher.""" |
| 150 | + if not isinstance(source_repository, str) or not _REPOSITORY.fullmatch(source_repository): |
| 151 | + raise StrategyWatcherArtifactError("invalid source repository") |
| 152 | + if not isinstance(workflow_file, str) or not re.fullmatch(r"[A-Za-z0-9_.-]+\.ya?ml", workflow_file): |
| 153 | + raise StrategyWatcherArtifactError("invalid workflow file") |
| 154 | + if not isinstance(current_run_id, str) or not current_run_id.isdigit() or not isinstance(baseline_run_id, str) or not baseline_run_id.isdigit(): |
| 155 | + raise StrategyWatcherArtifactError("invalid workflow run id") |
| 156 | + current = _performance_artifact(current_artifact, expected_repository=source_repository) |
| 157 | + baseline = _performance_artifact(baseline_artifact, expected_repository=source_repository) |
| 158 | + for key in ("strategy_profile", "candidate_kind", "domain"): |
| 159 | + if current[key] != baseline[key]: |
| 160 | + raise StrategyWatcherArtifactError("artifacts describe different research candidates") |
| 161 | + if _timestamp(baseline["generated_at"], "baseline generated_at") >= _timestamp(current["generated_at"], "current generated_at"): |
| 162 | + raise StrategyWatcherArtifactError("baseline must precede current observation") |
| 163 | + if baseline["as_of"] >= current["as_of"]: |
| 164 | + raise StrategyWatcherArtifactError("baseline must use an earlier data cutoff") |
| 165 | + return { |
| 166 | + "schema_version": SCHEMA_VERSION, |
| 167 | + "metrics_kind": METRICS_KIND, |
| 168 | + "repo": source_repository, |
| 169 | + "strategy_profile": current["strategy_profile"], |
| 170 | + "candidate_kind": current["candidate_kind"], |
| 171 | + "domain": current["domain"], |
| 172 | + "generated_at": current["generated_at"], |
| 173 | + "current_metrics": current["current_metrics"], |
| 174 | + "baseline_metrics": baseline["current_metrics"], |
| 175 | + "source": f"github_actions:{source_repository}:{workflow_file}:{baseline_run_id}-{current_run_id}", |
| 176 | + } |
| 177 | + |
| 178 | + |
| 179 | +def _arguments() -> argparse.Namespace: |
| 180 | + parser = argparse.ArgumentParser(description=__doc__) |
| 181 | + parser.add_argument("--current", required=True, type=Path) |
| 182 | + parser.add_argument("--baseline", required=True, type=Path) |
| 183 | + parser.add_argument("--source-repository", required=True) |
| 184 | + parser.add_argument("--workflow-file", required=True) |
| 185 | + parser.add_argument("--current-run-id", required=True) |
| 186 | + parser.add_argument("--baseline-run-id", required=True) |
| 187 | + parser.add_argument("--output", required=True, type=Path) |
| 188 | + return parser.parse_args() |
| 189 | + |
| 190 | + |
| 191 | +def _read_json(path: Path) -> object: |
| 192 | + try: |
| 193 | + return json.loads(path.read_bytes()) |
| 194 | + except (OSError, TypeError, json.JSONDecodeError) as exc: |
| 195 | + raise StrategyWatcherArtifactError("invalid strategy performance artifact") from exc |
| 196 | + |
| 197 | + |
| 198 | +def main() -> None: |
| 199 | + args = _arguments() |
| 200 | + payload = build_strategy_watcher_artifact_payload( |
| 201 | + current_artifact=_read_json(args.current), |
| 202 | + baseline_artifact=_read_json(args.baseline), |
| 203 | + source_repository=args.source_repository, |
| 204 | + workflow_file=args.workflow_file, |
| 205 | + current_run_id=args.current_run_id, |
| 206 | + baseline_run_id=args.baseline_run_id, |
| 207 | + ) |
| 208 | + args.output.write_text(json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False), encoding="utf-8") |
| 209 | + |
| 210 | + |
| 211 | +if __name__ == "__main__": |
| 212 | + main() |
0 commit comments