timmy
179f8fe6e9
Quality gates / quality (pull_request) Successful in 13m25s
The ROADMAP gates (macro-F1 >= 0.80, precision@coverage >= 0.90, ECE <= 0.05) were prose with no tool behind them. scripts/calibration_report.py now computes them from prediction records and refuses input it cannot honestly score: - a split is re-derived from the pseudonymous subject key with the ingestion rule, so a corpus cannot re-label its own splits; a contributor spanning two splits is refused before scoring - the threshold is selected on validation only, and selecting on test or train is refused - coverage, precision-at-coverage, macro-F1, per-class precision/recall, confusion, and expected calibration error are computed on accepted suggestions, with classes that abstained away named in excludedClasses instead of being scored as a silent zero - exit codes carry the verdict: 0 GO, 3 BLOCKED, 2 refused The release demo records the real harness against committed synthetic fixtures: a blocked three-candidate comparison, a fixture engineered to clear every gate, and a leaked corpus refused. Every number on the card is read back from the report the run wrote, and the captions state that no model was measured, because a fixture is not evidence about accuracy. The harness suite is wired into CI, check:syntax, and the release builder's own gates.
307 lines
15 KiB
Python
307 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""Tests for the contributor-held-out evaluation harness.
|
||
|
||
Issue #29 asks for the authoritative comparison and release-gate harness: macro-F1, per-class
|
||
precision/recall, a confusion matrix, calibration error, coverage, and precision-at-coverage, over
|
||
splits that are contributor-held-out and reproducible, comparing candidate models.
|
||
|
||
Every fixture in this file is synthetic: fictional pseudonymous subject keys and invented
|
||
prediction scores. No real photo, label, or patient record is read or written.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
import unittest
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
HARNESS = ROOT / "scripts" / "calibration_report.py"
|
||
|
||
|
||
def load_harness():
|
||
spec = importlib.util.spec_from_file_location("calibration_report", HARNESS)
|
||
assert spec and spec.loader, "scripts/calibration_report.py must exist"
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
harness = load_harness()
|
||
|
||
|
||
def record(subject_key: str, truth: int, predicted: int | None, confidence: float, *, split: str | None = None, model: str = "frozen-encoder") -> dict:
|
||
return {
|
||
"model": model,
|
||
"subjectKey": subject_key,
|
||
"split": split or harness.expected_split(subject_key),
|
||
"truth": {"bristolType": truth},
|
||
"prediction": None if predicted is None else {"bristolType": predicted, "confidence": confidence},
|
||
}
|
||
|
||
|
||
class ComparisonTests(unittest.TestCase):
|
||
def models(self):
|
||
return {
|
||
"specialist-candidate": [
|
||
record("cd00000000000001", 4, 4, 1.0),
|
||
record("ce00000000000002", 3, 3, 1.0),
|
||
record("e600000000000001", 4, 4, 1.0),
|
||
record("e700000000000002", 6, 6, 1.0),
|
||
],
|
||
"vlm-bootstrap": [
|
||
record("d000000000000004", 4, 4, 1.0),
|
||
record("e600000000000003", 4, 2, 1.0),
|
||
record("e700000000000004", 6, 6, 1.0),
|
||
],
|
||
}
|
||
|
||
def test_measures_each_candidate_on_test_at_a_threshold_chosen_on_validation(self):
|
||
verdict = harness.compare(self.models())
|
||
self.assertEqual([row["model"] for row in verdict["models"]], ["specialist-candidate", "vlm-bootstrap"])
|
||
specialist = verdict["models"][0]
|
||
self.assertEqual(specialist["selectionSplit"], "validation")
|
||
self.assertEqual(specialist["threshold"], 0.0)
|
||
self.assertEqual(specialist["evaluatedSplit"], "test")
|
||
self.assertAlmostEqual(specialist["macroF1"], 1.0, places=6)
|
||
self.assertTrue(specialist["gate"]["passed"])
|
||
self.assertEqual(specialist["gate"]["failed"], [])
|
||
self.assertEqual(verdict["recommended"], "specialist-candidate")
|
||
self.assertEqual(verdict["verdict"], "GO")
|
||
|
||
def test_a_candidate_that_is_only_half_right_fails_the_macro_f1_precision_and_calibration_gates(self):
|
||
vlm = next(row for row in harness.compare(self.models())["models"] if row["model"] == "vlm-bootstrap")
|
||
self.assertAlmostEqual(vlm["precisionAtCoverage"], 0.5, places=6)
|
||
self.assertFalse(vlm["gate"]["passed"])
|
||
self.assertEqual(vlm["gate"]["failed"], ["macroF1", "precisionAtCoverage", "ece"])
|
||
|
||
def test_blocks_the_release_when_no_candidate_clears_every_gate(self):
|
||
models = {"vlm-bootstrap": self.models()["vlm-bootstrap"]}
|
||
verdict = harness.compare(models)
|
||
self.assertEqual(verdict["verdict"], "BLOCKED")
|
||
self.assertIsNone(verdict["recommended"])
|
||
self.assertEqual(verdict["gates"], {"macroF1": 0.80, "precisionAtCoverage": 0.90, "ece": 0.05})
|
||
|
||
def test_gate_limits_are_inclusive_so_a_candidate_exactly_on_the_line_passes(self):
|
||
records = [
|
||
record("cd00000000000001", 4, 4, 0.9),
|
||
record("e600000000000001", 4, 4, 0.9),
|
||
record("e700000000000002", 3, 2, 0.9),
|
||
]
|
||
verdict = harness.compare({"frozen-encoder": records}, gates={"macroF1": 0.0, "precisionAtCoverage": 0.5, "ece": 1.0})
|
||
row = verdict["models"][0]
|
||
self.assertAlmostEqual(row["precisionAtCoverage"], 0.5, places=6)
|
||
self.assertTrue(row["gate"]["passed"], row["gate"])
|
||
|
||
def test_refuses_a_candidate_with_no_held_out_predictions(self):
|
||
with self.assertRaisesRegex(harness.EvaluationError, "no test-split predictions"):
|
||
harness.compare({"frozen-encoder": [record("cd00000000000001", 4, 4, 1.0)]})
|
||
|
||
|
||
class CliTests(unittest.TestCase):
|
||
def write_predictions(self, directory: Path, name: str, rows: list[dict]) -> Path:
|
||
path = directory / name
|
||
path.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8")
|
||
return path
|
||
|
||
def run_cli(self, *args: str) -> subprocess.CompletedProcess:
|
||
return subprocess.run(
|
||
[sys.executable, str(HARNESS), *args], cwd=ROOT, capture_output=True, text=True
|
||
)
|
||
|
||
def test_reports_a_go_with_the_candidate_table_and_writes_the_json_report(self):
|
||
with tempfile.TemporaryDirectory() as raw:
|
||
directory = Path(raw)
|
||
predictions = self.write_predictions(directory, "predictions.jsonl", [
|
||
{**record("cd00000000000001", 4, 4, 1.0), "model": "specialist-candidate"},
|
||
{**record("e600000000000001", 4, 4, 1.0), "model": "specialist-candidate"},
|
||
{**record("e700000000000002", 6, 6, 1.0), "model": "specialist-candidate"},
|
||
])
|
||
output = directory / "report.json"
|
||
run = self.run_cli("--predictions", str(predictions), "--json", str(output), "--min-coverage", "0.5")
|
||
self.assertEqual(run.returncode, 0, run.stderr)
|
||
self.assertIn("verdict: GO", run.stdout)
|
||
self.assertIn("specialist-candidate", run.stdout)
|
||
self.assertIn("macroF1", run.stdout)
|
||
report = json.loads(output.read_text(encoding="utf-8"))
|
||
self.assertEqual(report["verdict"], "GO")
|
||
self.assertEqual(report["recommended"], "specialist-candidate")
|
||
self.assertEqual(report["models"][0]["evaluatedSplit"], "test")
|
||
self.assertEqual(report["models"][0]["selectionSplit"], "validation")
|
||
|
||
def test_exits_three_when_the_release_gate_blocks(self):
|
||
with tempfile.TemporaryDirectory() as raw:
|
||
directory = Path(raw)
|
||
predictions = self.write_predictions(directory, "predictions.jsonl", [
|
||
{**record("cd00000000000001", 4, 3, 1.0), "model": "vlm-bootstrap"},
|
||
{**record("e600000000000001", 4, 2, 1.0), "model": "vlm-bootstrap"},
|
||
])
|
||
run = self.run_cli("--predictions", str(predictions))
|
||
self.assertEqual(run.returncode, 3, run.stderr)
|
||
self.assertIn("verdict: BLOCKED", run.stdout)
|
||
|
||
|
||
def test_refuses_an_out_of_range_label_instead_of_scoring_it(self):
|
||
with tempfile.TemporaryDirectory() as raw:
|
||
directory = Path(raw)
|
||
output = directory / "report.json"
|
||
bad_truth = self.write_predictions(directory, "truth.jsonl", [
|
||
{**record("cd00000000000001", 9, 9, 1.0), "model": "frozen-encoder"},
|
||
])
|
||
run = self.run_cli("--predictions", str(bad_truth), "--json", str(output))
|
||
self.assertEqual(run.returncode, 2)
|
||
self.assertIn("outside Bristol 1-7", run.stderr)
|
||
self.assertFalse(output.exists(), "a refused run must not leave a report behind")
|
||
bad_prediction = self.write_predictions(directory, "prediction.jsonl", [
|
||
{**record("e600000000000001", 4, 9, 1.0), "model": "frozen-encoder"},
|
||
])
|
||
run = self.run_cli("--predictions", str(bad_prediction))
|
||
self.assertEqual(run.returncode, 2)
|
||
self.assertIn("predicted bristolType must be 1-7", run.stderr)
|
||
|
||
def test_refuses_a_corpus_that_leaks_a_contributor_across_splits(self):
|
||
with tempfile.TemporaryDirectory() as raw:
|
||
directory = Path(raw)
|
||
predictions = self.write_predictions(directory, "predictions.jsonl", [
|
||
{**record("cd00000000000001", 4, 4, 1.0, split="train"), "model": "frozen-encoder"},
|
||
{**record("cd00000000000001", 4, 4, 1.0, split="validation"), "model": "frozen-encoder"},
|
||
{**record("e600000000000001", 4, 4, 1.0), "model": "frozen-encoder"},
|
||
])
|
||
run = self.run_cli("--predictions", str(predictions))
|
||
self.assertEqual(run.returncode, 2)
|
||
self.assertIn("leaks across", run.stderr)
|
||
|
||
|
||
class ThresholdSelectionTests(unittest.TestCase):
|
||
def validation_records(self):
|
||
# 0xcd–0xe5 bucket to validation under the ingestion rule. Confident answers are right and
|
||
# the least confident one is wrong, which is what a threshold is supposed to exploit.
|
||
return [
|
||
record("cd00000000000001", 4, 4, 0.90),
|
||
record("ce00000000000002", 4, 4, 0.80),
|
||
record("cf00000000000003", 3, 3, 0.60),
|
||
record("d000000000000004", 7, 4, 0.40),
|
||
]
|
||
|
||
def test_picks_the_most_precise_threshold_that_still_clears_the_coverage_floor(self):
|
||
records = self.validation_records()
|
||
# At a 0.5 floor, 0.60 and 0.80 are both perfect; the wider cut wins, so 0.60.
|
||
floored = harness.select_threshold(records, min_coverage=0.5)
|
||
self.assertEqual(floored["threshold"], 0.6)
|
||
self.assertEqual(floored["counts"], {"total": 4, "accepted": 3, "abstained": 1})
|
||
self.assertAlmostEqual(floored["precisionAtCoverage"], 1.0, places=6)
|
||
self.assertAlmostEqual(floored["coverage"], 0.75, places=6)
|
||
self.assertEqual(floored["selectionSplit"], "validation")
|
||
# A 0.8 floor forbids that cut, so the harness falls back to the widest honest threshold
|
||
# instead of silently returning a threshold that violates the floor.
|
||
strict = harness.select_threshold(records, min_coverage=0.8)
|
||
self.assertEqual(strict["threshold"], 0.0)
|
||
self.assertAlmostEqual(strict["precisionAtCoverage"], 0.75, places=6)
|
||
|
||
def test_refuses_an_unreachable_coverage_floor(self):
|
||
with self.assertRaisesRegex(harness.EvaluationError, "coverage floor"):
|
||
harness.select_threshold(self.validation_records(), min_coverage=1.5)
|
||
|
||
def test_refuses_to_choose_a_threshold_on_test_or_train_data(self):
|
||
test_only = [record("e700000000000001", 4, 4, 0.9)]
|
||
with self.assertRaisesRegex(harness.EvaluationError, "validation split only"):
|
||
harness.select_threshold(test_only, min_coverage=0.5)
|
||
train_mixed = [record("aa00000000000001", 4, 4, 0.9), record("cd00000000000001", 3, 3, 0.8)]
|
||
with self.assertRaisesRegex(harness.EvaluationError, "validation split only"):
|
||
harness.select_threshold(train_mixed, min_coverage=0.5)
|
||
|
||
|
||
class ContributorHeldOutTests(unittest.TestCase):
|
||
def test_refuses_a_split_that_cannot_be_derived_from_the_subject_key(self):
|
||
# 0xaa = 170, below the train/validation edge, so this subject can only ever be train.
|
||
records = [record("aa00000000000001", 4, 4, 0.9, split="test")]
|
||
with self.assertRaisesRegex(harness.EvaluationError, "not reproducible"):
|
||
harness.assert_contributor_held_out(records)
|
||
|
||
|
||
def test_refuses_a_contributor_whose_photos_span_two_splits(self):
|
||
records = [
|
||
record("aa00000000000001", 4, 4, 0.9, split="train"),
|
||
record("aa00000000000001", 3, 3, 0.8, split="validation"),
|
||
]
|
||
with self.assertRaisesRegex(harness.EvaluationError, "leaks across"):
|
||
harness.assert_contributor_held_out(records)
|
||
|
||
def test_allows_one_contributor_to_supply_several_photos_in_the_same_split(self):
|
||
records = [
|
||
record("aa00000000000001", 4, 4, 0.9),
|
||
record("aa00000000000001", 3, 3, 0.8),
|
||
]
|
||
harness.assert_contributor_held_out(records) # several photos, one person, one split: fine
|
||
|
||
|
||
class CalibrationTests(unittest.TestCase):
|
||
def test_calibration_error_weights_each_confidence_bin_by_its_share_of_accepted_suggestions(self):
|
||
records = [
|
||
record("aa00000000000001", 4, 4, 0.95),
|
||
record("ab00000000000002", 4, 4, 0.90),
|
||
record("ac00000000000003", 3, 3, 0.55),
|
||
record("ad00000000000004", 7, 7, 0.70),
|
||
]
|
||
# Bins [0.9,1.0) hold both 0.95 and 0.90: mean confidence 0.925, accuracy 1.0, gap 0.075.
|
||
self.assertAlmostEqual(harness.evaluate(records, threshold=0.0)["ece"], 0.225, places=6)
|
||
# Raising the threshold to 0.6 drops the 0.55 sample, so it can no longer pull calibration.
|
||
self.assertAlmostEqual(harness.evaluate(records, threshold=0.6)["ece"], 0.15, places=6)
|
||
|
||
|
||
class AbstentionTests(unittest.TestCase):
|
||
def test_low_confidence_and_explicit_abstentions_shrink_coverage_but_never_count_as_errors(self):
|
||
records = [
|
||
record("aa00000000000001", 4, 4, 0.90),
|
||
record("ab00000000000002", 4, 3, 0.55),
|
||
record("ac00000000000003", 3, 3, 0.40),
|
||
record("ad00000000000004", 7, None, 0.0),
|
||
]
|
||
report = harness.evaluate(records, threshold=0.6)
|
||
self.assertEqual(report["counts"], {"total": 4, "accepted": 1, "abstained": 3})
|
||
self.assertAlmostEqual(report["coverage"], 0.25, places=6)
|
||
self.assertAlmostEqual(report["precisionAtCoverage"], 1.0, places=6)
|
||
# Only the accepted suggestion is scored; the class that abstained away is named, not hidden.
|
||
self.assertEqual(sorted(report["perClass"]), [4])
|
||
self.assertAlmostEqual(report["macroF1"], 1.0, places=6)
|
||
self.assertEqual(report["excludedClasses"], [3, 7])
|
||
self.assertEqual(report["abstentions"], {"low-confidence": 2, "no-suggestion": 1})
|
||
|
||
def test_a_threshold_above_every_confidence_abstains_on_everything_without_crashing(self):
|
||
records = [
|
||
record("aa00000000000001", 4, 4, 0.30),
|
||
record("ab00000000000002", 2, 2, 0.50),
|
||
]
|
||
report = harness.evaluate(records, threshold=0.95)
|
||
self.assertEqual(report["counts"], {"total": 2, "accepted": 0, "abstained": 2})
|
||
self.assertAlmostEqual(report["coverage"], 0.0, places=6)
|
||
self.assertAlmostEqual(report["macroF1"], 0.0, places=6)
|
||
self.assertEqual(report["perClass"], {})
|
||
|
||
|
||
class MacroF1Tests(unittest.TestCase):
|
||
def test_scores_each_bristol_class_and_averages_only_the_classes_present_in_truth(self):
|
||
records = [
|
||
record("aa00000000000001", 4, 4, 0.90),
|
||
record("ab00000000000002", 4, 3, 0.80),
|
||
record("ac00000000000003", 3, 3, 0.70),
|
||
record("ad00000000000004", 7, 7, 0.60),
|
||
]
|
||
report = harness.evaluate(records, threshold=0.0)
|
||
self.assertEqual(sorted(report["perClass"]), [3, 4, 7])
|
||
self.assertAlmostEqual(report["perClass"][4]["precision"], 1.0, places=6)
|
||
self.assertAlmostEqual(report["perClass"][4]["recall"], 0.5, places=6)
|
||
self.assertAlmostEqual(report["perClass"][4]["f1"], 0.666667, places=6)
|
||
self.assertAlmostEqual(report["perClass"][3]["precision"], 0.5, places=6)
|
||
self.assertAlmostEqual(report["perClass"][3]["recall"], 1.0, places=6)
|
||
self.assertAlmostEqual(report["perClass"][7]["f1"], 1.0, places=6)
|
||
self.assertAlmostEqual(report["macroF1"], 0.777778, places=6)
|
||
self.assertEqual(report["confusion"], {"4->4": 1, "4->3": 1, "3->3": 1, "7->7": 1})
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|