File size: 5,322 Bytes
5bfc214 | 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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """Error analysis (spec §12): partitions FP/FN by CWE (head/tail), project,
function length quartile, cyclomatic complexity, context requirement
(intra-function vs needs-callee), data-flow distance (trace_gt length),
training-frequency of CWE, and trace-validity of the emitted explanation.
Pure partitioning/aggregation functions over prediction records; plotting and
the human-audit sampling tool are separate (matplotlib-dependent) concerns
layered on top, not needed for the partition logic itself to be correct and
testable.
"""
from __future__ import annotations
import random
from collections import defaultdict
from dataclasses import dataclass, field
from src.eval.metrics import head_tail_cwe_split
@dataclass
class ErrorRecord:
sample_id: str
project: str
cwe: str | None
label_true: int
label_pred: int
func_len: int
cyclomatic_complexity: int
has_context: bool
trace_gt_len: int # 0 if no trace ground truth
train_cwe_frequency: int = 0
def is_false_positive(r: ErrorRecord) -> bool:
return r.label_true == 0 and r.label_pred == 1
def is_false_negative(r: ErrorRecord) -> bool:
return r.label_true == 1 and r.label_pred == 0
def partition_by_cwe_head_tail(
records: list[ErrorRecord], cwe_counts: dict[str, int], head_k: int = 12
) -> dict[str, dict]:
head, tail = head_tail_cwe_split(cwe_counts, head_k=head_k)
def bucket(r: ErrorRecord) -> str:
if r.cwe is None:
return "unknown"
return "head" if r.cwe in head else "tail"
return _error_rates_by_key(records, bucket)
def partition_by_project(records: list[ErrorRecord]) -> dict[str, dict]:
return _error_rates_by_key(records, lambda r: r.project)
def _quartile_labels(values: list[int]) -> list[str]:
if not values:
return []
sorted_vals = sorted(values)
n = len(sorted_vals)
q1 = sorted_vals[n // 4]
q2 = sorted_vals[n // 2]
q3 = sorted_vals[(3 * n) // 4]
def label(v: int) -> str:
if v <= q1:
return "Q1"
if v <= q2:
return "Q2"
if v <= q3:
return "Q3"
return "Q4"
return [label(v) for v in values]
def partition_by_function_length_quartile(records: list[ErrorRecord]) -> dict[str, dict]:
lengths = [r.func_len for r in records]
labels = _quartile_labels(lengths)
label_by_id = dict(zip((r.sample_id for r in records), labels))
return _error_rates_by_key(records, lambda r: label_by_id[r.sample_id])
def partition_by_cyclomatic_complexity_quartile(records: list[ErrorRecord]) -> dict[str, dict]:
values = [r.cyclomatic_complexity for r in records]
labels = _quartile_labels(values)
label_by_id = dict(zip((r.sample_id for r in records), labels))
return _error_rates_by_key(records, lambda r: label_by_id[r.sample_id])
def partition_by_context_requirement(records: list[ErrorRecord]) -> dict[str, dict]:
return _error_rates_by_key(records, lambda r: "needs_callee" if r.has_context else "intra_function")
def partition_by_dataflow_distance(records: list[ErrorRecord]) -> dict[str, dict]:
def bucket(r: ErrorRecord) -> str:
if r.trace_gt_len == 0:
return "no_trace_gt"
if r.trace_gt_len <= 2:
return "short (<=2 hops)"
if r.trace_gt_len <= 5:
return "medium (3-5 hops)"
return "long (>5 hops)"
return _error_rates_by_key(records, bucket)
def partition_by_cwe_training_frequency(
records: list[ErrorRecord], low_freq_threshold: int = 10
) -> dict[str, dict]:
def bucket(r: ErrorRecord) -> str:
if r.cwe is None:
return "unknown"
return "low_frequency" if r.train_cwe_frequency < low_freq_threshold else "high_frequency"
return _error_rates_by_key(records, bucket)
def _error_rates_by_key(records: list[ErrorRecord], key_fn) -> dict[str, dict]:
groups: dict[str, list[ErrorRecord]] = defaultdict(list)
for r in records:
groups[key_fn(r)].append(r)
out: dict[str, dict] = {}
for key, group in groups.items():
n = len(group)
n_fp = sum(1 for r in group if is_false_positive(r))
n_fn = sum(1 for r in group if is_false_negative(r))
n_correct = sum(1 for r in group if r.label_true == r.label_pred)
out[key] = {
"n": n,
"fp_rate": n_fp / n if n else 0.0,
"fn_rate": n_fn / n if n else 0.0,
"accuracy": n_correct / n if n else 0.0,
}
return out
@dataclass
class HumanAuditSample:
sample_id: str
split_name: str
def sample_for_human_audit(
by_split: dict[str, list[str]], n_per_split: int = 50, seed: int = 42
) -> list[HumanAuditSample]:
"""spec §12 human audit protocol: "sample 150 model outputs (50 each
from chronological/repo-disjoint/OOD)". `by_split` maps split name to the
sample_ids available in it (typically outputs from the final TraceGRPO
model's predictions on that split's test set)."""
rng = random.Random(seed)
out: list[HumanAuditSample] = []
for split_name, sample_ids in by_split.items():
chosen = sample_ids[:]
rng.shuffle(chosen)
for sid in chosen[:n_per_split]:
out.append(HumanAuditSample(sample_id=sid, split_name=split_name))
return out
|