File size: 4,368 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 95 96 97 98 99 100 101 102 103 104 105 106 | """Report the row-level Wilson gate and pair-cluster VALID accuracy interval."""
from __future__ import annotations
import argparse
import hashlib
import json
import math
from collections import defaultdict
from pathlib import Path
import numpy as np
def wilson(k: int, n: int, z: float = 1.96) -> list[float]:
p = k / n
denom = 1 + z * z / n
center = p + z * z / (2 * n)
half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n))
return [(center - half) / denom, (center + half) / denom]
def audit(predictions_path: Path, *, replicates: int = 20_000, seed: int = 42) -> dict:
pairs = defaultdict(list)
with predictions_path.open() as stream:
for line in stream:
row = json.loads(line)
pairs[row["pair_id"]].append(row)
if not pairs or any(len(rows) != 2 for rows in pairs.values()):
raise ValueError("expected exactly two validation rows per pair")
pair_values = []
class_counts = {"SAFE": {"correct": 0, "n": 0},
"VULNERABLE": {"correct": 0, "n": 0}}
outcome_counts = {"both_correct": 0, "safe_only": 0,
"vulnerable_only": 0, "neither": 0}
for rows in pairs.values():
by_target = {int(row["target"]): row for row in rows}
if set(by_target) != {0, 1}:
raise ValueError("each pair must contain one SAFE and one VULNERABLE row")
safe = by_target[0]
vuln = by_target[1]
safe_ok = bool(safe["format_ok"] and safe["pred_label"] == "SAFE")
vuln_ok = bool(vuln["format_ok"] and vuln["pred_label"] == "VULNERABLE")
pair_values.append((int(safe_ok), int(vuln_ok)))
class_counts["SAFE"]["correct"] += int(safe_ok)
class_counts["SAFE"]["n"] += 1
class_counts["VULNERABLE"]["correct"] += int(vuln_ok)
class_counts["VULNERABLE"]["n"] += 1
key = ("both_correct" if safe_ok and vuln_ok else
"safe_only" if safe_ok else
"vulnerable_only" if vuln_ok else "neither")
outcome_counts[key] += 1
array = np.asarray(pair_values, dtype=np.float64)
total_pairs = int(len(array))
total_rows = int(array.size)
total_correct = int(array.sum())
rng = np.random.default_rng(seed)
indices = rng.integers(0, len(array), size=(replicates, len(array)))
pair_bootstrap = array[indices].mean(axis=(1, 2))
threshold = next(k for k in range(total_rows + 1)
if wilson(k, total_rows)[0] > 0.5)
classes = {}
for label, values in class_counts.items():
k, n = values["correct"], values["n"]
classes[label] = {"correct": k, "n": n, "accuracy": k / n,
"wilson_95": wilson(k, n)}
digest = hashlib.sha256(predictions_path.read_bytes()).hexdigest()
return {
"source": str(predictions_path),
"source_sha256": digest,
"pairs": total_pairs,
"rows": total_rows,
"row_accuracy": total_correct / total_rows,
"row_correct": total_correct,
"row_wilson_95": wilson(total_correct, total_rows),
"minimum_row_correct_for_wilson_lower_above_0_50": threshold,
"additional_correct_needed_from_observed": max(0, threshold - total_correct),
"per_class": classes,
"pair_outcome_counts": outcome_counts,
"pair_cluster_bootstrap_accuracy_95": [
float(x) for x in np.quantile(pair_bootstrap, [0.025, 0.975])
],
"pair_bootstrap_replicates": replicates,
"pair_bootstrap_seed": seed,
}
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("--output", type=Path, default=Path(
"results/phase6/scale_20261007_v5/valid_gate_precision_audit.json"))
parser.add_argument("--replicates", type=int, default=20_000)
parser.add_argument("--seed", type=int, default=42)
args = parser.parse_args()
report = audit(args.predictions, replicates=args.replicates, seed=args.seed)
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()
|