From f18d2e62eaae1097ed9825ef9594333e0b9f2794 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:51:47 +0800 Subject: [PATCH] fix: sync trusted lifecycle artifacts for quant monitor Co-Authored-By: Codex --- ops/quant-monitor/README.md | 12 + ops/quant-monitor/scripts/common_env.sh | 6 +- ops/quant-monitor/scripts/health_check.sh | 1 + ops/quant-monitor/scripts/health_cycle.py | 125 ++- .../scripts/setup_vps_runtime.sh | 7 +- .../scripts/sync_lifecycle_artifacts.py | 772 ++++++++++++++++++ .../tests/test_deploy_scripts.py | 29 +- .../tests/test_lifecycle_artifact_sync.py | 342 ++++++++ .../tests/test_monitor_fail_closed.py | 95 +++ 9 files changed, 1378 insertions(+), 11 deletions(-) create mode 100644 ops/quant-monitor/scripts/sync_lifecycle_artifacts.py create mode 100644 ops/quant-monitor/tests/test_lifecycle_artifact_sync.py diff --git a/ops/quant-monitor/README.md b/ops/quant-monitor/README.md index 376a7918..f7099dbf 100644 --- a/ops/quant-monitor/README.md +++ b/ops/quant-monitor/README.md @@ -11,6 +11,18 @@ bash scripts/health_check.sh bash scripts/daily_briefing.sh ``` +`health_check.sh` 会先更新代码仓库，再从四个策略仓库选择最近 7 天内、 +由 `main` 分支定时或手动 workflow 的成功 `preflight_backtests` 任务生成的 +`lifecycle-preflight-*` 工件。工件经路径、文件类型、domain/profile、JSON/CSV +contract 和大小限制校验后原子切换；代码仓库与 lifecycle 数据分别保存在： + +- `PROJECTS_ROOT`：策略代码和 `QuantPlatformKit`； +- `QUANT_PROJECTS_ROOT`：只读收益矩阵镜像； +- `LIFECYCLE_LOCAL_ROOT`：backtest、monitor snapshot 和 drift 状态。 + +任一 domain 缺少可信工件时只阻断该 domain，并写入 +`data/lifecycle-artifacts/status.json`；不会回退到演示或合成数据。 + ## Telegram（量化哨兵） Token 从 GCP Secret `quant-sentinel-telegram-bot-token` 加载；**不要**把 token 或 chat id 写进 git。 diff --git a/ops/quant-monitor/scripts/common_env.sh b/ops/quant-monitor/scripts/common_env.sh index 73fdb3ea..4c149fdc 100755 --- a/ops/quant-monitor/scripts/common_env.sh +++ b/ops/quant-monitor/scripts/common_env.sh @@ -4,8 +4,9 @@ set -euo pipefail ROOT="${QUANT_MONITOR_ROOT:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}" AAB_ROOT="${AIAUDIT_BRIDGE_ROOT:-$(cd "$ROOT/../.." && pwd)}" -PROJECTS_ROOT="${QUANT_PROJECTS_ROOT:-${PROJECTS_ROOT:-$HOME/Projects}}" -QUANT_PROJECTS_ROOT="$PROJECTS_ROOT" +PROJECTS_ROOT="${PROJECTS_ROOT:-$HOME/Projects}" +QUANT_PROJECTS_ROOT="${QUANT_PROJECTS_ROOT:-$ROOT/data/lifecycle-projects}" +LIFECYCLE_LOCAL_ROOT="${LIFECYCLE_LOCAL_ROOT:-$ROOT/data/lifecycle-store}" QPK_ROOT="${QUANT_PLATFORM_KIT_ROOT:-$PROJECTS_ROOT/QuantPlatformKit}" VENV="${QUANT_MONITOR_VENV:-$ROOT/.venv}" @@ -14,6 +15,7 @@ export AIAUDIT_BRIDGE_ROOT="$AAB_ROOT" export QUANT_PLATFORM_KIT_ROOT="$QPK_ROOT" export PROJECTS_ROOT export QUANT_PROJECTS_ROOT +export LIFECYCLE_LOCAL_ROOT if [[ -x "$VENV/bin/python" ]]; then export PATH="$VENV/bin:$PATH" diff --git a/ops/quant-monitor/scripts/health_check.sh b/ops/quant-monitor/scripts/health_check.sh index d995fa23..e161248f 100755 --- a/ops/quant-monitor/scripts/health_check.sh +++ b/ops/quant-monitor/scripts/health_check.sh @@ -5,4 +5,5 @@ ROOT="${QUANT_MONITOR_ROOT:-$(cd "$(dirname "$0")/.." && pwd)}" source "$ROOT/scripts/common_env.sh" bash "$ROOT/scripts/sync_strategy_repos.sh" +python3 "$ROOT/scripts/sync_lifecycle_artifacts.py" python3 "$ROOT/scripts/health_cycle.py" diff --git a/ops/quant-monitor/scripts/health_cycle.py b/ops/quant-monitor/scripts/health_cycle.py index d575bd05..3a1f8cf8 100755 --- a/ops/quant-monitor/scripts/health_cycle.py +++ b/ops/quant-monitor/scripts/health_cycle.py @@ -6,10 +6,11 @@ import hashlib import json import os +import re import subprocess import sys import tempfile -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any @@ -18,6 +19,9 @@ DRIFT_REVIEW = 0.50 DRIFT_CRITICAL = 0.75 _ALERT_STATE_RELATIVE_PATH = Path("data/alert-state/health_cycle.json") +_ARTIFACT_STATUS_RELATIVE_PATH = Path("data/lifecycle-artifacts/status.json") +_ARTIFACT_STATUS_SCHEMA = "quant_monitor_lifecycle_artifact_status.v1" +_SAFE_TOKEN = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,99}$") def _collect_drift_results(run_drift_detection, *, domains=DOMAINS): @@ -37,6 +41,111 @@ def _collect_drift_results(run_drift_detection, *, domains=DOMAINS): return results, errors +def _refresh_and_collect_drift(run_monitor, run_drift_detection, *, domains=DOMAINS): + snapshots: dict[str, list[Any]] = {} + results: dict[str, list[Any]] = {} + errors: list[dict[str, str]] = [] + for domain in domains: + try: + domain_snapshots = list(run_monitor(domain)) + if not domain_snapshots: + raise RuntimeError("monitor produced no snapshots") + snapshots[domain] = domain_snapshots + except Exception as exc: + errors.append( + { + "domain": domain, + "code": "monitor_data_unavailable", + "error_type": type(exc).__name__, + } + ) + continue + try: + results[domain] = list(run_drift_detection(domain)) + except Exception as exc: + errors.append( + { + "domain": domain, + "code": "drift_data_unavailable", + "error_type": type(exc).__name__, + } + ) + return snapshots, results, errors + + +def _artifact_status_error( + domain: str, + *, + code: str = "artifact_sync_status_unavailable", + error_type: str = "RuntimeError", +) -> dict[str, str]: + safe_code = code if _SAFE_TOKEN.fullmatch(code) else "artifact_sync_status_unavailable" + safe_error_type = error_type if _SAFE_TOKEN.fullmatch(error_type) else "RuntimeError" + return {"domain": domain, "code": safe_code, "error_type": safe_error_type} + + +def _load_lifecycle_artifact_status( + root: Path, + *, + domains=DOMAINS, + now: datetime | None = None, + max_age: timedelta = timedelta(hours=2), +) -> tuple[tuple[str, ...], list[dict[str, str]]]: + path = root / _ARTIFACT_STATUS_RELATIVE_PATH + try: + payload = json.loads(path.read_text(encoding="utf-8")) + as_of = datetime.fromisoformat(str(payload["as_of"]).replace("Z", "+00:00")) + if as_of.tzinfo is None: + raise ValueError("status timestamp has no timezone") + current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + age = current - as_of.astimezone(timezone.utc) + domain_statuses = payload["domains"] + if ( + not isinstance(payload, dict) + or payload.get("schema_version") != _ARTIFACT_STATUS_SCHEMA + or not isinstance(domain_statuses, dict) + or age > max_age + or age < -timedelta(minutes=5) + ): + raise ValueError("artifact status is invalid or stale") + except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + return (), [ + _artifact_status_error(domain, error_type=type(exc).__name__) + for domain in domains + ] + + ready: list[str] = [] + errors: list[dict[str, str]] = [] + for domain in domains: + status = domain_statuses.get(domain) + if not isinstance(status, dict): + errors.append(_artifact_status_error(domain)) + continue + profiles = status.get("profiles") + valid_ready = ( + status.get("status") == "ready" + and isinstance(status.get("artifact_id"), int) + and status["artifact_id"] > 0 + and isinstance(status.get("run_id"), int) + and status["run_id"] > 0 + and re.fullmatch(r"[0-9a-f]{40}", str(status.get("head_sha") or "")) + and isinstance(profiles, list) + and bool(profiles) + and all(isinstance(profile, str) and profile for profile in profiles) + ) + if valid_ready: + ready.append(domain) + continue + errors.append( + _artifact_status_error( + domain, + code=str(status.get("code") or "artifact_sync_status_unavailable"), + error_type=str(status.get("error_type") or "RuntimeError"), + ) + ) + return tuple(ready), errors + + def _alert_fingerprint(lines: list[str]) -> str: payload = "\n".join(sorted(str(line) for line in lines)) return hashlib.sha256(payload.encode("utf-8")).hexdigest() @@ -161,8 +270,15 @@ def main() -> int: from quant_platform_kit.strategy_lifecycle.codex_integration import create_issues_for_domain from quant_platform_kit.strategy_lifecycle.drift_detector import run_drift_detection from quant_platform_kit.strategy_lifecycle.health_dashboard import build_dashboard + from quant_platform_kit.strategy_lifecycle.performance_monitor import run_monitor - drift_results, drift_errors = _collect_drift_results(run_drift_detection) + ready_domains, artifact_errors = _load_lifecycle_artifact_status(root) + snapshot_results, drift_results, lifecycle_errors = _refresh_and_collect_drift( + run_monitor, + run_drift_detection, + domains=ready_domains, + ) + data_errors = artifact_errors + lifecycle_errors build_dashboard(output_dir=str(dash_dir), output_format="json") strategies: list[dict[str, Any]] = [] @@ -250,7 +366,7 @@ def main() -> int: ) data_error_lines: list[str] = [] - for error in drift_errors: + for error in data_errors: data_error_lines.append( f"[{error['domain']}] {error['code']} ({error['error_type']})" ) @@ -281,7 +397,8 @@ def main() -> int: "telegram_alerts": notify_lines, "telegram_sent": telegram_sent, "duplicate_alert_suppressed": duplicate_alert_suppressed, - "data_errors": drift_errors, + "data_errors": data_errors, + "snapshot_count": sum(len(rows) for rows in snapshot_results.values()), "issues_created": len([r for r in issue_results if r.get("issue_url")]), "ok": not notify_lines and not collector_payload_invalid, "collector_payload_valid": not collector_payload_invalid, diff --git a/ops/quant-monitor/scripts/setup_vps_runtime.sh b/ops/quant-monitor/scripts/setup_vps_runtime.sh index 2c4d34f4..99a3fab9 100755 --- a/ops/quant-monitor/scripts/setup_vps_runtime.sh +++ b/ops/quant-monitor/scripts/setup_vps_runtime.sh @@ -34,7 +34,12 @@ python3 -m venv "$VENV" "$VENV/bin/pip" install numpy pandas google-cloud-storage if ! command -v gh >/dev/null 2>&1; then - echo "[setup] warning: gh CLI not installed; drift issues will be skipped" >&2 + echo "[setup] gh CLI is required for trusted lifecycle artifact synchronization" >&2 + exit 1 +fi +if ! gh auth status >/dev/null 2>&1; then + echo "[setup] gh CLI authentication is required for lifecycle artifacts" >&2 + exit 1 fi if ! command -v gcloud >/dev/null 2>&1; then echo "[setup] warning: gcloud not installed; telegram env load may fail" >&2 diff --git a/ops/quant-monitor/scripts/sync_lifecycle_artifacts.py b/ops/quant-monitor/scripts/sync_lifecycle_artifacts.py new file mode 100644 index 00000000..bf6bfcf9 --- /dev/null +++ b/ops/quant-monitor/scripts/sync_lifecycle_artifacts.py @@ -0,0 +1,772 @@ +#!/usr/bin/env python3 +"""Synchronize trusted lifecycle preflight artifacts for quant-monitor.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import io +import json +import math +import os +import re +import shutil +import stat +import subprocess +import tempfile +import time +import zipfile +from datetime import date, datetime, timedelta, timezone +from pathlib import Path, PurePosixPath +from typing import Any, Callable, Mapping, Sequence + + +DOMAIN_CONFIGS: dict[str, dict[str, str]] = { + "cn_equity": { + "domain": "cn_equity", + "repository": "QuantStrategyLab/CnEquityStrategies", + "snapshot_repository": "CnEquitySnapshotPipelines", + "artifact_prefix": "lifecycle-preflight-", + "workflow_path": ".github/workflows/drift-check.yml", + "preflight_job": "preflight_backtests", + "benchmark_column": "buy_hold_510300", + }, + "hk_equity": { + "domain": "hk_equity", + "repository": "QuantStrategyLab/HkEquityStrategies", + "snapshot_repository": "HkEquitySnapshotPipelines", + "artifact_prefix": "lifecycle-preflight-", + "workflow_path": ".github/workflows/drift-check.yml", + "preflight_job": "preflight_backtests", + "benchmark_column": "buy_hold_2800", + }, + "us_equity": { + "domain": "us_equity", + "repository": "QuantStrategyLab/UsEquityStrategies", + "snapshot_repository": "UsEquitySnapshotPipelines", + "artifact_prefix": "lifecycle-preflight-", + "workflow_path": ".github/workflows/drift-check.yml", + "preflight_job": "preflight_backtests", + "benchmark_column": "buy_hold_SPY", + }, + "crypto": { + "domain": "crypto", + "repository": "QuantStrategyLab/CryptoStrategies", + "snapshot_repository": "CryptoLivePoolPipelines", + "artifact_prefix": "lifecycle-preflight-", + "workflow_path": ".github/workflows/drift-check.yml", + "preflight_job": "preflight_backtests", + "benchmark_column": "buy_hold_BTC", + }, +} + +_TRUSTED_EVENTS = frozenset({"schedule", "workflow_dispatch"}) +_SHA_PATTERN = re.compile(r"^[0-9a-f]{40}$") +_PROFILE_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,99}$") +_MAX_FILES = 256 +_MAX_FILE_BYTES = 32 * 1024 * 1024 +_MAX_TOTAL_BYTES = 256 * 1024 * 1024 +_MANIFEST_NAME = ".artifact-manifest.json" + + +class LifecycleArtifactError(RuntimeError): + """A fail-closed lifecycle artifact synchronization error.""" + + def __init__(self, message: str, *, code: str = "artifact_invalid") -> None: + super().__init__(message) + self.code = code + + +def _parse_timestamp(value: Any) -> datetime: + text = str(value or "").strip() + if text.endswith("Z"): + text = f"{text[:-1]}+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise LifecycleArtifactError("artifact timestamp is invalid") from exc + if parsed.tzinfo is None: + raise LifecycleArtifactError("artifact timestamp must include a timezone") + return parsed.astimezone(timezone.utc) + + +def _trusted_run( + config: Mapping[str, str], + run: Mapping[str, Any], + jobs_payload: Mapping[str, Any], +) -> bool: + repository = run.get("head_repository") + jobs = jobs_payload.get("jobs") + if not isinstance(repository, Mapping) or not isinstance(jobs, list): + return False + if ( + run.get("status") != "completed" + or run.get("event") not in _TRUSTED_EVENTS + or run.get("head_branch") != "main" + or run.get("path") != config["workflow_path"] + or repository.get("full_name") != config["repository"] + or not _SHA_PATTERN.fullmatch(str(run.get("head_sha") or "")) + ): + return False + return any( + isinstance(job, Mapping) + and job.get("name") == config["preflight_job"] + and job.get("status") == "completed" + and job.get("conclusion") == "success" + for job in jobs + ) + + +def select_trusted_artifact( + config: Mapping[str, str], + artifacts: Sequence[Mapping[str, Any]], + *, + load_run: Callable[[int], Mapping[str, Any]], + load_jobs: Callable[[int], Mapping[str, Any]], + now: datetime, + max_age: timedelta, +) -> dict[str, Any]: + """Return the newest artifact whose workflow provenance is trusted.""" + + now = now.astimezone(timezone.utc) + prefix = config["artifact_prefix"] + candidates: list[tuple[datetime, Mapping[str, Any]]] = [] + for artifact in artifacts: + try: + created_at = _parse_timestamp(artifact.get("created_at")) + except LifecycleArtifactError: + continue + if ( + artifact.get("expired") is True + or not str(artifact.get("name") or "").startswith(prefix) + or created_at > now + timedelta(minutes=5) + or now - created_at > max_age + ): + continue + candidates.append((created_at, artifact)) + + for created_at, artifact in sorted(candidates, key=lambda item: item[0], reverse=True): + workflow_run = artifact.get("workflow_run") + if not isinstance(workflow_run, Mapping): + continue + try: + run_id = int(workflow_run.get("id")) + artifact_id = int(artifact.get("id")) + except (TypeError, ValueError): + continue + expected_name = re.fullmatch( + rf"{re.escape(prefix)}{run_id}-[1-9][0-9]*", + str(artifact.get("name") or ""), + ) + if expected_name is None: + continue + run = load_run(run_id) + if int(run.get("id") or 0) != run_id: + continue + jobs = load_jobs(run_id) + if not _trusted_run(config, run, jobs): + continue + return { + **artifact, + "id": artifact_id, + "_trusted_run": dict(run), + "_created_at": created_at.isoformat(), + } + + raise LifecycleArtifactError( + f"no trusted lifecycle artifact for {config['domain']}", + code="trusted_artifact_unavailable", + ) + + +def _safe_member_path(name: str) -> PurePosixPath: + if not name or "\x00" in name or "\\" in name: + raise LifecycleArtifactError("archive contains an unsafe path") + path = PurePosixPath(name) + if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts): + raise LifecycleArtifactError("archive contains an unsafe path") + return path + + +def _classify_member( + path: PurePosixPath, + config: Mapping[str, str], +) -> tuple[str, str] | None: + parts = path.parts + domain = config["domain"] + snapshot_repository = config["snapshot_repository"] + if ( + len(parts) == 6 + and parts[:4] == ("data", "lifecycle_store", "backtest", domain) + and _PROFILE_PATTERN.fullmatch(parts[4]) + and parts[5].startswith("backtest_") + and parts[5].endswith(".json") + ): + return "backtest", parts[4] + if ( + len(parts) == 6 + and parts[:4] == ("external", snapshot_repository, "data", "output") + and _PROFILE_PATTERN.fullmatch(parts[4]) + and parts[5] == "portfolio_and_tracker_returns.csv" + ): + return "matrix", parts[4] + return None + + +def _validate_backtest( + raw: bytes, + *, + domain: str, + profile: str, +) -> None: + try: + payload = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise LifecycleArtifactError("backtest JSON is invalid") from exc + if not isinstance(payload, Mapping): + raise LifecycleArtifactError("backtest JSON must contain an object") + if payload.get("domain") != domain or payload.get("strategy_profile") != profile: + raise LifecycleArtifactError("backtest domain or profile does not match its path") + if ( + payload.get("schema_version") != "strategy_lifecycle.v1" + or not str(payload.get("param_set_id") or "").strip() + or not isinstance(payload.get("params"), Mapping) + or isinstance(payload.get("param_version"), bool) + or not isinstance(payload.get("param_version"), int) + or payload["param_version"] < 0 + or isinstance(payload.get("observation_count"), bool) + or not isinstance(payload.get("observation_count"), int) + or payload["observation_count"] < 2 + or not str(payload.get("source_script") or "").strip() + ): + raise LifecycleArtifactError("backtest JSON lifecycle contract is invalid") + for key in ("sharpe_ratio", "max_drawdown", "cagr", "volatility"): + try: + value = float(payload[key]) + except (KeyError, TypeError, ValueError) as exc: + raise LifecycleArtifactError( + "backtest JSON lifecycle metrics are invalid" + ) from exc + if not math.isfinite(value): + raise LifecycleArtifactError("backtest JSON lifecycle metrics are invalid") + try: + start_date = date.fromisoformat(str(payload["start_date"])) + end_date = date.fromisoformat(str(payload["end_date"])) + _parse_timestamp(payload["computed_at"]) + except (KeyError, ValueError) as exc: + raise LifecycleArtifactError("backtest JSON lifecycle dates are invalid") from exc + if end_date < start_date: + raise LifecycleArtifactError("backtest JSON lifecycle dates are invalid") + + +def _validate_matrix(raw: bytes, *, profile: str, benchmark_column: str) -> None: + try: + rows = list(csv.reader(io.StringIO(raw.decode("utf-8-sig")))) + except (UnicodeDecodeError, csv.Error) as exc: + raise LifecycleArtifactError("return matrix CSV is invalid") from exc + if len(rows) < 3: + raise LifecycleArtifactError("return matrix must contain at least two data rows") + header = [column.strip() for column in rows[0]] + if ( + len(header) != 3 + or header[0] != "as_of" + or header[1] != profile + or header[2] != benchmark_column + ): + raise LifecycleArtifactError("return matrix profile or header is invalid") + previous_date = "" + observation_counts = [0, 0] + for row in rows[1:]: + if len(row) != len(header): + raise LifecycleArtifactError("return matrix row width is invalid") + date_value = row[0].strip() + try: + datetime.strptime(date_value, "%Y-%m-%d") + except (ValueError, TypeError) as exc: + raise LifecycleArtifactError("return matrix contains invalid values") from exc + for index, raw_value in enumerate(row[1:]): + if not raw_value.strip(): + continue + try: + value = float(raw_value) + except ValueError as exc: + raise LifecycleArtifactError( + "return matrix contains invalid values" + ) from exc + if not math.isfinite(value): + raise LifecycleArtifactError("return matrix contains non-finite values") + observation_counts[index] += 1 + if previous_date and date_value <= previous_date: + raise LifecycleArtifactError("return matrix dates must be strictly increasing") + previous_date = date_value + if any(count < 2 for count in observation_counts): + raise LifecycleArtifactError("return matrix has insufficient finite observations") + + +def extract_validated_archive( + archive_path: Path, + output_dir: Path, + config: Mapping[str, str], +) -> dict[str, Any]: + """Validate a lifecycle artifact completely before writing allowlisted files.""" + + files: dict[PurePosixPath, bytes] = {} + total_size = 0 + backtest_profiles: set[str] = set() + matrix_profiles: set[str] = set() + try: + archive = zipfile.ZipFile(archive_path) + except (OSError, zipfile.BadZipFile) as exc: + raise LifecycleArtifactError("artifact archive is invalid") from exc + + with archive: + members = archive.infolist() + if len(members) > _MAX_FILES: + raise LifecycleArtifactError("artifact archive contains too many entries") + for member in members: + path = _safe_member_path(member.filename) + mode = (member.external_attr >> 16) & 0xFFFF + if member.is_dir(): + continue + file_type = stat.S_IFMT(mode) + if file_type and not stat.S_ISREG(mode): + raise LifecycleArtifactError("artifact archive contains a non-regular file") + classification = _classify_member(path, config) + if classification is None: + raise LifecycleArtifactError("artifact archive contains an unexpected file") + if path in files: + raise LifecycleArtifactError("artifact archive contains a duplicate path") + if member.file_size > _MAX_FILE_BYTES: + raise LifecycleArtifactError("artifact archive file exceeds the size limit") + total_size += member.file_size + if total_size > _MAX_TOTAL_BYTES: + raise LifecycleArtifactError("artifact archive exceeds the total size limit") + try: + raw = archive.read(member) + except (OSError, RuntimeError, zipfile.BadZipFile) as exc: + raise LifecycleArtifactError("artifact archive could not be read") from exc + if len(raw) != member.file_size: + raise LifecycleArtifactError("artifact archive entry size is inconsistent") + kind, profile = classification + if kind == "backtest": + _validate_backtest(raw, domain=config["domain"], profile=profile) + backtest_profiles.add(profile) + else: + _validate_matrix( + raw, + profile=profile, + benchmark_column=config["benchmark_column"], + ) + matrix_profiles.add(profile) + files[path] = raw + + if not files or not backtest_profiles or backtest_profiles != matrix_profiles: + raise LifecycleArtifactError("backtest and return matrix profiles do not match") + if output_dir.exists() or os.path.lexists(output_dir): + raise LifecycleArtifactError("artifact output directory already exists") + output_dir.mkdir(parents=True) + for path, raw in files.items(): + destination = output_dir.joinpath(*path.parts) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(raw) + return { + "profiles": sorted(backtest_profiles), + "file_count": len(files), + "total_bytes": total_size, + "sha256": { + path.as_posix(): hashlib.sha256(raw).hexdigest() + for path, raw in sorted(files.items(), key=lambda item: item[0].as_posix()) + }, + } + + +def validate_stored_version( + version_dir: Path, + manifest: Mapping[str, Any], + config: Mapping[str, str], +) -> None: + """Revalidate cached files and hashes before each activation.""" + + hashes = manifest.get("sha256") + profiles = manifest.get("profiles") + if not isinstance(hashes, Mapping) or not hashes or not isinstance(profiles, list): + raise LifecycleArtifactError("stored artifact manifest is incomplete") + expected_files: set[str] = set() + backtest_profiles: set[str] = set() + matrix_profiles: set[str] = set() + for raw_name, raw_digest in hashes.items(): + path = _safe_member_path(str(raw_name)) + classification = _classify_member(path, config) + digest = str(raw_digest or "") + if classification is None or not re.fullmatch(r"[0-9a-f]{64}", digest): + raise LifecycleArtifactError("stored artifact manifest contains invalid files") + destination = version_dir.joinpath(*path.parts) + managed_parent = version_dir + has_symlink_parent = version_dir.is_symlink() + for part in path.parts[:-1]: + managed_parent /= part + has_symlink_parent = has_symlink_parent or managed_parent.is_symlink() + if destination.is_symlink() or has_symlink_parent: + raise LifecycleArtifactError("stored artifact contains a symlink") + if not destination.is_file(): + raise LifecycleArtifactError("stored artifact file is missing") + raw = destination.read_bytes() + if hashlib.sha256(raw).hexdigest() != digest: + raise LifecycleArtifactError("stored artifact file hash does not match") + kind, profile = classification + if kind == "backtest": + _validate_backtest(raw, domain=config["domain"], profile=profile) + backtest_profiles.add(profile) + else: + _validate_matrix( + raw, + profile=profile, + benchmark_column=config["benchmark_column"], + ) + matrix_profiles.add(profile) + expected_files.add(path.as_posix()) + + actual_files: set[str] = set() + for path in version_dir.rglob("*"): + if path.is_symlink(): + raise LifecycleArtifactError("stored artifact contains a symlink") + if path.is_file(): + relative = path.relative_to(version_dir).as_posix() + if relative != _MANIFEST_NAME: + actual_files.add(relative) + if ( + actual_files != expected_files + or backtest_profiles != matrix_profiles + or sorted(backtest_profiles) != sorted(profiles) + ): + raise LifecycleArtifactError("stored artifact manifest does not match its files") + + +def _replace_managed_symlink(link: Path, target: Path) -> None: + link.parent.mkdir(parents=True, exist_ok=True) + if os.path.lexists(link) and not link.is_symlink(): + raise LifecycleArtifactError( + f"refusing to replace unmanaged path: {link}", + code="consumer_path_conflict", + ) + temporary = link.parent / f".{link.name}.tmp-{os.getpid()}" + try: + if os.path.lexists(temporary): + temporary.unlink() + temporary.symlink_to(target) + temporary.replace(link) + finally: + if os.path.lexists(temporary): + temporary.unlink() + + +def activate_version( + version_dir: Path, + config: Mapping[str, str], + *, + projects_root: Path, + lifecycle_root: Path, +) -> None: + """Atomically activate a validated version for both lifecycle consumers.""" + + version_dir = version_dir.absolute() + domain = config["domain"] + if ( + not version_dir.is_dir() + or version_dir.parent.name != domain + or version_dir.parent.parent.name != "versions" + ): + raise LifecycleArtifactError("artifact version path is invalid") + artifacts_root = version_dir.parents[2] + current_link = artifacts_root / "current" / domain + _replace_managed_symlink(current_link, version_dir) + + matrix_target = ( + current_link + / "external" + / config["snapshot_repository"] + / "data" + / "output" + ) + backtest_target = ( + current_link + / "data" + / "lifecycle_store" + / "backtest" + / domain + ) + if not matrix_target.is_dir() or not backtest_target.is_dir(): + raise LifecycleArtifactError("activated artifact is missing required directories") + _replace_managed_symlink( + projects_root / config["snapshot_repository"] / "data" / "output", + matrix_target, + ) + _replace_managed_symlink( + lifecycle_root / "backtest" / domain, + backtest_target, + ) + + +def _run_gh(args: Sequence[str], *, binary: bool = False) -> Any: + env = {**os.environ, "GH_PROMPT": "disabled"} + for attempt in range(3): + result = subprocess.run( + ["gh", "api", *args], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + check=False, + text=not binary, + ) + if result.returncode == 0: + if binary: + return result.stdout + try: + return json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise LifecycleArtifactError( + "GitHub API returned invalid JSON", + code="github_api_invalid", + ) from exc + if attempt < 2: + time.sleep(2**attempt) + raise LifecycleArtifactError( + "GitHub API request failed", + code="github_api_unavailable", + ) + + +def _atomic_write_json(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + delete=False, + ) as temp_file: + temp_path = Path(temp_file.name) + json.dump(payload, temp_file, ensure_ascii=False, indent=2, sort_keys=True) + temp_file.write("\n") + temp_file.flush() + os.fsync(temp_file.fileno()) + try: + temp_path.replace(path) + finally: + temp_path.unlink(missing_ok=True) + + +def _load_or_download_version( + artifact: Mapping[str, Any], + config: Mapping[str, str], + *, + artifacts_root: Path, +) -> tuple[Path, Mapping[str, Any]]: + artifact_id = int(artifact["id"]) + version_dir = artifacts_root / "versions" / config["domain"] / str(artifact_id) + manifest_path = version_dir / _MANIFEST_NAME + if manifest_path.is_file(): + if manifest_path.is_symlink(): + raise LifecycleArtifactError("stored artifact manifest must not be a symlink") + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise LifecycleArtifactError("stored artifact manifest is invalid") from exc + if ( + not isinstance(manifest, Mapping) + or manifest.get("artifact_id") != artifact_id + or manifest.get("domain") != config["domain"] + or manifest.get("repository") != config["repository"] + or manifest.get("artifact_name") != artifact["name"] + or manifest.get("run_id") != int(artifact["_trusted_run"]["id"]) + or manifest.get("head_sha") != artifact["_trusted_run"]["head_sha"] + or not isinstance(manifest.get("profiles"), list) + ): + raise LifecycleArtifactError("stored artifact manifest does not match") + validate_stored_version(version_dir, manifest, config) + return version_dir, manifest + if os.path.lexists(version_dir): + raise LifecycleArtifactError("stored artifact version is missing its manifest") + + version_dir.parent.mkdir(parents=True, exist_ok=True) + staging = version_dir.parent / f".{artifact_id}.staging-{os.getpid()}" + if os.path.lexists(staging): + shutil.rmtree(staging) + try: + archive_bytes = _run_gh( + [ + f"/repos/{config['repository']}/actions/artifacts/{artifact_id}/zip", + "-H", + "Accept: application/vnd.github+json", + ], + binary=True, + ) + with tempfile.NamedTemporaryFile( + dir=version_dir.parent, + prefix=f".{artifact_id}.", + suffix=".zip", + delete=False, + ) as archive_file: + archive_path = Path(archive_file.name) + archive_file.write(archive_bytes) + try: + validation = extract_validated_archive(archive_path, staging, config) + finally: + archive_path.unlink(missing_ok=True) + run = artifact["_trusted_run"] + manifest = { + "schema_version": "quant_monitor_lifecycle_artifact.v1", + "domain": config["domain"], + "repository": config["repository"], + "artifact_id": artifact_id, + "artifact_name": artifact["name"], + "run_id": int(run["id"]), + "head_sha": run["head_sha"], + "created_at": artifact["_created_at"], + **validation, + } + _atomic_write_json(staging / _MANIFEST_NAME, manifest) + validate_stored_version(staging, manifest, config) + try: + staging.replace(version_dir) + except FileExistsError: + shutil.rmtree(staging) + existing_manifest_path = version_dir / _MANIFEST_NAME + if existing_manifest_path.is_symlink() or not existing_manifest_path.is_file(): + raise LifecycleArtifactError("stored artifact version is invalid") + existing_manifest = json.loads( + existing_manifest_path.read_text(encoding="utf-8") + ) + if not isinstance(existing_manifest, Mapping): + raise LifecycleArtifactError("stored artifact manifest is invalid") + validate_stored_version(version_dir, existing_manifest, config) + return version_dir, existing_manifest + return version_dir, manifest + finally: + if staging.exists(): + shutil.rmtree(staging) + + +def _sync_domain( + config: Mapping[str, str], + *, + artifacts_root: Path, + projects_root: Path, + lifecycle_root: Path, + max_age: timedelta, + now: datetime, +) -> dict[str, Any]: + repository = config["repository"] + payload = _run_gh( + [ + f"/repos/{repository}/actions/artifacts?per_page=100", + "-H", + "Accept: application/vnd.github+json", + ] + ) + artifacts = payload.get("artifacts") if isinstance(payload, Mapping) else None + if not isinstance(artifacts, list): + raise LifecycleArtifactError( + "GitHub artifact listing is invalid", + code="github_api_invalid", + ) + selected = select_trusted_artifact( + config, + artifacts, + load_run=lambda run_id: _run_gh( + [f"/repos/{repository}/actions/runs/{run_id}"] + ), + load_jobs=lambda run_id: _run_gh( + [f"/repos/{repository}/actions/runs/{run_id}/jobs?per_page=100"] + ), + now=now, + max_age=max_age, + ) + version_dir, manifest = _load_or_download_version( + selected, + config, + artifacts_root=artifacts_root, + ) + activate_version( + version_dir, + config, + projects_root=projects_root, + lifecycle_root=lifecycle_root, + ) + created_at = _parse_timestamp(selected["created_at"]) + return { + "status": "ready", + "artifact_id": int(selected["id"]), + "run_id": int(selected["_trusted_run"]["id"]), + "head_sha": selected["_trusted_run"]["head_sha"], + "created_at": created_at.isoformat(), + "age_seconds": max(0, int((now - created_at).total_seconds())), + "profiles": list(manifest["profiles"]), + } + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--domain", action="append", choices=tuple(DOMAIN_CONFIGS)) + args = parser.parse_args(argv) + + root = Path( + os.environ.get("QUANT_MONITOR_ROOT") + or Path(__file__).resolve().parents[1] + ) + artifacts_root = root / "data" / "lifecycle-artifacts" + projects_root = Path( + os.environ.get("QUANT_PROJECTS_ROOT") + or root / "data" / "lifecycle-projects" + ) + lifecycle_root = Path( + os.environ.get("LIFECYCLE_LOCAL_ROOT") + or root / "data" / "lifecycle-store" + ) + try: + max_age_hours = int( + os.environ.get("QUANT_LIFECYCLE_ARTIFACT_MAX_AGE_HOURS") or "168" + ) + except ValueError: + max_age_hours = 168 + max_age = timedelta(hours=max(1, max_age_hours)) + now = datetime.now(timezone.utc) + domains = tuple(args.domain or DOMAIN_CONFIGS) + statuses: dict[str, dict[str, Any]] = {} + for domain in domains: + try: + statuses[domain] = _sync_domain( + DOMAIN_CONFIGS[domain], + artifacts_root=artifacts_root, + projects_root=projects_root, + lifecycle_root=lifecycle_root, + max_age=max_age, + now=now, + ) + except LifecycleArtifactError as exc: + statuses[domain] = { + "status": "error", + "code": exc.code, + "error_type": type(exc).__name__, + } + except Exception as exc: + statuses[domain] = { + "status": "error", + "code": "artifact_sync_unexpected", + "error_type": type(exc).__name__, + } + summary = { + "schema_version": "quant_monitor_lifecycle_artifact_status.v1", + "as_of": now.isoformat(), + "domains": statuses, + "ok": all(status.get("status") == "ready" for status in statuses.values()), + } + _atomic_write_json(artifacts_root / "status.json", summary) + print(json.dumps(summary, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ops/quant-monitor/tests/test_deploy_scripts.py b/ops/quant-monitor/tests/test_deploy_scripts.py index a5d2598a..3d823cba 100644 --- a/ops/quant-monitor/tests/test_deploy_scripts.py +++ b/ops/quant-monitor/tests/test_deploy_scripts.py @@ -9,23 +9,29 @@ class DeployScriptTests(unittest.TestCase): - def test_common_env_exports_canonical_projects_root(self) -> None: + def test_common_env_separates_code_and_lifecycle_data_roots(self) -> None: with tempfile.TemporaryDirectory() as home: env = os.environ.copy() for name in ( "PROJECTS_ROOT", "QUANT_PROJECTS_ROOT", "QUANT_PLATFORM_KIT_ROOT", + "LIFECYCLE_LOCAL_ROOT", ): env.pop(name, None) env["HOME"] = home + monitor_root = Path(home) / "monitor" + monitor_root.mkdir() + env["QUANT_MONITOR_ROOT"] = str(monitor_root) result = subprocess.run( [ "bash", "-c", ( f"source {ROOT / 'scripts' / 'common_env.sh'}; " - 'printf "%s\\n%s\\n" "${PROJECTS_ROOT-}" "${QUANT_PROJECTS_ROOT-}"' + 'printf "%s\\n%s\\n%s\\n%s\\n" ' + '"${PROJECTS_ROOT-}" "${QUANT_PROJECTS_ROOT-}" ' + '"${LIFECYCLE_LOCAL_ROOT-}" "${QUANT_PLATFORM_KIT_ROOT-}"' ), ], env=env, @@ -34,8 +40,23 @@ def test_common_env_exports_canonical_projects_root(self) -> None: check=True, ) - expected = str(Path(home) / "Projects") - self.assertEqual(result.stdout.splitlines(), [expected, expected]) + self.assertEqual( + result.stdout.splitlines(), + [ + str(Path(home) / "Projects"), + str(monitor_root / "data" / "lifecycle-projects"), + str(monitor_root / "data" / "lifecycle-store"), + str(Path(home) / "Projects" / "QuantPlatformKit"), + ], + ) + + def test_health_check_syncs_artifacts_before_monitoring(self) -> None: + script = (ROOT / "scripts" / "health_check.sh").read_text(encoding="utf-8") + + self.assertLess( + script.index("sync_lifecycle_artifacts.py"), + script.index("health_cycle.py"), + ) def test_deploy_script_normalizes_chat_id_and_excludes_bytecode(self) -> None: script = (ROOT / "scripts" / "deploy_to_vps.sh").read_text(encoding="utf-8") diff --git a/ops/quant-monitor/tests/test_lifecycle_artifact_sync.py b/ops/quant-monitor/tests/test_lifecycle_artifact_sync.py new file mode 100644 index 00000000..ef8032fa --- /dev/null +++ b/ops/quant-monitor/tests/test_lifecycle_artifact_sync.py @@ -0,0 +1,342 @@ +import importlib.util +import json +import os +import stat +import sys +import tempfile +import unittest +import zipfile +from datetime import datetime, timedelta, timezone +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def _load_script(name: str): + spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +SYNC = _load_script("sync_lifecycle_artifacts") + + +class LifecycleArtifactSyncTests(unittest.TestCase): + def setUp(self) -> None: + self.config = SYNC.DOMAIN_CONFIGS["us_equity"] + self.now = datetime(2026, 7, 30, 7, 0, tzinfo=timezone.utc) + + def _artifact(self, artifact_id: int, run_id: int, created_at: datetime): + return { + "id": artifact_id, + "name": f"lifecycle-preflight-{run_id}-1", + "expired": False, + "created_at": created_at.isoformat().replace("+00:00", "Z"), + "workflow_run": {"id": run_id}, + } + + def _run(self, run_id: int, **overrides): + payload = { + "id": run_id, + "status": "completed", + "event": "schedule", + "head_branch": "main", + "path": ".github/workflows/drift-check.yml", + "head_sha": "a" * 40, + "head_repository": {"full_name": self.config["repository"]}, + } + payload.update(overrides) + return payload + + @staticmethod + def _jobs(conclusion: str = "success"): + return { + "jobs": [ + { + "name": "preflight_backtests", + "conclusion": conclusion, + "status": "completed", + } + ] + } + + def test_selects_latest_trusted_preflight_artifact(self) -> None: + older = self._artifact(1, 11, self.now - timedelta(days=2)) + latest = self._artifact(2, 22, self.now - timedelta(hours=1)) + runs = {11: self._run(11), 22: self._run(22)} + + selected = SYNC.select_trusted_artifact( + self.config, + [older, latest], + load_run=lambda run_id: runs[run_id], + load_jobs=lambda _run_id: self._jobs(), + now=self.now, + max_age=timedelta(days=7), + ) + + self.assertEqual(selected["id"], 2) + + def test_rejects_untrusted_or_stale_artifacts(self) -> None: + untrusted = self._artifact(2, 22, self.now - timedelta(hours=1)) + stale = self._artifact(1, 11, self.now - timedelta(days=8)) + runs = { + 22: self._run(22, event="pull_request"), + 11: self._run(11), + } + + with self.assertRaisesRegex( + SYNC.LifecycleArtifactError, + "no trusted lifecycle artifact", + ): + SYNC.select_trusted_artifact( + self.config, + [untrusted, stale], + load_run=lambda run_id: runs[run_id], + load_jobs=lambda _run_id: self._jobs(), + now=self.now, + max_age=timedelta(days=7), + ) + + def test_requires_successful_preflight_job(self) -> None: + artifact = self._artifact(2, 22, self.now - timedelta(hours=1)) + + with self.assertRaises(SYNC.LifecycleArtifactError): + SYNC.select_trusted_artifact( + self.config, + [artifact], + load_run=lambda run_id: self._run(run_id), + load_jobs=lambda _run_id: self._jobs("failure"), + now=self.now, + max_age=timedelta(days=7), + ) + + def _write_valid_archive( + self, + path: Path, + *, + profile: str = "global_etf_rotation", + matrix_profile: str | None = None, + ) -> None: + matrix_profile = matrix_profile or profile + backtest = { + "strategy_profile": profile, + "domain": "us_equity", + "param_set_id": f"{profile}_wf", + "params": {}, + "param_version": 1, + "sharpe_ratio": 0.8, + "max_drawdown": -0.1, + "cagr": 0.12, + "volatility": 0.2, + "observation_count": 252, + "start_date": "2025-07-29", + "end_date": "2026-07-29", + "computed_at": "2026-07-29T08:00:00+00:00", + "source_script": "tests.fixture", + "schema_version": "strategy_lifecycle.v1", + } + with zipfile.ZipFile(path, "w") as bundle: + bundle.writestr( + f"data/lifecycle_store/backtest/us_equity/{profile}/backtest_v1.json", + json.dumps(backtest), + ) + bundle.writestr( + "external/UsEquitySnapshotPipelines/data/output/" + f"{profile}/portfolio_and_tracker_returns.csv", + f"as_of,{matrix_profile},buy_hold_SPY\n" + "2026-07-28,0.01,0.005\n" + "2026-07-29,-0.002,-0.001\n", + ) + + def test_extracts_validated_allowlisted_files(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + archive = root / "artifact.zip" + output = root / "version" + self._write_valid_archive(archive) + + result = SYNC.extract_validated_archive(archive, output, self.config) + + self.assertEqual(result["profiles"], ["global_etf_rotation"]) + self.assertTrue( + ( + output + / "external/UsEquitySnapshotPipelines/data/output/global_etf_rotation" + / "portfolio_and_tracker_returns.csv" + ).is_file() + ) + + def test_allows_sparse_benchmark_values_with_enough_history(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + archive = root / "artifact.zip" + with zipfile.ZipFile(archive, "w") as bundle: + bundle.writestr( + "data/lifecycle_store/backtest/us_equity/global_etf_rotation/" + "backtest_v1.json", + json.dumps( + { + "strategy_profile": "global_etf_rotation", + "domain": "us_equity", + "param_set_id": "global_etf_rotation_wf", + "params": {}, + "param_version": 1, + "sharpe_ratio": 0.8, + "max_drawdown": -0.1, + "cagr": 0.12, + "volatility": 0.2, + "observation_count": 252, + "start_date": "2025-07-29", + "end_date": "2026-07-29", + "computed_at": "2026-07-29T08:00:00+00:00", + "source_script": "tests.fixture", + "schema_version": "strategy_lifecycle.v1", + } + ), + ) + matrix_path = ( + "external/UsEquitySnapshotPipelines/data/output/" + "global_etf_rotation/portfolio_and_tracker_returns.csv" + ) + bundle.writestr( + matrix_path, + "as_of,global_etf_rotation,buy_hold_SPY\n" + "2026-07-27,0.003,0.002\n" + "2026-07-28,0.01,0.005\n" + "2026-07-29,-0.002,\n", + ) + + result = SYNC.extract_validated_archive( + archive, + root / "version", + self.config, + ) + + self.assertEqual(result["profiles"], ["global_etf_rotation"]) + + def test_rejects_wrong_domain_benchmark(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + archive = root / "artifact.zip" + self._write_valid_archive(archive) + replacement = root / "wrong-benchmark.zip" + with ( + zipfile.ZipFile(archive) as source, + zipfile.ZipFile(replacement, "w") as target, + ): + for member in source.infolist(): + raw = source.read(member) + if member.filename.endswith("portfolio_and_tracker_returns.csv"): + raw = raw.replace(b"buy_hold_SPY", b"buy_hold_BTC") + target.writestr(member, raw) + + with self.assertRaises(SYNC.LifecycleArtifactError): + SYNC.extract_validated_archive( + replacement, + root / "version", + self.config, + ) + + def test_rejects_traversal_and_symlink_members(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + for name, configure in ( + ("traversal", lambda bundle: bundle.writestr("../escape", "bad")), + ("symlink", self._write_symlink_member), + ): + archive = root / f"{name}.zip" + with zipfile.ZipFile(archive, "w") as bundle: + configure(bundle) + with self.assertRaises(SYNC.LifecycleArtifactError): + SYNC.extract_validated_archive( + archive, + root / f"{name}-out", + self.config, + ) + + @staticmethod + def _write_symlink_member(bundle: zipfile.ZipFile) -> None: + info = zipfile.ZipInfo( + "external/UsEquitySnapshotPipelines/data/output/example/" + "portfolio_and_tracker_returns.csv" + ) + info.create_system = 3 + info.external_attr = (stat.S_IFLNK | 0o777) << 16 + bundle.writestr(info, "/tmp/outside") + + def test_rejects_profile_mismatch(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + archive = root / "artifact.zip" + self._write_valid_archive(archive, matrix_profile="wrong_profile") + + with self.assertRaisesRegex( + SYNC.LifecycleArtifactError, + "profile", + ): + SYNC.extract_validated_archive( + archive, + root / "version", + self.config, + ) + + def test_atomically_switches_consumer_links(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + archive = root / "artifact.zip" + version = root / "versions" / "us_equity" / "2" + self._write_valid_archive(archive) + SYNC.extract_validated_archive(archive, version, self.config) + projects_root = root / "projects" + lifecycle_root = root / "store" + + SYNC.activate_version( + version, + self.config, + projects_root=projects_root, + lifecycle_root=lifecycle_root, + ) + + output_link = ( + projects_root + / "UsEquitySnapshotPipelines" + / "data" + / "output" + ) + backtest_link = lifecycle_root / "backtest" / "us_equity" + self.assertTrue(output_link.is_symlink()) + self.assertTrue(backtest_link.is_symlink()) + self.assertTrue(output_link.resolve().is_dir()) + self.assertTrue(backtest_link.resolve().is_dir()) + self.assertNotIn("..", os.readlink(output_link)) + + def test_detects_tampered_stored_version(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + archive = root / "artifact.zip" + version = root / "versions" / "us_equity" / "2" + self._write_valid_archive(archive) + validation = SYNC.extract_validated_archive( + archive, + version, + self.config, + ) + manifest = { + "profiles": validation["profiles"], + "sha256": validation["sha256"], + } + + SYNC.validate_stored_version(version, manifest, self.config) + matrix = next(version.rglob("portfolio_and_tracker_returns.csv")) + matrix.write_text("tampered", encoding="utf-8") + + with self.assertRaises(SYNC.LifecycleArtifactError): + SYNC.validate_stored_version(version, manifest, self.config) + + +if __name__ == "__main__": + unittest.main() diff --git a/ops/quant-monitor/tests/test_monitor_fail_closed.py b/ops/quant-monitor/tests/test_monitor_fail_closed.py index e2d3328b..58a0845c 100644 --- a/ops/quant-monitor/tests/test_monitor_fail_closed.py +++ b/ops/quant-monitor/tests/test_monitor_fail_closed.py @@ -52,6 +52,101 @@ def unavailable(_domain): ) self.assertNotIn("sensitive path", str(errors)) + def test_health_cycle_refreshes_snapshots_before_drift(self) -> None: + calls: list[tuple[str, str]] = [] + + snapshots, results, errors = HEALTH_CYCLE._refresh_and_collect_drift( + lambda domain: calls.append(("monitor", domain)) or [object()], + lambda domain: calls.append(("drift", domain)) or [domain], + domains=("us_equity", "crypto"), + ) + + self.assertEqual( + calls, + [ + ("monitor", "us_equity"), + ("drift", "us_equity"), + ("monitor", "crypto"), + ("drift", "crypto"), + ], + ) + self.assertEqual(set(snapshots), {"us_equity", "crypto"}) + self.assertEqual(results, {"us_equity": ["us_equity"], "crypto": ["crypto"]}) + self.assertEqual(errors, []) + + def test_health_cycle_skips_drift_when_snapshot_refresh_fails(self) -> None: + drift_calls: list[str] = [] + + snapshots, results, errors = HEALTH_CYCLE._refresh_and_collect_drift( + lambda _domain: (_ for _ in ()).throw(RuntimeError("sensitive details")), + lambda domain: drift_calls.append(domain) or [], + domains=("hk_equity",), + ) + + self.assertEqual(snapshots, {}) + self.assertEqual(results, {}) + self.assertEqual(drift_calls, []) + self.assertEqual( + errors, + [ + { + "domain": "hk_equity", + "code": "monitor_data_unavailable", + "error_type": "RuntimeError", + } + ], + ) + self.assertNotIn("sensitive details", str(errors)) + + def test_health_cycle_only_accepts_ready_artifact_domains(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + status_path = root / "data" / "lifecycle-artifacts" / "status.json" + status_path.parent.mkdir(parents=True) + status_path.write_text( + json.dumps( + { + "schema_version": "quant_monitor_lifecycle_artifact_status.v1", + "as_of": "2026-07-30T07:00:00+00:00", + "domains": { + "us_equity": { + "status": "ready", + "artifact_id": 1, + "run_id": 2, + "head_sha": "a" * 40, + "profiles": ["global_etf_rotation"], + }, + "crypto": { + "status": "error", + "code": "trusted_artifact_unavailable", + "error_type": "LifecycleArtifactError", + }, + }, + } + ), + encoding="utf-8", + ) + + ready, errors = HEALTH_CYCLE._load_lifecycle_artifact_status( + root, + domains=("us_equity", "crypto"), + now=HEALTH_CYCLE.datetime.fromisoformat( + "2026-07-30T07:30:00+00:00" + ), + ) + + self.assertEqual(ready, ("us_equity",)) + self.assertEqual( + errors, + [ + { + "domain": "crypto", + "code": "trusted_artifact_unavailable", + "error_type": "LifecycleArtifactError", + } + ], + ) + def test_daily_briefing_collects_drift_errors_without_aborting(self) -> None: def unavailable(_domain): raise RuntimeError("sensitive path must not escape")