e30764fb19
Add negative_cases_from_universe to expand an explicit file-by-rule universe into labeled negatives (enabling TP/FP/TN/FN, not just TP/FP), plus project-eval harness updates in run_project_eval and the score/score_fp/quality-gate helpers.
331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""Metric calculations shared by deterministic and randomized evaluations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import random
|
|
import statistics
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from typing import Any
|
|
|
|
CLASSIFICATION_THRESHOLD_METRICS = (
|
|
"precision",
|
|
"recall",
|
|
"f1",
|
|
"specificity",
|
|
"accuracy",
|
|
"balanced_accuracy",
|
|
"matthews_correlation_coefficient",
|
|
)
|
|
|
|
|
|
def _count(name: str, value: int) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise ValueError(f"{name} must be a non-negative integer")
|
|
return value
|
|
|
|
|
|
def classification_metrics(
|
|
true_positive: int,
|
|
false_positive: int,
|
|
false_negative: int,
|
|
true_negative: int | None = None,
|
|
*,
|
|
zero_division: float = 0.0,
|
|
) -> dict[str, Any]:
|
|
"""Return confusion counts and derived binary-classification metrics.
|
|
|
|
Static-analysis corpora often do not enumerate all true negatives. In that
|
|
case ``true_negative`` must remain ``None`` and specificity/FPR are omitted
|
|
rather than being fabricated from unlabelled source locations.
|
|
"""
|
|
|
|
tp = _count("true_positive", true_positive)
|
|
fp = _count("false_positive", false_positive)
|
|
fn = _count("false_negative", false_negative)
|
|
tn = None if true_negative is None else _count("true_negative", true_negative)
|
|
if not math.isfinite(zero_division):
|
|
raise ValueError("zero_division must be finite")
|
|
|
|
precision_denominator = tp + fp
|
|
recall_denominator = tp + fn
|
|
precision_defined = precision_denominator > 0
|
|
recall_defined = recall_denominator > 0
|
|
precision = tp / precision_denominator if precision_defined else zero_division
|
|
recall = tp / recall_denominator if recall_defined else zero_division
|
|
f1_denominator = 2 * tp + fp + fn
|
|
f1_defined = f1_denominator > 0
|
|
f1 = 2 * tp / f1_denominator if f1_defined else zero_division
|
|
|
|
result: dict[str, Any] = {
|
|
"true_positive": tp,
|
|
"false_positive": fp,
|
|
"false_negative": fn,
|
|
"precision": precision,
|
|
"recall": recall,
|
|
"f1": f1,
|
|
"defined": {
|
|
"precision": precision_defined,
|
|
"recall": recall_defined,
|
|
"f1": f1_defined,
|
|
},
|
|
}
|
|
if tn is not None:
|
|
negative_denominator = tn + fp
|
|
negative_prediction_denominator = tn + fn
|
|
total = tp + fp + fn + tn
|
|
specificity_defined = negative_denominator > 0
|
|
negative_predictive_value_defined = negative_prediction_denominator > 0
|
|
accuracy_defined = total > 0
|
|
balanced_accuracy_defined = recall_defined and specificity_defined
|
|
mcc_denominator = math.sqrt(
|
|
(tp + fp) * (tp + fn) * (tn + fp) * (tn + fn)
|
|
)
|
|
mcc_defined = mcc_denominator > 0
|
|
result.update(
|
|
{
|
|
"true_negative": tn,
|
|
"specificity": (
|
|
tn / negative_denominator
|
|
if specificity_defined
|
|
else zero_division
|
|
),
|
|
"false_positive_rate": (
|
|
fp / negative_denominator
|
|
if specificity_defined
|
|
else zero_division
|
|
),
|
|
"negative_predictive_value": (
|
|
tn / negative_prediction_denominator
|
|
if negative_predictive_value_defined
|
|
else zero_division
|
|
),
|
|
"accuracy": (
|
|
(tp + tn) / total if accuracy_defined else zero_division
|
|
),
|
|
"balanced_accuracy": (
|
|
(recall + tn / negative_denominator) / 2
|
|
if balanced_accuracy_defined
|
|
else zero_division
|
|
),
|
|
"matthews_correlation_coefficient": (
|
|
(tp * tn - fp * fn) / mcc_denominator
|
|
if mcc_defined
|
|
else zero_division
|
|
),
|
|
}
|
|
)
|
|
result["defined"].update(
|
|
{
|
|
"specificity": specificity_defined,
|
|
"false_positive_rate": specificity_defined,
|
|
"negative_predictive_value": negative_predictive_value_defined,
|
|
"accuracy": accuracy_defined,
|
|
"balanced_accuracy": balanced_accuracy_defined,
|
|
"matthews_correlation_coefficient": mcc_defined,
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def distribution_summary(values: Sequence[float]) -> dict[str, float | int]:
|
|
"""Summarize repeated measurements without external statistics packages.
|
|
|
|
The quartile indexing intentionally preserves the historical Fold benchmark
|
|
convention, keeping existing performance baselines comparable.
|
|
"""
|
|
|
|
finite = [float(value) for value in values]
|
|
if any(not math.isfinite(value) for value in finite):
|
|
raise ValueError("distribution values must be finite")
|
|
if not finite:
|
|
return {"n": 0, "median": 0.0, "min": 0.0, "max": 0.0, "iqr": 0.0}
|
|
ordered = sorted(finite)
|
|
q1 = ordered[len(ordered) // 4]
|
|
q3 = ordered[(3 * len(ordered)) // 4]
|
|
return {
|
|
"n": len(ordered),
|
|
"median": round(float(statistics.median(ordered)), 3),
|
|
"min": round(ordered[0], 3),
|
|
"max": round(ordered[-1], 3),
|
|
"iqr": round(q3 - q1, 3),
|
|
}
|
|
|
|
|
|
def bootstrap_interval(
|
|
values: Sequence[float],
|
|
*,
|
|
statistic: Callable[[Sequence[float]], float] = statistics.median,
|
|
confidence: float = 0.95,
|
|
resamples: int = 2_000,
|
|
seed: int = 0,
|
|
) -> dict[str, float | int]:
|
|
"""Return a deterministic percentile-bootstrap interval.
|
|
|
|
This is intended for randomized DAST/fuzzing campaigns. It should not be
|
|
used to manufacture certainty from one or two smoke-test repetitions.
|
|
"""
|
|
|
|
sample = [float(value) for value in values]
|
|
if not sample:
|
|
raise ValueError("bootstrap_interval requires at least one value")
|
|
if any(not math.isfinite(value) for value in sample):
|
|
raise ValueError("bootstrap values must be finite")
|
|
if not 0.0 < confidence < 1.0:
|
|
raise ValueError("confidence must be between zero and one")
|
|
if resamples < 1:
|
|
raise ValueError("resamples must be positive")
|
|
|
|
rng = random.Random(seed)
|
|
draws = sorted(
|
|
float(statistic([rng.choice(sample) for _ in sample]))
|
|
for _ in range(resamples)
|
|
)
|
|
alpha = (1.0 - confidence) / 2.0
|
|
low_index = max(0, min(resamples - 1, math.floor(alpha * resamples)))
|
|
high_index = max(
|
|
0, min(resamples - 1, math.ceil((1.0 - alpha) * resamples) - 1)
|
|
)
|
|
return {
|
|
"estimate": float(statistic(sample)),
|
|
"low": draws[low_index],
|
|
"high": draws[high_index],
|
|
"confidence": confidence,
|
|
"resamples": resamples,
|
|
"seed": seed,
|
|
}
|
|
|
|
|
|
def aggregate_classification_metrics(
|
|
slices: Mapping[str, Mapping[str, int | None]],
|
|
) -> dict[str, Any]:
|
|
"""Return per-slice, micro, and macro classification measurements.
|
|
|
|
A pooled score can hide a weak detector/corpus stratum behind a large easy
|
|
stratum. This helper preserves every count, computes the micro aggregate by
|
|
summing counts, and computes each macro ratio only across slices where that
|
|
ratio is actually defined. True negatives remain unavailable unless every
|
|
slice explicitly enumerates them.
|
|
"""
|
|
|
|
per_slice: dict[str, dict[str, Any]] = {}
|
|
total_tp = total_fp = total_fn = 0
|
|
total_tn = 0
|
|
all_have_tn = True
|
|
for name in sorted(slices):
|
|
row = slices[name]
|
|
tp = _count(f"{name}.true_positive", row.get("true_positive", 0))
|
|
fp = _count(f"{name}.false_positive", row.get("false_positive", 0))
|
|
fn = _count(f"{name}.false_negative", row.get("false_negative", 0))
|
|
raw_tn = row.get("true_negative")
|
|
tn = None if raw_tn is None else _count(f"{name}.true_negative", raw_tn)
|
|
per_slice[name] = classification_metrics(tp, fp, fn, tn)
|
|
total_tp += tp
|
|
total_fp += fp
|
|
total_fn += fn
|
|
if tn is None:
|
|
all_have_tn = False
|
|
else:
|
|
total_tn += tn
|
|
|
|
micro = classification_metrics(
|
|
total_tp,
|
|
total_fp,
|
|
total_fn,
|
|
total_tn if all_have_tn and per_slice else None,
|
|
)
|
|
ratio_names = (
|
|
"precision",
|
|
"recall",
|
|
"f1",
|
|
"specificity",
|
|
"false_positive_rate",
|
|
"negative_predictive_value",
|
|
"accuracy",
|
|
"balanced_accuracy",
|
|
"matthews_correlation_coefficient",
|
|
)
|
|
macro: dict[str, float] = {}
|
|
macro_support: dict[str, int] = {}
|
|
for metric in ratio_names:
|
|
values = [
|
|
float(row[metric])
|
|
for row in per_slice.values()
|
|
if metric in row and row.get("defined", {}).get(metric, False)
|
|
]
|
|
if values:
|
|
macro[metric] = sum(values) / len(values)
|
|
macro_support[metric] = len(values)
|
|
return {
|
|
"slices": per_slice,
|
|
"micro": micro,
|
|
"macro": macro,
|
|
"macro_support": macro_support,
|
|
}
|
|
|
|
|
|
def paired_comparison(
|
|
baseline: Sequence[float],
|
|
candidate: Sequence[float],
|
|
*,
|
|
lower_is_better: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Compare paired trials with an effect size and exact sign test.
|
|
|
|
Pairing is by seed/target/trial identity. The common-language effect is the
|
|
fraction of non-tied pairs where the candidate wins; the two-sided sign
|
|
test is dependency-free and makes no distributional assumption. This is a
|
|
useful minimum for DAST/fuzz comparisons, not a substitute for enough
|
|
independent trials or a predeclared experiment design.
|
|
"""
|
|
|
|
before = [float(value) for value in baseline]
|
|
after = [float(value) for value in candidate]
|
|
if not before or len(before) != len(after):
|
|
raise ValueError("paired samples must be non-empty and have equal length")
|
|
if any(not math.isfinite(value) for value in (*before, *after)):
|
|
raise ValueError("paired samples must be finite")
|
|
|
|
deltas = [right - left for left, right in zip(before, after, strict=True)]
|
|
better = [
|
|
delta < 0 if lower_is_better else delta > 0
|
|
for delta in deltas
|
|
]
|
|
worse = [
|
|
delta > 0 if lower_is_better else delta < 0
|
|
for delta in deltas
|
|
]
|
|
wins = sum(better)
|
|
losses = sum(worse)
|
|
ties = len(deltas) - wins - losses
|
|
non_ties = wins + losses
|
|
|
|
if non_ties:
|
|
tail = sum(
|
|
math.comb(non_ties, count)
|
|
for count in range(0, min(wins, losses) + 1)
|
|
) / (2**non_ties)
|
|
sign_test_p = min(1.0, 2.0 * tail)
|
|
common_language = wins / non_ties
|
|
else:
|
|
sign_test_p = 1.0
|
|
common_language = 0.5
|
|
|
|
relative = [
|
|
100.0 * delta / left
|
|
for left, delta in zip(before, deltas, strict=True)
|
|
if left != 0
|
|
]
|
|
return {
|
|
"pairs": len(deltas),
|
|
"wins": wins,
|
|
"losses": losses,
|
|
"ties": ties,
|
|
"median_delta": float(statistics.median(deltas)),
|
|
"median_relative_change_pct": (
|
|
float(statistics.median(relative)) if relative else None
|
|
),
|
|
"common_language_effect": common_language,
|
|
"two_sided_sign_test_p": sign_test_p,
|
|
"lower_is_better": lower_is_better,
|
|
}
|