#!/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())