ml/scripts/quality_gate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
quality_gate.py — decide whether a candidate model bundle may replace the
baseline (deploy/latest). A regression here costs real money: false positives
pay unearned TrashCoin, false negatives block honest users into needs_review.

Compares model_card.json metrics:
  - ultralytics "metrics/mAP50(B)"      (detection quality)
  - evaluation.macro_accuracy           (classification quality)
  - evaluation.per_class_f1             (no class may silently collapse)

Rules (all configurable):
  1. candidate mAP50   >= baseline mAP50   - max_map_drop
  2. candidate macro_accuracy >= baseline - max_acc_drop
  3. for every class with baseline F1 > f1_floor: candidate F1 >= baseline F1 - max_f1_drop
     (classes at ~0 in the baseline, e.g. organic/ewaste today, never block)

Exit code 0 = PASS (writes gate_report.json into the candidate dir), 1 = FAIL.

Usage:
  python ml/scripts/quality_gate.py --candidate artifacts/v2026xx --baseline deploy/latest
"""
from __future__ import annotations

import argparse
import hashlib
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple


def model_sha256(bundle_dir: Path) -> Optional[str]:
    """Hash of the model the metrics belong to — binds gate_report.json to
    exactly this model.onnx so a stale report can never promote a newer,
    never-gated model."""
    p = Path(bundle_dir) / "model.onnx"
    if not p.exists():
        return None
    return hashlib.sha256(p.read_bytes()).hexdigest()


def load_metrics(bundle_dir: Path) -> Dict[str, Any]:
    card_path = Path(bundle_dir) / "model_card.json"
    card = json.loads(card_path.read_text(encoding="utf-8"))
    metrics = card.get("metrics") or {}
    ultra = metrics.get("ultralytics") or {}
    ev = metrics.get("evaluation") or {}
    return {
        "map50": ultra.get("metrics/mAP50(B)"),
        "macro_accuracy": ev.get("macro_accuracy"),
        "per_class_f1": ev.get("per_class_f1") or {},
    }


def evaluate_gate(
    candidate: Dict[str, Any],
    baseline: Dict[str, Any],
    max_map_drop: float = 0.02,
    max_acc_drop: float = 0.02,
    max_f1_drop: float = 0.05,
    f1_floor: float = 0.05,
) -> Tuple[bool, List[str]]:
    """Pure gate logic. Returns (passed, list of failure reasons)."""
    reasons: List[str] = []

    def _cmp(name: str, cand: Optional[float], base: Optional[float], max_drop: float):
        if base is None:
            return  # nothing to regress against
        if cand is None:
            reasons.append(f"{name}: candidate metric missing (baseline={base:.4f})")
            return
        if float(cand) < float(base) - max_drop:
            reasons.append(f"{name}: {cand:.4f} < baseline {base:.4f} - allowed drop {max_drop}")

    _cmp("mAP50", candidate.get("map50"), baseline.get("map50"), max_map_drop)
    _cmp("macro_accuracy", candidate.get("macro_accuracy"), baseline.get("macro_accuracy"), max_acc_drop)

    base_f1 = baseline.get("per_class_f1") or {}
    cand_f1 = candidate.get("per_class_f1") or {}
    for cls, bf1 in base_f1.items():
        try:
            bf1 = float(bf1)
        except (TypeError, ValueError):
            continue
        if bf1 <= f1_floor:
            continue  # dead classes in the baseline never block (organic/ewaste today)
        cf1 = cand_f1.get(cls)
        if cf1 is None:
            reasons.append(f"per_class_f1[{cls}]: missing in candidate (baseline={bf1:.4f})")
            continue
        if float(cf1) < bf1 - max_f1_drop:
            reasons.append(f"per_class_f1[{cls}]: {float(cf1):.4f} < baseline {bf1:.4f} - allowed drop {max_f1_drop}")

    return (len(reasons) == 0), reasons


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description="Quality gate: candidate bundle vs baseline bundle.")
    ap.add_argument("--candidate", required=True, type=Path)
    ap.add_argument("--baseline", default="deploy/latest", type=Path)
    ap.add_argument("--max-map-drop", type=float, default=0.02)
    ap.add_argument("--max-acc-drop", type=float, default=0.02)
    ap.add_argument("--max-f1-drop", type=float, default=0.05)
    ap.add_argument("--f1-floor", type=float, default=0.05)
    ap.add_argument("--allow-missing-baseline", action="store_true",
                    help="explicitly accept a candidate with no baseline (first model ever). "
                         "Without this flag an unreadable baseline FAILS the gate — "
                         "a wrong cwd or corrupt model_card must never disable the gate.")
    args = ap.parse_args(argv)

    try:
        cand = load_metrics(args.candidate)
    except Exception as e:
        print(f"GATE FAIL: cannot read candidate metrics: {e}")
        return 1
    try:
        base = load_metrics(args.baseline)
    except Exception as e:
        if not args.allow_missing_baseline:
            print(f"GATE FAIL: cannot read baseline metrics from '{args.baseline}' ({e}). "
                  "If this really is the first model ever, re-run with --allow-missing-baseline; "
                  "otherwise fix the baseline path (are you running from the repo root?).")
            return 1
        print(f"GATE WARN: no baseline metrics ({e}) — accepting candidate as first model "
              "(--allow-missing-baseline).")
        base = {"map50": None, "macro_accuracy": None, "per_class_f1": {}}

    passed, reasons = evaluate_gate(
        cand, base,
        max_map_drop=args.max_map_drop,
        max_acc_drop=args.max_acc_drop,
        max_f1_drop=args.max_f1_drop,
        f1_floor=args.f1_floor,
    )

    report = {
        "passed": passed,
        "candidate": str(args.candidate),
        "baseline": str(args.baseline),
        "model_sha256": model_sha256(args.candidate),
        "candidate_metrics": cand,
        "baseline_metrics": base,
        "reasons": reasons,
        "thresholds": {
            "max_map_drop": args.max_map_drop,
            "max_acc_drop": args.max_acc_drop,
            "max_f1_drop": args.max_f1_drop,
            "f1_floor": args.f1_floor,
        },
    }
    out = Path(args.candidate) / "gate_report.json"
    out.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")

    if passed:
        print(f"GATE PASS — report: {out}")
        return 0
    print("GATE FAIL:")
    for r in reasons:
        print(f"  - {r}")
    print(f"Report: {out}")
    return 1


if __name__ == "__main__":
    sys.exit(main())