""" Compute v1-vs-v3 metrics on expanded corpus (n=28). Uses labels_curator_v3.json + labels_llm_v3.json as labelers, labels_detector_v1_v3corpus.json as v1 baseline, and labels_detector_v3.json as v3 candidate. Tests pre-registered hypotheses from DESIGN_v2.md ยง2: H-A1: FVS-012 F1 >= 0.35 H-A2: FVS-016 F1 >= 0.35 H-A3: FVS-008 F1 >= 0.35 H-A4: FVS-015 F1 >= 0.35 H-A5: macro-F1 >= 0.4 (H3 threshold from DESIGN.md) H-A7: macro-recall on curator-flagged >= 0.5 Writes results/results_v3.json + human-readable summary. """ from __future__ import annotations import json from collections import defaultdict from pathlib import Path from statistics import mean REPO_ROOT = Path(__file__).resolve().parents[2] STUDY_DIR = REPO_ROOT / "fvs_eval" / "validation_study" CORPUS_DIR = STUDY_DIR / "corpus" LABELS_DIR = STUDY_DIR / "labels" RESULTS_DIR = STUDY_DIR / "results" DETECTABLE_FRAMES = [ "FVS-001", "FVS-002", "FVS-007", "FVS-008", "FVS-009", "FVS-010", "FVS-011", "FVS-012", "FVS-014", "FVS-015", "FVS-016", ] def _cohen_kappa(a, b): n = len(a) if n == 0: return 0.0 agree = sum(1 for x, y in zip(a, b) if x == y) / n pa1 = sum(a) / n pb1 = sum(b) / n pe = pa1 * pb1 + (1 - pa1) * (1 - pb1) if pe == 1.0: return 1.0 if agree == 1.0 else 0.0 return (agree - pe) / (1 - pe) def _f1(pred, ref): tp = sum(1 for p, r in zip(pred, ref) if p == 1 and r == 1) fp = sum(1 for p, r in zip(pred, ref) if p == 1 and r == 0) fn = sum(1 for p, r in zip(pred, ref) if p == 0 and r == 1) tn = sum(1 for p, r in zip(pred, ref) if p == 0 and r == 0) prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0 rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0 f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0 return f1, prec, rec, tp, fp, fn, tn def _binary(labels_obj, doc_id, frame_ids): doc = labels_obj["labels"].get(doc_id, {}) return [1 if doc.get(f, {}).get("exhibits") else 0 for f in frame_ids] def _majority_union(a, b): return [1 if (x or y) else 0 for x, y in zip(a, b)] def main(): with open(CORPUS_DIR / "manifest.json") as f: manifest = json.load(f) with open(LABELS_DIR / "labels_curator_v3.json") as f: curator = json.load(f) with open(LABELS_DIR / "labels_llm_v3.json") as f: llm = json.load(f) with open(LABELS_DIR / "labels_detector_v1_v3corpus.json") as f: det_v1 = json.load(f) with open(LABELS_DIR / "labels_detector_v3.json") as f: det_v3 = json.load(f) doc_ids = [d["doc_id"] for d in manifest["documents"]] strata = {d["doc_id"]: d["stratum"] for d in manifest["documents"]} per_frame = {} for i, fid in enumerate(DETECTABLE_FRAMES): a_vec = [_binary(curator, d, DETECTABLE_FRAMES)[i] for d in doc_ids] b_vec = [_binary(llm, d, DETECTABLE_FRAMES)[i] for d in doc_ids] v1_vec = [_binary(det_v1, d, DETECTABLE_FRAMES)[i] for d in doc_ids] v3_vec = [_binary(det_v3, d, DETECTABLE_FRAMES)[i] for d in doc_ids] maj_u = _majority_union(a_vec, b_vec) f1_v1, p_v1, r_v1, *_ = _f1(v1_vec, maj_u) f1_v3, p_v3, r_v3, tp3, fp3, fn3, tn3 = _f1(v3_vec, maj_u) f1_v3_A, *_ = _f1(v3_vec, a_vec) per_frame[fid] = { "n_A_exhibits": sum(a_vec), "n_B_exhibits": sum(b_vec), "n_v1_exhibits": sum(v1_vec), "n_v3_exhibits": sum(v3_vec), "n_majority_union": sum(maj_u), "f1_v1_vs_maj_u": round(f1_v1, 3), "f1_v3_vs_maj_u": round(f1_v3, 3), "f1_v3_vs_A": round(f1_v3_A, 3), "precision_v3": round(p_v3, 3), "recall_v3": round(r_v3, 3), "delta_f1": round(f1_v3 - f1_v1, 3), "tp_fp_fn_tn_v3": [tp3, fp3, fn3, tn3], } macro_v1 = mean(f["f1_v1_vs_maj_u"] for f in per_frame.values()) macro_v3 = mean(f["f1_v3_vs_maj_u"] for f in per_frame.values()) macro_v3_A = mean(f["f1_v3_vs_A"] for f in per_frame.values()) macro_recall_v3 = mean(f["recall_v3"] for f in per_frame.values()) # Per-stratum per_doc = {} per_stratum_v1 = defaultdict(list) per_stratum_v3 = defaultdict(list) for doc_id in doc_ids: a = _binary(curator, doc_id, DETECTABLE_FRAMES) b = _binary(llm, doc_id, DETECTABLE_FRAMES) v1 = _binary(det_v1, doc_id, DETECTABLE_FRAMES) v3 = _binary(det_v3, doc_id, DETECTABLE_FRAMES) maj = _majority_union(a, b) f1_v1_d = _f1(v1, maj)[0] f1_v3_d = _f1(v3, maj)[0] per_doc[doc_id] = { "stratum": strata[doc_id], "A_labels": [DETECTABLE_FRAMES[i] for i, v in enumerate(a) if v], "B_labels": [DETECTABLE_FRAMES[i] for i, v in enumerate(b) if v], "v1_labels": [DETECTABLE_FRAMES[i] for i, v in enumerate(v1) if v], "v3_labels": [DETECTABLE_FRAMES[i] for i, v in enumerate(v3) if v], "f1_v1": round(f1_v1_d, 3), "f1_v3": round(f1_v3_d, 3), } per_stratum_v1[strata[doc_id]].append(f1_v1_d) per_stratum_v3[strata[doc_id]].append(f1_v3_d) # Hypothesis tests def _threshold_verdict(f1_val, threshold=0.35): return "PASS" if f1_val >= threshold else "FAIL" hypotheses = { "H-A1_FVS-012_F1_>=_0.35": { "observed": per_frame["FVS-012"]["f1_v3_vs_maj_u"], "threshold": 0.35, "verdict": _threshold_verdict(per_frame["FVS-012"]["f1_v3_vs_maj_u"]), }, "H-A2_FVS-016_F1_>=_0.35": { "observed": per_frame["FVS-016"]["f1_v3_vs_maj_u"], "threshold": 0.35, "verdict": _threshold_verdict(per_frame["FVS-016"]["f1_v3_vs_maj_u"]), }, "H-A3_FVS-008_F1_>=_0.35": { "observed": per_frame["FVS-008"]["f1_v3_vs_maj_u"], "threshold": 0.35, "verdict": _threshold_verdict(per_frame["FVS-008"]["f1_v3_vs_maj_u"]), }, "H-A4_FVS-015_F1_>=_0.35": { "observed": per_frame["FVS-015"]["f1_v3_vs_maj_u"], "threshold": 0.35, "verdict": _threshold_verdict(per_frame["FVS-015"]["f1_v3_vs_maj_u"]), }, "H-A5_macro-F1_>=_0.4": { "observed": round(macro_v3, 3), "threshold": 0.4, "verdict": "PASS" if macro_v3 >= 0.4 else "FAIL", }, "H-A7_macro-recall_>=_0.5": { "observed": round(macro_recall_v3, 3), "threshold": 0.5, "verdict": "PASS" if macro_recall_v3 >= 0.5 else "FAIL", }, } results = { "version": 3, "study": "Track A: expanded corpus (n=28) + v3 signal-level changes", "corpus_size": len(doc_ids), "per_stratum_n": {s: sum(1 for x in strata.values() if x == s) for s in "ABC"}, "aggregate": { "macro_f1_v1_vs_maj_union": round(macro_v1, 3), "macro_f1_v3_vs_maj_union": round(macro_v3, 3), "macro_f1_v3_vs_curator": round(macro_v3_A, 3), "macro_recall_v3_vs_maj_union": round(macro_recall_v3, 3), "delta_macro_f1": round(macro_v3 - macro_v1, 3), "per_stratum_v1_f1": {s: round(mean(vs), 3) for s, vs in per_stratum_v1.items()}, "per_stratum_v3_f1": {s: round(mean(vs), 3) for s, vs in per_stratum_v3.items()}, }, "per_frame": per_frame, "per_document": per_doc, "hypothesis_tests": hypotheses, } with open(RESULTS_DIR / "results_v3.json", "w") as f: json.dump(results, f, indent=2) print("=" * 72) print("Track A Results: v1 vs v3 on expanded corpus (n=28)") print("=" * 72) print(f"\nCorpus: A={results['per_stratum_n']['A']}, B={results['per_stratum_n']['B']}, C={results['per_stratum_n']['C']}") print(f"\n--- Aggregate ---") print(f" Macro-F1 v1: {macro_v1:.3f}") print(f" Macro-F1 v3: {macro_v3:.3f} (delta {macro_v3-macro_v1:+.3f})") print(f" Macro-F1 v3 vs curator only: {macro_v3_A:.3f}") print(f" Macro-recall v3: {macro_recall_v3:.3f}") print(f"\n--- Per-stratum F1 ---") for s in "ABC": v1v = mean(per_stratum_v1[s]) v3v = mean(per_stratum_v3[s]) print(f" Stratum {s} (n={len(per_stratum_v1[s])}): v1={v1v:.3f} -> v3={v3v:.3f} (delta {v3v-v1v:+.3f})") print(f"\n--- Per-frame F1 (v1 -> v3) ---") print(f" {'Frame':<8} {'nA':>3} {'nB':>3} {'nM':>3} {'v1':>3} {'v3':>3} {'F1v1':>5} {'F1v3':>5} {'P_v3':>5} {'R_v3':>5} {'delta':>6}") for fid in DETECTABLE_FRAMES: f = per_frame[fid] print(f" {fid:<8} {f['n_A_exhibits']:>3} {f['n_B_exhibits']:>3} {f['n_majority_union']:>3} {f['n_v1_exhibits']:>3} {f['n_v3_exhibits']:>3} {f['f1_v1_vs_maj_u']:>5.3f} {f['f1_v3_vs_maj_u']:>5.3f} {f['precision_v3']:>5.3f} {f['recall_v3']:>5.3f} {f['delta_f1']:>+6.3f}") print(f"\n--- Pre-registered hypothesis tests ---") for name, h in hypotheses.items(): print(f" {name}: observed={h['observed']:.3f}, threshold={h['threshold']}, {h['verdict']}") print(f"\nResults: {RESULTS_DIR / 'results_v3.json'}") if __name__ == "__main__": main()