File size: 8,265 Bytes
31082cc | 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 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | """Select a deterministic, diverse panel of errors for the final TFM analysis."""
from __future__ import annotations
import argparse
import json
import re
from collections import Counter
from pathlib import Path
import pandas as pd
def _model_name(path: Path) -> str:
value = str(path).lower()
if "b2_1_random_forest" in value:
return "B2.1 Random Forest"
if "b2_2_xgboost" in value:
return "B2.2 XGBoost"
if "baseline3" in value:
return "B3 EdgeMLP"
if "graphsage" in value:
return "GraphSAGE"
raise ValueError(f"Cannot infer model from prediction path: {path}")
def _seed(path: Path) -> int:
match = re.search(r"seed[-_](\d+)", str(path))
if match:
return int(match.group(1))
raise ValueError(f"Prediction path does not identify its seed: {path}")
def _range_pattern(value: str) -> str:
clauses = [clause.strip() for clause in value.split(",") if clause.strip()]
has_lower = any(clause.startswith((">", ">=")) for clause in clauses)
has_upper = any(clause.startswith(("<", "<=")) for clause in clauses)
if any(clause.startswith(("==", "===")) for clause in clauses):
return "exact_or_wildcard"
if has_lower and has_upper:
return "bounded"
if has_lower:
return "lower_only"
if has_upper:
return "upper_only"
if any(clause.startswith("~=") for clause in clauses):
return "compatible_release"
return "other"
def _persistence_level(error_count: int) -> str:
"""Classify recurrence across the 40 comparable model/seed runs."""
if error_count >= 30:
return "high"
if error_count >= 15:
return "medium"
return "low"
def _prediction_paths(roots: list[Path]) -> list[Path]:
paths: list[Path] = []
for root in roots:
paths.extend(root.glob("**/predictions/*.parquet"))
return sorted(set(paths))
def _load_runs(paths: list[Path]) -> pd.DataFrame:
frames: list[pd.DataFrame] = []
required = {"pair_id", "label", "risk_incompatible", "prediction"}
for path in paths:
frame = pd.read_parquet(path)
missing = required.difference(frame.columns)
if missing:
raise ValueError(f"{path} is missing columns: {sorted(missing)}")
if len(frame) != 1_200 or frame["pair_id"].duplicated().any():
raise ValueError(f"Unexpected test prediction contract in {path}")
frame = frame[list(required)].copy()
frame["model"] = _model_name(path)
frame["seed"] = _seed(path)
frames.append(frame)
if not frames:
raise ValueError("No prediction files were found")
result = pd.concat(frames, ignore_index=True)
result["error_kind"] = ""
result.loc[(result["label"] == 0) & (result["prediction"] == 1), "error_kind"] = "FP"
result.loc[(result["label"] == 1) & (result["prediction"] == 0), "error_kind"] = "FN"
return result[result["error_kind"] != ""].copy()
def _aggregate_errors(errors: pd.DataFrame, dataset: pd.DataFrame) -> pd.DataFrame:
grouped = errors.groupby(["pair_id", "error_kind", "label"], as_index=False).agg(
error_count=("prediction", "size"),
mean_risk=("risk_incompatible", "mean"),
min_risk=("risk_incompatible", "min"),
max_risk=("risk_incompatible", "max"),
models=("model", lambda values: ", ".join(sorted(set(values)))),
seeds=("seed", lambda values: ", ".join(str(value) for value in sorted(set(values)))),
)
test = dataset.loc[dataset["split"] == "test"].copy()
if len(test) != 1_200 or test["pair_id"].duplicated().any():
raise ValueError("Dataset test split does not contain 1,200 unique pairs")
merged = grouped.merge(test, on="pair_id", how="left", validate="one_to_one", suffixes=("", "_data"))
if merged["pa"].isna().any():
raise ValueError("Some prediction pair IDs do not exist in the dataset")
if not (merged["label"] == merged["label_data"]).all():
raise ValueError("Prediction labels disagree with the dataset")
merged = merged.drop(columns=["label_data"])
merged["package_pair"] = merged["pa"] + " -> " + merged["pb"]
merged["range_pattern"] = merged["declared_range"].astype(str).map(_range_pattern)
merged["decision_impact"] = merged["error_kind"].map(
{
"FN": "high: incompatible pair accepted",
"FP": "moderate: compatible pair rejected",
}
)
merged["persistence"] = merged["error_count"].map(_persistence_level)
merged["temporal_gap_days"] = (
pd.to_datetime(merged["va_upload_time"], utc=True)
- pd.to_datetime(merged["vb_upload_time"], utc=True)
).abs().dt.total_seconds() / 86_400
return merged
def _select_diverse(frame: pd.DataFrame, kind: str, size: int, max_per_parent: int) -> pd.DataFrame:
candidates = frame.loc[frame["error_kind"] == kind].copy()
ascending_risk = kind == "FN"
candidates = candidates.sort_values(
["error_count", "mean_risk", "pair_id"],
ascending=[False, ascending_risk, True],
kind="mergesort",
)
selected: list[int] = []
parent_counts: Counter[str] = Counter()
for index, row in candidates.iterrows():
if parent_counts[row["pa"]] >= max_per_parent:
continue
selected.append(index)
parent_counts[row["pa"]] += 1
if len(selected) == size:
break
if len(selected) < size:
for index in candidates.index:
if index not in selected:
selected.append(index)
if len(selected) == size:
break
if len(selected) < size:
raise ValueError(f"Only {len(selected)} distinct {kind} cases are available; requested {size}")
result = candidates.loc[selected].copy()
result["selection_rank"] = range(1, len(result) + 1)
return result
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--dataset", type=Path, required=True)
parser.add_argument("--prediction-root", type=Path, action="append", required=True)
parser.add_argument("--output-dir", type=Path, required=True)
parser.add_argument("--per-kind", type=int, default=30)
parser.add_argument("--max-per-parent", type=int, default=5)
args = parser.parse_args()
dataset = pd.read_parquet(args.dataset)
paths = _prediction_paths(args.prediction_root)
errors = _load_runs(paths)
aggregate = _aggregate_errors(errors, dataset)
panel = pd.concat(
[
_select_diverse(aggregate, "FP", args.per_kind, args.max_per_parent),
_select_diverse(aggregate, "FN", args.per_kind, args.max_per_parent),
],
ignore_index=True,
)
args.output_dir.mkdir(parents=True, exist_ok=True)
aggregate.to_parquet(args.output_dir / "all_errors.parquet", index=False)
panel.to_parquet(args.output_dir / "selected_60.parquet", index=False)
panel.to_csv(args.output_dir / "selected_60.csv", index=False)
summary = {
"dataset": str(args.dataset),
"prediction_files": [str(path) for path in paths],
"runs": int(len(paths)),
"models": sorted(errors["model"].unique()),
"seeds": sorted(int(seed) for seed in errors["seed"].unique()),
"error_events": int(len(errors)),
"distinct_error_pairs": int(aggregate["pair_id"].nunique()),
"selected": panel["error_kind"].value_counts().sort_index().to_dict(),
"decision_impact": panel["decision_impact"].value_counts().sort_index().to_dict(),
"persistence": panel["persistence"].value_counts().sort_index().to_dict(),
"selection_rule": {
"primary": "descending error frequency across model/seed runs",
"secondary_fp": "descending mean predicted risk",
"secondary_fn": "ascending mean predicted risk",
"parent_cap": args.max_per_parent,
"tie_break": "pair_id ascending",
},
}
(args.output_dir / "summary.json").write_text(
json.dumps(summary, indent=2, ensure_ascii=False, sort_keys=True), encoding="utf-8"
)
print(json.dumps(summary, indent=2, ensure_ascii=False, sort_keys=True))
if __name__ == "__main__":
main()
|