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.
261 lines
9.2 KiB
Python
261 lines
9.2 KiB
Python
"""Ground-truth matching for normalized Fold/comparator finding records."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from pathlib import PurePosixPath
|
|
from typing import Any
|
|
|
|
from .metrics import classification_metrics
|
|
|
|
|
|
def _path(value: Any) -> str:
|
|
return str(value or "").replace("\\", "/").removeprefix("file://")
|
|
|
|
|
|
def path_matches(expected: str, actual: str) -> bool:
|
|
"""Match a gold path without collapsing distinct nested paths to basenames."""
|
|
|
|
wanted = _path(expected)
|
|
while wanted.startswith("./"):
|
|
wanted = wanted[2:]
|
|
observed = _path(actual)
|
|
if not wanted or not observed:
|
|
return False
|
|
if "/" not in wanted:
|
|
return PurePosixPath(observed).name == wanted
|
|
return observed == wanted or observed.endswith(f"/{wanted}")
|
|
|
|
|
|
def _kind_matches(expected: str, observed: set[str]) -> bool:
|
|
return (
|
|
expected == "any"
|
|
or expected in observed
|
|
or any(
|
|
kind.startswith(f"{expected}{separator}")
|
|
for kind in observed
|
|
for separator in ("-", ":", ".", "/")
|
|
)
|
|
)
|
|
|
|
|
|
def _group_matches_gold(group: list[dict[str, Any]], gold: dict[str, Any]) -> bool:
|
|
expected_kind = str(gold.get("kind", "any"))
|
|
observed_kinds = {str(finding.get("kind", "")) for finding in group}
|
|
if not _kind_matches(expected_kind, observed_kinds):
|
|
return False
|
|
|
|
observed_files = [str(finding.get("file", "")) for finding in group]
|
|
expected_files = gold.get("files", [])
|
|
return isinstance(expected_files, list) and all(
|
|
any(path_matches(str(expected), actual) for actual in observed_files)
|
|
for expected in expected_files
|
|
)
|
|
|
|
|
|
def _maximum_matching(
|
|
candidates: dict[int, list[str]],
|
|
) -> tuple[dict[int, str], dict[str, int]]:
|
|
"""Compute a deterministic one-to-one gold-to-finding-group matching."""
|
|
|
|
group_to_gold: dict[str, int] = {}
|
|
|
|
def augment(gold_index: int, seen: set[str]) -> bool:
|
|
for group_id in candidates.get(gold_index, []):
|
|
if group_id in seen:
|
|
continue
|
|
seen.add(group_id)
|
|
previous = group_to_gold.get(group_id)
|
|
if previous is None or augment(previous, seen):
|
|
group_to_gold[group_id] = gold_index
|
|
return True
|
|
return False
|
|
|
|
# Constrained cases choose first, reducing order-dependent ambiguity.
|
|
for gold_index in sorted(candidates, key=lambda index: (len(candidates[index]), index)):
|
|
augment(gold_index, set())
|
|
gold_to_group = {gold: group for group, gold in group_to_gold.items()}
|
|
return gold_to_group, group_to_gold
|
|
|
|
|
|
def score_grouped_findings(
|
|
findings: list[dict[str, Any]], gold_entries: list[dict[str, Any]]
|
|
) -> dict[str, Any]:
|
|
"""Score finding groups against gold cases with one-to-one credit.
|
|
|
|
A clone/equivalence cluster can contain several location records, hence the
|
|
group-by-id model. One emitted group may satisfy at most one gold case;
|
|
allowing it to satisfy several silently inflates recall.
|
|
"""
|
|
|
|
groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for index, finding in enumerate(findings):
|
|
raw_id = finding.get("id")
|
|
group_id = str(raw_id) if raw_id not in (None, "") else f"__finding_{index}"
|
|
groups[group_id].append(finding)
|
|
|
|
candidates: dict[int, list[str]] = {}
|
|
for gold_index, gold in enumerate(gold_entries):
|
|
candidates[gold_index] = sorted(
|
|
group_id
|
|
for group_id, group in groups.items()
|
|
if _group_matches_gold(group, gold)
|
|
)
|
|
gold_to_group, group_to_gold = _maximum_matching(candidates)
|
|
|
|
true_positive = len(gold_to_group)
|
|
false_positive = len(groups) - len(group_to_gold)
|
|
false_negative = len(gold_entries) - true_positive
|
|
metrics = classification_metrics(true_positive, false_positive, false_negative)
|
|
return {
|
|
"true_positive": true_positive,
|
|
"false_positive": false_positive,
|
|
"false_negative": false_negative,
|
|
"gold_total": len(gold_entries),
|
|
"finding_groups": len(groups),
|
|
"matched_findings": len(group_to_gold),
|
|
"precision": metrics["precision"],
|
|
"recall": metrics["recall"],
|
|
"f1": metrics["f1"],
|
|
"metric_defined": metrics["defined"],
|
|
"matched_gold_indices": sorted(gold_to_group),
|
|
"missed_gold_indices": sorted(set(range(len(gold_entries))) - gold_to_group.keys()),
|
|
"unmatched_finding_ids": sorted(set(groups) - group_to_gold.keys()),
|
|
}
|
|
|
|
|
|
def negative_cases_from_universe(
|
|
positive_entries: list[dict[str, Any]], universe: dict[str, Any]
|
|
) -> list[dict[str, Any]]:
|
|
"""Expand an explicit file-by-rule universe into labeled negative cases."""
|
|
|
|
files = universe.get("files", [])
|
|
kinds = universe.get("kinds", [])
|
|
if (
|
|
not isinstance(files, list)
|
|
or not files
|
|
or not all(isinstance(item, str) and item for item in files)
|
|
or not isinstance(kinds, list)
|
|
or not kinds
|
|
or not all(isinstance(item, str) and item for item in kinds)
|
|
):
|
|
raise ValueError("gold universe requires non-empty string files and kinds")
|
|
|
|
positive_pairs = {
|
|
(str(file_name), str(entry.get("kind", "")))
|
|
for entry in positive_entries
|
|
for file_name in entry.get("files", [])
|
|
if isinstance(entry.get("files"), list)
|
|
}
|
|
return [
|
|
{
|
|
"id": f"negative:{kind}:{file_name}",
|
|
"kind": kind,
|
|
"files": [file_name],
|
|
}
|
|
for kind in kinds
|
|
for file_name in files
|
|
if (file_name, kind) not in positive_pairs
|
|
]
|
|
|
|
|
|
def score_labeled_cases(
|
|
findings: list[dict[str, Any]],
|
|
positive_entries: list[dict[str, Any]],
|
|
negative_entries: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
"""Score explicit positive and negative cases as a confusion matrix."""
|
|
|
|
positive_score = score_grouped_findings(findings, positive_entries)
|
|
groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for index, finding in enumerate(findings):
|
|
raw_id = finding.get("id")
|
|
group_id = str(raw_id) if raw_id not in (None, "") else f"__finding_{index}"
|
|
groups[group_id].append(finding)
|
|
|
|
negative_hits: dict[int, list[str]] = {
|
|
index: sorted(
|
|
group_id
|
|
for group_id, group in groups.items()
|
|
if _group_matches_gold(group, negative)
|
|
)
|
|
for index, negative in enumerate(negative_entries)
|
|
}
|
|
false_positive_indices = sorted(
|
|
index for index, matches in negative_hits.items() if matches
|
|
)
|
|
true_negative_indices = sorted(
|
|
set(range(len(negative_entries))) - set(false_positive_indices)
|
|
)
|
|
false_positive = len(false_positive_indices)
|
|
true_negative = len(true_negative_indices)
|
|
metrics = classification_metrics(
|
|
positive_score["true_positive"],
|
|
false_positive,
|
|
positive_score["false_negative"],
|
|
true_negative,
|
|
)
|
|
scored_group_ids = {
|
|
group_id
|
|
for group_ids in negative_hits.values()
|
|
for group_id in group_ids
|
|
} | set(groups) - set(positive_score["unmatched_finding_ids"])
|
|
return {
|
|
**metrics,
|
|
"positive_total": len(positive_entries),
|
|
"negative_total": len(negative_entries),
|
|
"finding_groups": len(groups),
|
|
"matched_gold_indices": positive_score["matched_gold_indices"],
|
|
"missed_gold_indices": positive_score["missed_gold_indices"],
|
|
"missed_gold_cases": [
|
|
positive_entries[index]
|
|
for index in positive_score["missed_gold_indices"]
|
|
],
|
|
"false_positive_case_indices": false_positive_indices,
|
|
"false_positive_cases": [
|
|
{
|
|
**negative_entries[index],
|
|
"finding_ids": negative_hits[index],
|
|
}
|
|
for index in false_positive_indices
|
|
],
|
|
"unscored_finding_ids": sorted(set(groups) - scored_group_ids),
|
|
"metric_defined": metrics["defined"],
|
|
}
|
|
|
|
|
|
def score_findings_by_rule(
|
|
findings: list[dict[str, Any]],
|
|
gold_entries: list[dict[str, Any]],
|
|
negative_entries: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, dict[str, Any]]:
|
|
"""Score each rule independently so large easy classes cannot hide regressions."""
|
|
|
|
rules = {
|
|
str(entry.get("kind", ""))
|
|
for entry in gold_entries
|
|
if str(entry.get("kind", "")) not in {"", "any"}
|
|
}
|
|
rules.update(str(finding.get("kind", "")) for finding in findings if finding.get("kind"))
|
|
rules.update(
|
|
str(entry.get("kind", ""))
|
|
for entry in negative_entries or []
|
|
if str(entry.get("kind", "")) not in {"", "any"}
|
|
)
|
|
scores: dict[str, dict[str, Any]] = {}
|
|
for rule in sorted(rules):
|
|
rule_findings = [
|
|
finding for finding in findings if _kind_matches(rule, {str(finding.get("kind", ""))})
|
|
]
|
|
rule_gold = [entry for entry in gold_entries if str(entry.get("kind", "")) == rule]
|
|
rule_negative = [
|
|
entry
|
|
for entry in negative_entries or []
|
|
if str(entry.get("kind", "")) == rule
|
|
]
|
|
scores[rule] = (
|
|
score_labeled_cases(rule_findings, rule_gold, rule_negative)
|
|
if negative_entries is not None
|
|
else score_grouped_findings(rule_findings, rule_gold)
|
|
)
|
|
return scores
|