| """Held-out Phase 4 SFT validation metrics and acceptance gate.""" |
| from __future__ import annotations |
|
|
| import math |
|
|
| from src.rewards.checker import statement_iou |
|
|
|
|
| def wilson_bounds(correct: int, n: int) -> list[float] | None: |
| if not n: |
| return None |
| p = correct / n |
| z = 1.96 |
| denom = 1 + z*z/n |
| center = (p + z*z/(2*n)) / denom |
| radius = z * math.sqrt(p*(1-p)/n + z*z/(4*n*n)) / denom |
| return [max(0.0, center-radius), min(1.0, center+radius)] |
|
|
|
|
| def class_diagnostics(rows: list[dict]) -> dict: |
| classes = {} |
| predictions = {name: 0 for name in ("SAFE", "VULNERABLE", "INVALID")} |
| for name, target in (("SAFE", 0), ("VULNERABLE", 1)): |
| subset = [r for r in rows if int(r["target"]) == target] |
| correct = sum(bool(r["format_ok"]) and r.get("pred_label") == name for r in subset) |
| classes[name] = {"n": len(subset), "correct": correct, |
| "accuracy": correct / len(subset) if subset else None, |
| "wilson_95": wilson_bounds(correct, len(subset))} |
| for row in rows: |
| name = row.get("pred_label") if row["format_ok"] else "INVALID" |
| predictions[name if name in predictions else "INVALID"] += 1 |
| fractions = {k: v/len(rows) for k,v in predictions.items()} if rows else {} |
| acc = [classes[k]["accuracy"] for k in ("SAFE", "VULNERABLE")] |
| gap = abs(acc[0]-acc[1]) if None not in acc else None |
| flags = [] |
| if gap is not None and gap > .30: |
| flags.append("per_class_accuracy_gap_gt_30pp") |
| if any(fractions.get(k, 0) > .80 for k in ("SAFE", "VULNERABLE")): |
| flags.append("single_class_prediction_gt_80pct") |
| return {"per_class_label_accuracy": classes, "prediction_counts": predictions, |
| "prediction_fractions": fractions, "per_class_accuracy_gap": gap, |
| "collapse_flags": flags, "collapse_flagged": bool(flags)} |
|
|
|
|
| def summarize_validation(rows: list[dict]) -> dict: |
| if not rows: |
| raise ValueError("validation rows are empty") |
| n = len(rows) |
| n_vuln = sum(int(row["target"]) == 1 for row in rows) |
| majority_baseline = max(n_vuln, n - n_vuln) / n |
| n_parsed = sum(bool(row["format_ok"]) for row in rows) |
| n_correct = sum(bool(row["format_ok"]) and row.get("pred_label") == |
| ("VULNERABLE" if int(row["target"]) == 1 else "SAFE") for row in rows) |
| vulnerable_parsed = [row for row in rows if int(row["target"]) == 1 and row["format_ok"]] |
| ious = [statement_iou(set(row.get("pred_lines", [])), set(row.get("gold_lines", []))) |
| for row in vulnerable_parsed] |
| accuracy = n_correct / n |
| pairs: dict[str, dict[int, bool]] = {} |
| for row in rows: |
| if "pair_id" not in row: |
| continue |
| target = int(row["target"]) |
| correct = bool(row["format_ok"]) and row.get("pred_label") == ( |
| "VULNERABLE" if target == 1 else "SAFE") |
| pairs.setdefault(row["pair_id"], {})[target] = correct |
| pair_pass = (sum(p.get(0, False) and p.get(1, False) for p in pairs.values()) / len(pairs) |
| if pairs else None) |
| z = 1.96 |
| denominator = 1 + z * z / n |
| wilson_lower = (accuracy + z * z / (2 * n) - z * math.sqrt( |
| accuracy * (1 - accuracy) / n + z * z / (4 * n * n))) / denominator |
| |
| |
| mean_iou = sum(ious) / len(ious) if ious else None |
| gate_passed = (n_parsed / n >= 0.99 and wilson_lower > majority_baseline |
| and mean_iou is not None and mean_iou > 0) |
| return {"n_rows": n, "n_parsed": n_parsed, "parse_rate": n_parsed / n, |
| "n_label_correct": n_correct, "label_accuracy": accuracy, |
| "majority_baseline": majority_baseline, |
| "label_accuracy_wilson_95_lower": wilson_lower, |
| "n_vulnerable_parsed_for_iou": len(ious), |
| "mean_statement_iou": mean_iou, |
| "val_p_pass_at_1": pair_pass, |
| "n_pairs_for_p_pass": len(pairs), |
| "phase4_gate_passed": gate_passed, **class_diagnostics(rows)} |
|
|
