File size: 3,571 Bytes
646ccfc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """Audit eligible step-200 VALID R2 scores by gold label (CPU-only)."""
from __future__ import annotations
import argparse
import hashlib
import json
from collections import defaultdict
from pathlib import Path
import statistics
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def audit(predictions_path: Path, annotations_path: Path) -> dict:
annotations = {}
with annotations_path.open() as stream:
for line in stream:
row = json.loads(line)
annotations[row["sample_id"]] = row
by_label: dict[str, list[float]] = defaultdict(list)
both_empty: dict[str, int] = defaultdict(int)
eligible_correct: dict[str, int] = defaultdict(int)
correct_but_ineligible: dict[str, int] = defaultdict(int)
total = 0
with predictions_path.open() as stream:
for line in stream:
row = json.loads(line)
total += 1
label = "VULNERABLE" if int(row["target"]) == 1 else "SAFE"
if row["pred_label"] != label:
continue
# Mirrors score_sample.r2_eligible: all SAFE rows are eligible;
# vulnerable rows require at least one annotated L1 line.
eligible = label == "SAFE" or bool(
annotations.get(row["sample_id"], {}).get("l1_lines")
)
if not eligible:
correct_but_ineligible[label] += 1
continue
pred = set(row["pred_lines"])
gold = set(row["gold_lines"])
union = pred | gold
iou = 1.0 if not union else len(pred & gold) / len(union)
by_label[label].append(iou)
eligible_correct[label] += 1
both_empty[label] += int(not pred and not gold)
return {
"predictions": str(predictions_path),
"predictions_sha256": sha256(predictions_path),
"annotations": str(annotations_path),
"annotations_sha256": sha256(annotations_path),
"total_valid_rows": total,
"correct_but_r2_ineligible": dict(correct_but_ineligible),
"eligible_correct_predictions": {
label: {
"n": eligible_correct[label],
"mean_r2": statistics.mean(scores) if scores else None,
"both_empty_n": both_empty[label],
"both_empty_fraction": (
both_empty[label] / len(scores) if scores else None
),
}
for label, scores in sorted(by_label.items())
},
"scope": "fixed-output R2 replay; no inference, training, or policy counterfactual",
}
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--predictions", type=Path, default=Path(
"results/phase6/scale_20261007_v5/valid_predictions.jsonl"))
parser.add_argument("--annotations", type=Path, default=Path(
"data/processed/annotations.jsonl"))
parser.add_argument("--output", type=Path, default=Path(
"results/phase6/scale_20261007_v5/valid_empty_set_r2_audit.json"))
args = parser.parse_args()
report = audit(args.predictions, args.annotations)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
print(json.dumps(report, indent=2, sort_keys=True))
if __name__ == "__main__":
main()
|