scripts/close_loop_validate.py

#!/usr/bin/env python3
"""Closed-loop validation registry and promotion gate.

The gate is intentionally small and importable: production publishability can
check the promoted feature version without running market-data backtests.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from models.predictor import FEATURE_VERSION

REGISTRY_PATH = ROOT / os.getenv("VALIDATION_REGISTRY_PATH", "docs/validation_registry.json")
VALIDATION_RUNS_DIR = ROOT / "docs" / "validation_runs"


@dataclass(frozen=True)
class GateThresholds:
    min_accuracy_delta_pp: float = 5.0
    min_direction_accuracy_delta_pp: float = -2.0
    min_signal_count_ratio: float = 0.70
    min_buy_signal_win_rate_delta_pp: float = 0.0
    max_severe_data_warnings: int = 0


@dataclass(frozen=True)
class GateConfig:
    min_accuracy_delta_pp: float = 5.0
    min_direction_delta_pp: float = -2.0
    min_signal_count_ratio: float = 0.70
    min_buy_signal_win_rate_delta_pp: float = 0.0
    max_severe_data_warnings: int = 0

    def to_thresholds(self) -> GateThresholds:
        return GateThresholds(
            min_accuracy_delta_pp=self.min_accuracy_delta_pp,
            min_direction_accuracy_delta_pp=self.min_direction_delta_pp,
            min_signal_count_ratio=self.min_signal_count_ratio,
            min_buy_signal_win_rate_delta_pp=self.min_buy_signal_win_rate_delta_pp,
            max_severe_data_warnings=self.max_severe_data_warnings,
        )


@dataclass(frozen=True)
class ValidationMetrics:
    accuracy: float = 0.0
    direction_accuracy: float = 0.0
    signal_count: float = 0.0
    buy_signal_win_rate: float = 0.0
    severe_data_warnings: int = 0


@dataclass(frozen=True)
class PromotionDecision:
    passed: bool
    failed_checks: list[str]
    deltas: dict[str, float]
    checks: dict[str, bool]


DEFAULT_GATE = GateThresholds()
NO_PROMOTED_FEATURE_VERSION = ""


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _registry_path(path: Path | str | None = None) -> Path:
    return Path(path) if path is not None else REGISTRY_PATH


def _display_path(path: Path) -> str:
    try:
        return str(path.relative_to(ROOT))
    except ValueError:
        return str(path)


def _metric(metrics: dict[str, Any], *names: str, default: float = 0.0) -> float:
    for name in names:
        value = metrics.get(name)
        if value is not None:
            try:
                return float(value)
            except (TypeError, ValueError):
                return default
    return default


def _percent(value: float) -> float:
    """Normalize likely ratios to percentages while preserving percentage inputs."""
    if -1.0 <= value <= 1.0:
        return value * 100.0
    return value


def normalize_metrics(metrics: dict[str, Any] | None) -> dict[str, float]:
    """Extract the small metric set used by the promotion gate."""
    if isinstance(metrics, ValidationMetrics):
        metrics = asdict(metrics)
    metrics = metrics or {}
    stats = metrics.get("stats") if isinstance(metrics.get("stats"), dict) else {}
    merged = {**stats, **metrics}
    return {
        "accuracy": _percent(_metric(merged, "accuracy", "signal_accuracy_overall")),
        "direction_accuracy": _percent(
            _metric(merged, "direction_accuracy", "directional_accuracy", "signal_accuracy_overall")
        ),
        "buy_signal_win_rate": _percent(
            _metric(merged, "buy_signal_win_rate", "buy_win_rate", "up_precision")
        ),
        "signal_count": _metric(merged, "signal_count", "total_signals", "total_days"),
        "severe_data_warnings": _metric(merged, "severe_data_warnings", default=0.0),
    }


def evaluate_gate(
    baseline_metrics: dict[str, Any] | None,
    candidate_metrics: dict[str, Any] | None,
    *,
    thresholds: GateThresholds = DEFAULT_GATE,
) -> dict[str, Any]:
    """Compare candidate metrics against baseline and return pass/fail details."""
    baseline = normalize_metrics(baseline_metrics)
    candidate = normalize_metrics(candidate_metrics)

    accuracy_delta_pp = candidate["accuracy"] - baseline["accuracy"]
    direction_accuracy_delta_pp = candidate["direction_accuracy"] - baseline["direction_accuracy"]
    buy_signal_win_rate_delta_pp = candidate["buy_signal_win_rate"] - baseline["buy_signal_win_rate"]
    min_signal_count = baseline["signal_count"] * thresholds.min_signal_count_ratio

    checks = {
        "accuracy_delta": accuracy_delta_pp >= thresholds.min_accuracy_delta_pp,
        "direction_accuracy_delta": (
            direction_accuracy_delta_pp >= thresholds.min_direction_accuracy_delta_pp
        ),
        "signal_count": candidate["signal_count"] >= min_signal_count,
        "buy_signal_win_rate_delta": (
            buy_signal_win_rate_delta_pp >= thresholds.min_buy_signal_win_rate_delta_pp
        ),
        "severe_data_warnings": (
            candidate["severe_data_warnings"] <= thresholds.max_severe_data_warnings
        ),
    }

    return {
        "passed": all(checks.values()),
        "checks": checks,
        "baseline": baseline,
        "candidate": candidate,
        "deltas": {
            "accuracy_delta_pp": round(accuracy_delta_pp, 4),
            "direction_accuracy_delta_pp": round(direction_accuracy_delta_pp, 4),
            "buy_signal_win_rate_delta_pp": round(buy_signal_win_rate_delta_pp, 4),
            "signal_count_delta": candidate["signal_count"] - baseline["signal_count"],
        },
        "thresholds": asdict(thresholds),
    }


def evaluate_promotion_gate(
    baseline: ValidationMetrics | dict[str, Any],
    candidate: ValidationMetrics | dict[str, Any],
    config: GateConfig | None = None,
) -> PromotionDecision:
    """Compatibility wrapper used by unit tests and future callers."""
    gate_result = evaluate_gate(
        asdict(baseline) if isinstance(baseline, ValidationMetrics) else baseline,
        asdict(candidate) if isinstance(candidate, ValidationMetrics) else candidate,
        thresholds=(config or GateConfig()).to_thresholds(),
    )
    failed_checks = [name for name, ok in gate_result["checks"].items() if not ok]
    check_name_map = {
        "accuracy_delta": "accuracy_delta_pp",
        "direction_accuracy_delta": "direction_accuracy_delta_pp",
        "signal_count": "signal_count_ratio",
        "buy_signal_win_rate_delta": "buy_signal_win_rate_delta_pp",
        "severe_data_warnings": "severe_data_warnings",
    }
    failed_checks = [check_name_map.get(name, name) for name in failed_checks]
    return PromotionDecision(
        passed=bool(gate_result["passed"]),
        failed_checks=failed_checks,
        deltas=gate_result["deltas"],
        checks=gate_result["checks"],
    )


def default_registry(*, feature_version: str = FEATURE_VERSION) -> dict[str, Any]:
    now = _now_iso()
    return {
        "schema_version": 1,
        "updated_at": now,
        "promotion_gate": asdict(DEFAULT_GATE),
        "promoted": {
            "feature_version": feature_version,
            "experiment_id": "current_production",
            "status": "pass",
            "promoted_at": now,
            "validation_run_path": None,
            "notes": "Initial registry entry for the current production feature version.",
        },
        "latest_candidate": None,
    }


def load_validation_registry(
    path: Path | str | None = None,
    *,
    allow_default: bool = True,
) -> dict[str, Any] | None:
    path = _registry_path(path)
    if not path.exists():
        return default_registry() if allow_default else None
    try:
        data = json.loads(path.read_text())
    except Exception:
        return default_registry() if allow_default else None
    if not isinstance(data, dict):
        return default_registry() if allow_default else None
    data.setdefault("schema_version", 1)
    data.setdefault("promotion_gate", asdict(DEFAULT_GATE))
    data.setdefault("promoted", default_registry()["promoted"])
    data.setdefault("latest_candidate", None)
    return data


class ValidationRegistry:
    """Small registry facade for promoted feature-version checks."""

    def __init__(self, path: Path | str | None = None) -> None:
        self.path = _registry_path(path)

    def load(self) -> dict[str, Any]:
        return load_validation_registry(self.path, allow_default=False) or {}

    def get_promoted_feature_version(self) -> str:
        registry = self.load()
        legacy = registry.get("promoted_feature_version")
        if legacy:
            return str(legacy)
        promoted = registry.get("promoted") if isinstance(registry.get("promoted"), dict) else {}
        if promoted.get("status") != "pass":
            return NO_PROMOTED_FEATURE_VERSION
        return str(promoted.get("feature_version") or NO_PROMOTED_FEATURE_VERSION)

    def is_promoted_feature_version(self, feature_version: str | None) -> bool:
        return bool(feature_version) and str(feature_version) == self.get_promoted_feature_version()

    def assert_promoted_feature_version(self, feature_version: str | None) -> None:
        if not self.is_promoted_feature_version(feature_version):
            raise ValueError(
                f"feature_version {feature_version!r} is not promoted "
                f"(promoted={self.get_promoted_feature_version()!r})"
            )


def save_validation_registry(registry: dict[str, Any], path: Path | str | None = None) -> None:
    path = _registry_path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    registry["updated_at"] = _now_iso()
    path.write_text(json.dumps(registry, indent=2, ensure_ascii=False))


def get_promoted_feature_version(path: Path | str | None = None) -> str:
    return ValidationRegistry(path).get_promoted_feature_version()


def is_feature_version_promoted(feature_version: str | None, path: Path | str | None = None) -> bool:
    return ValidationRegistry(path).is_promoted_feature_version(feature_version)


def _load_metrics_file(path: str) -> dict[str, Any]:
    data = json.loads(Path(path).read_text())
    if isinstance(data, dict) and isinstance(data.get("stats"), dict):
        return data["stats"]
    if isinstance(data, dict):
        return data
    raise ValueError(f"metrics file must contain a JSON object: {path}")


def _gate_thresholds_from_registry(registry: dict[str, Any] | None) -> GateThresholds:
    gate = registry.get("promotion_gate") if isinstance(registry, dict) else {}
    if not isinstance(gate, dict):
        gate = {}
    return GateThresholds(
        min_accuracy_delta_pp=float(
            gate.get("min_accuracy_delta_pp", DEFAULT_GATE.min_accuracy_delta_pp)
        ),
        min_direction_accuracy_delta_pp=float(
            gate.get(
                "min_direction_accuracy_delta_pp",
                DEFAULT_GATE.min_direction_accuracy_delta_pp,
            )
        ),
        min_signal_count_ratio=float(gate.get("min_signal_count_ratio", DEFAULT_GATE.min_signal_count_ratio)),
        min_buy_signal_win_rate_delta_pp=float(
            gate.get(
                "min_buy_signal_win_rate_delta_pp",
                DEFAULT_GATE.min_buy_signal_win_rate_delta_pp,
            )
        ),
        max_severe_data_warnings=int(
            gate.get("max_severe_data_warnings", DEFAULT_GATE.max_severe_data_warnings)
        ),
    )


def write_validation_run(
    *,
    experiment_id: str,
    feature_version: str,
    baseline_metrics: dict[str, Any],
    candidate_metrics: dict[str, Any],
    result: dict[str, Any],
    output_dir: Path | str = VALIDATION_RUNS_DIR,
) -> Path:
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    path = output_dir / f"{timestamp}_{experiment_id}.json"
    payload = {
        "experiment_id": experiment_id,
        "feature_version": feature_version,
        "created_at": _now_iso(),
        "baseline_metrics": baseline_metrics,
        "candidate_metrics": candidate_metrics,
        "gate_result": result,
    }
    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False))
    return path


def update_registry_with_validation(
    *,
    experiment_id: str,
    feature_version: str,
    gate_result: dict[str, Any],
    validation_run_path: Path | str | None = None,
    registry_path: Path | str = REGISTRY_PATH,
    promote_on_pass: bool = True,
    baseline_metrics: dict[str, Any] | None = None,
    candidate_metrics: dict[str, Any] | None = None,
) -> dict[str, Any]:
    registry = load_validation_registry(registry_path, allow_default=True) or default_registry()
    status = "pass" if gate_result.get("passed") else "fail"
    candidate = {
        "feature_version": feature_version,
        "experiment_id": experiment_id,
        "status": status,
        "validated_at": _now_iso(),
        "validation_run_path": str(validation_run_path) if validation_run_path else None,
        "deltas": gate_result.get("deltas", {}),
        "checks": gate_result.get("checks", {}),
    }
    if candidate_metrics is not None:
        candidate["metrics"] = candidate_metrics
    if baseline_metrics is not None:
        candidate["baseline_metrics"] = baseline_metrics
    registry["latest_candidate"] = candidate
    if promote_on_pass and status == "pass":
        registry["promoted"] = {
            **candidate,
            "promoted_at": _now_iso(),
        }
    save_validation_registry(registry, registry_path)
    return registry


def main() -> int:
    parser = argparse.ArgumentParser(description="Evaluate candidate metrics against the promotion gate.")
    parser.add_argument("--experiment", required=True, help="Experiment ID, for example C22_label_gate")
    parser.add_argument("--feature-version", default=FEATURE_VERSION)
    parser.add_argument("--baseline-json", required=True, help="Baseline metrics JSON")
    parser.add_argument("--candidate-json", required=True, help="Candidate metrics JSON")
    parser.add_argument("--registry", default=str(REGISTRY_PATH))
    parser.add_argument("--runs-dir", default=str(VALIDATION_RUNS_DIR), help="Directory for validation run JSON")
    parser.add_argument("--no-promote", action="store_true", help="Write result but do not promote on pass")
    parser.add_argument("--min-accuracy-delta-pp", type=float, default=None)
    parser.add_argument(
        "--min-direction-accuracy-delta-pp",
        type=float,
        default=None,
    )
    parser.add_argument("--min-signal-count-ratio", type=float, default=None)
    parser.add_argument(
        "--min-buy-signal-win-rate-delta-pp",
        type=float,
        default=None,
    )
    parser.add_argument("--max-severe-data-warnings", type=int, default=None)
    args = parser.parse_args()

    registry = load_validation_registry(args.registry, allow_default=True)
    registry_thresholds = _gate_thresholds_from_registry(registry)
    thresholds = GateThresholds(
        min_accuracy_delta_pp=(
            args.min_accuracy_delta_pp
            if args.min_accuracy_delta_pp is not None
            else registry_thresholds.min_accuracy_delta_pp
        ),
        min_direction_accuracy_delta_pp=(
            args.min_direction_accuracy_delta_pp
            if args.min_direction_accuracy_delta_pp is not None
            else registry_thresholds.min_direction_accuracy_delta_pp
        ),
        min_signal_count_ratio=(
            args.min_signal_count_ratio
            if args.min_signal_count_ratio is not None
            else registry_thresholds.min_signal_count_ratio
        ),
        min_buy_signal_win_rate_delta_pp=(
            args.min_buy_signal_win_rate_delta_pp
            if args.min_buy_signal_win_rate_delta_pp is not None
            else registry_thresholds.min_buy_signal_win_rate_delta_pp
        ),
        max_severe_data_warnings=(
            args.max_severe_data_warnings
            if args.max_severe_data_warnings is not None
            else registry_thresholds.max_severe_data_warnings
        ),
    )
    baseline = _load_metrics_file(args.baseline_json)
    candidate = _load_metrics_file(args.candidate_json)
    gate_result = evaluate_gate(baseline, candidate, thresholds=thresholds)
    run_path = write_validation_run(
        experiment_id=args.experiment,
        feature_version=args.feature_version,
        baseline_metrics=baseline,
        candidate_metrics=candidate,
        result=gate_result,
        output_dir=args.runs_dir,
    )
    update_registry_with_validation(
        experiment_id=args.experiment,
        feature_version=args.feature_version,
        gate_result=gate_result,
        validation_run_path=_display_path(run_path),
        registry_path=args.registry,
        promote_on_pass=not args.no_promote,
    )
    print(json.dumps({"validation_run": str(run_path), **gate_result}, indent=2, ensure_ascii=False))
    return 0 if gate_result["passed"] else 2


if __name__ == "__main__":
    raise SystemExit(main())