""" Deliverable B: re-run detector with v2 rules and compute v1-vs-v2 comparison. Uses frame_library_v2.suggest_frames_v2 on the study corpus. Produces: labels/labels_detector_v2.json results/results_v2.json results/v1_vs_v2_comparison.json Does NOT modify v1 artifacts. v1 remains canonical for the study's primary result; v2 is the post-audit candidate. """ from __future__ import annotations import json import sys 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" sys.path.insert(0, str(REPO_ROOT)) sys.path.insert(0, str(STUDY_DIR)) from framing import ( # noqa: E402 detect_coverage, detect_voice, temporal_orientation, detect_epistemic_basis, ) from frame_library_v2 import suggest_frames_v2, V2_STATUS # noqa: E402 FRAME_IDS_ALL = [f"FVS-{i:03d}" for i in range(1, 21)] 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 run_v2(): # Load corpus + existing labelers with open(CORPUS_DIR / "manifest.json") as f: manifest = json.load(f) with open(LABELS_DIR / "labels_curator.json") as f: curator = json.load(f) with open(LABELS_DIR / "labels_llm.json") as f: llm = json.load(f) with open(LABELS_DIR / "labels_detector.json") as f: detector_v1 = json.load(f) # Compute v2 detector labels by re-running framing + v2 suggest_frames detector_v2 = { "version": 2, "labeler": "detector_v2", "labeler_note": ( "frame_library_v2.suggest_frames_v2 per RULE_AUDIT.md proposed " "revisions. Rule-level changes only; framing.py signals unchanged." ), "v2_status": V2_STATUS, "frame_universe": FRAME_IDS_ALL, "labels": {}, } for doc in manifest["documents"]: doc_id = doc["doc_id"] path = CORPUS_DIR / f"{doc_id}.txt" if not path.exists(): continue text = path.read_text(encoding="utf-8") cov = detect_coverage(text) voice = detect_voice(text) temp = temporal_orientation(text) epist = detect_epistemic_basis(text) suggs = suggest_frames_v2(cov, voice, temp, epist) suggested_ids = {s["fvs_id"] for s in suggs} detector_v2["labels"][doc_id] = { f: {"exhibits": f in suggested_ids} for f in FRAME_IDS_ALL } with open(LABELS_DIR / "labels_detector_v2.json", "w") as f: json.dump(detector_v2, f, indent=2) # ─── Compute v2 metrics ─── doc_ids = [d["doc_id"] for d in manifest["documents"]] strata = {d["doc_id"]: d["stratum"] for d in manifest["documents"]} per_frame_v2 = {} per_doc_v2 = {} 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] c1_vec = [_binary(detector_v1, d, DETECTABLE_FRAMES)[i] for d in doc_ids] c2_vec = [_binary(detector_v2, d, DETECTABLE_FRAMES)[i] for d in doc_ids] maj_u = _majority_union(a_vec, b_vec) f1_v1, *_ = _f1(c1_vec, maj_u) f1_v2, p_v2, r_v2, tp, fp, fn, tn = _f1(c2_vec, maj_u) per_frame_v2[fid] = { "n_A_exhibits": sum(a_vec), "n_B_exhibits": sum(b_vec), "n_C1_exhibits": sum(c1_vec), "n_C2_exhibits": sum(c2_vec), "f1_v1_vs_maj_union": round(f1_v1, 3), "f1_v2_vs_maj_union": round(f1_v2, 3), "delta_f1": round(f1_v2 - f1_v1, 3), "precision_v2": round(p_v2, 3), "recall_v2": round(r_v2, 3), "tp_fp_fn_tn_v2": [tp, fp, fn, tn], "v2_action": V2_STATUS.get(fid, ""), } for doc_id in doc_ids: a = _binary(curator, doc_id, DETECTABLE_FRAMES) b = _binary(llm, doc_id, DETECTABLE_FRAMES) c1 = _binary(detector_v1, doc_id, DETECTABLE_FRAMES) c2 = _binary(detector_v2, doc_id, DETECTABLE_FRAMES) maj = _majority_union(a, b) per_doc_v2[doc_id] = { "stratum": strata[doc_id], "v1_labels": [DETECTABLE_FRAMES[i] for i, v in enumerate(c1) if v], "v2_labels": [DETECTABLE_FRAMES[i] for i, v in enumerate(c2) if v], "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], "f1_v1_vs_maj_union": round(_f1(c1, maj)[0], 3), "f1_v2_vs_maj_union": round(_f1(c2, maj)[0], 3), "kappa_v1_AC": round(_cohen_kappa(a, c1), 3), "kappa_v2_AC": round(_cohen_kappa(a, c2), 3), } macro_v1 = mean(f["f1_v1_vs_maj_union"] for f in per_frame_v2.values()) macro_v2 = mean(f["f1_v2_vs_maj_union"] for f in per_frame_v2.values()) per_stratum_v1 = defaultdict(list) per_stratum_v2 = defaultdict(list) for doc_id, m in per_doc_v2.items(): per_stratum_v1[m["stratum"]].append(m["f1_v1_vs_maj_union"]) per_stratum_v2[m["stratum"]].append(m["f1_v2_vs_maj_union"]) comparison = { "version": 1, "study": "FVS detector validation v2 (post-audit rule revisions)", "corpus_size": len(doc_ids), "aggregate": { "macro_f1_v1": round(macro_v1, 3), "macro_f1_v2": round(macro_v2, 3), "delta_macro_f1": round(macro_v2 - macro_v1, 3), "per_stratum_v1": {s: round(mean(vs), 3) for s, vs in per_stratum_v1.items()}, "per_stratum_v2": {s: round(mean(vs), 3) for s, vs in per_stratum_v2.items()}, }, "per_frame": per_frame_v2, "per_document": per_doc_v2, } with open(RESULTS_DIR / "v1_vs_v2_comparison.json", "w") as f: json.dump(comparison, f, indent=2) # ─── Human-readable summary ─── print("=" * 72) print("FVS v1 vs v2 Detector Comparison") print("=" * 72) print(f"\nMacro-F1 v1: {macro_v1:.3f}") print(f"Macro-F1 v2: {macro_v2:.3f}") print(f"Delta: {macro_v2 - macro_v1:+.3f}") # Threshold interpretation if macro_v2 >= 0.7: verdict = "v2 reaches state-of-art zone (>=0.7)" elif macro_v2 >= 0.6: verdict = "v2 reaches 'useful' zone (>=0.6)" elif macro_v2 >= 0.4: verdict = "v2 reaches 'partial alignment' zone (>=0.4)" else: verdict = "v2 remains in H3 falsification zone (<0.4)" print(f"Verdict: {verdict}") print(f"\n--- Per-stratum macro-F1 (v1 -> v2) ---") for s in sorted(set(strata.values())): v1v = per_stratum_v1[s] v2v = per_stratum_v2[s] print(f" Stratum {s}: {mean(v1v):.3f} -> {mean(v2v):.3f} (delta {mean(v2v) - mean(v1v):+.3f}, n={len(v1v)})") print(f"\n--- Per-frame (v1 F1 -> v2 F1, delta) ---") print(f" {'Frame':<8} {'Action':<35} {'nC1':>3} {'nC2':>3} {'F1v1':>5} {'F1v2':>5} {'ΔF1':>6}") for fid in DETECTABLE_FRAMES: f = per_frame_v2[fid] print(f" {fid:<8} {f['v2_action']:<35} {f['n_C1_exhibits']:>3} {f['n_C2_exhibits']:>3} " f"{f['f1_v1_vs_maj_union']:>5.3f} {f['f1_v2_vs_maj_union']:>5.3f} " f"{f['delta_f1']:>+6.3f}") print(f"\n--- Per-document changes (where v1 != v2) ---") for doc_id in doc_ids: m = per_doc_v2[doc_id] if set(m["v1_labels"]) != set(m["v2_labels"]): print(f"\n {doc_id} ({m['stratum']})") print(f" A: {m['A_labels']}") print(f" B: {m['B_labels']}") print(f" v1: {m['v1_labels']} (F1 {m['f1_v1_vs_maj_union']:.3f})") print(f" v2: {m['v2_labels']} (F1 {m['f1_v2_vs_maj_union']:.3f})") print(f"\nArtifacts: labels/labels_detector_v2.json, results/v1_vs_v2_comparison.json") if __name__ == "__main__": run_v2()