backend/scripts/eval_matching.py

"""Matching evaluation harness for threshold tuning (spec §9.6).

Given a labeled set of photo pairs (same-dog vs different-dog), report ROC/AUC and
precision/recall at candidate thresholds. Use this to tune REVIEW_THRESHOLD / STRONG_THRESHOLD
and to compare embedders. Works with whatever Embedder is configured (mock by default).

Input CSV format (header required):
    image_a,image_b,same
    /path/1.jpg,/path/2.jpg,1
    /path/1.jpg,/path/3.jpg,0

Usage:
    python -m scripts.eval_matching pairs.csv
"""
from __future__ import annotations

import csv
import sys

import numpy as np

from app.ml import get_embedder


def _load_pairs(path: str) -> list[tuple[str, str, int]]:
    out = []
    with open(path, newline="", encoding="utf-8") as fh:
        for row in csv.DictReader(fh):
            out.append((row["image_a"], row["image_b"], int(row["same"])))
    return out


def _cosine(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b))  # embeddings are L2-normalized


def evaluate(pairs: list[tuple[str, str, int]]) -> dict:
    embedder = get_embedder()
    scores: list[float] = []
    labels: list[int] = []
    for a, b, same in pairs:
        va, vb = embedder.embed([a, b])
        scores.append(_cosine(va, vb))
        labels.append(same)

    scores_arr = np.array(scores)
    labels_arr = np.array(labels)

    # Sweep thresholds; compute precision/recall and a trapezoidal ROC-AUC.
    thresholds = np.linspace(0, 1, 101)
    rows = []
    tpr_list, fpr_list = [], []
    pos = max(int(labels_arr.sum()), 1)
    neg = max(int((1 - labels_arr).sum()), 1)
    for t in thresholds:
        pred = scores_arr >= t
        tp = int(((pred == 1) & (labels_arr == 1)).sum())
        fp = int(((pred == 1) & (labels_arr == 0)).sum())
        fn = int(((pred == 0) & (labels_arr == 1)).sum())
        precision = tp / (tp + fp) if (tp + fp) else 1.0
        recall = tp / (tp + fn) if (tp + fn) else 0.0
        rows.append((float(t), precision, recall))
        tpr_list.append(tp / pos)
        fpr_list.append(fp / neg)

    # AUC via trapezoid over sorted FPR.
    order = np.argsort(fpr_list)
    auc = float(np.trapz(np.array(tpr_list)[order], np.array(fpr_list)[order]))
    return {"auc": abs(auc), "embedder": embedder.name, "rows": rows}


def main() -> None:
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)
    pairs = _load_pairs(sys.argv[1])
    result = evaluate(pairs)
    print(f"Embedder: {result['embedder']}  ROC-AUC: {result['auc']:.3f}")
    print(f"{'thresh':>7} {'precision':>10} {'recall':>8}")
    for t, p, r in result["rows"][::10]:  # every 0.1
        print(f"{t:7.2f} {p:10.3f} {r:8.3f}")
    print("\nTune REVIEW_THRESHOLD / STRONG_THRESHOLD from these curves; record findings in DECISIONS.md.")


if __name__ == "__main__":
    main()