"""Descriptive VALID error slices for the frozen ten-seed character baseline. This script does not select features, thresholds, or models. VALID is used only for subgroup reporting; chronological TEST is never loaded. """ from __future__ import annotations import hashlib import json from pathlib import Path import numpy as np import pandas as pd from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import SGDClassifier ROOT = Path(__file__).resolve().parents[1] CORPUS = ROOT / "data/processed/corpus.parquet" TRAIN = ROOT / "data/splits/chronological/train.json" VALID = ROOT / "data/splits/chronological/valid.json" OUT_JSON = ROOT / "results/phase7/char_baseline_valid_subgroups.json" OUT_MD = ROOT / "results/phase7/char_baseline_valid_subgroups.md" SEEDS = range(10) def wilson(k: int, n: int) -> list[float] | None: if n == 0: return None z = 1.96 p = k / n d = 1 + z * z / n center = p + z * z / (2 * n) half = z * np.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) return [float((center - half) / d), float((center + half) / d)] def metrics(mask: np.ndarray, y: np.ndarray, pred: np.ndarray) -> dict: n = int(mask.sum()) k = int(np.sum((y == pred) & mask)) safe = mask & (y == 0) vuln = mask & (y == 1) ks = int(np.sum((y == pred) & safe)) kv = int(np.sum((y == pred) & vuln)) return { "n_rows": n, "accuracy": k / n if n else None, "wilson_95": wilson(k, n), "safe_n": int(safe.sum()), "safe_accuracy": ks / int(safe.sum()) if safe.any() else None, "safe_wilson_95": wilson(ks, int(safe.sum())), "vulnerable_n": int(vuln.sum()), "vulnerable_accuracy": kv / int(vuln.sum()) if vuln.any() else None, "vulnerable_wilson_95": wilson(kv, int(vuln.sum())), "predicted_vulnerable_fraction": float(pred[mask].mean()) if n else None, } def main() -> None: frame = pd.read_parquet(CORPUS) train_ids = set(map(str, json.loads(TRAIN.read_text()))) valid_ids = set(map(str, json.loads(VALID.read_text()))) if train_ids & valid_ids: raise ValueError("TRAIN/VALID pair overlap") def materialize(ids: set[str]): sub = frame[frame.pair_id.astype(str).isin(ids)] rows = [] for pid, grp in sub.groupby("pair_id", sort=True): if len(grp) != 2 or set(grp.target.astype(int)) != {0, 1}: raise ValueError(f"pair integrity failure: {pid}") if grp.func_vuln.nunique() != 1 or grp.func_patch.nunique() != 1: raise ValueError(f"paired code mismatch: {pid}") if grp.cwe.nunique(dropna=False) != 1: raise ValueError(f"paired CWE mismatch: {pid}") row = grp[grp.target.astype(int) == 1].iloc[0] for target in (0, 1): code_row = grp[grp.target.astype(int) == target].iloc[0] code = str(code_row.func_vuln if target else code_row.func_patch) rows.append({ "pair_id": str(pid), "target": target, "code": code, "cwe": str(row.cwe) if pd.notna(row.cwe) else "UNKNOWN", "project": str(row.project), "vuln_lines": max(1, len(str(row.func_vuln).splitlines())), "patch_lines": max(1, len(str(row.func_patch).splitlines())), "commit_date": str(row.commit_date), }) return pd.DataFrame(rows) tr = materialize(train_ids) va = materialize(valid_ids) if tr.pair_id.nunique() != len(train_ids) or va.pair_id.nunique() != len(valid_ids): raise ValueError("split pair IDs did not materialize one-to-one") vec = TfidfVectorizer(analyzer="char", ngram_range=(3, 5), min_df=2, max_features=300_000, sublinear_tf=True, dtype=np.float32) xtr = vec.fit_transform(tr.code) xva = vec.transform(va.code) ytr = tr.target.to_numpy(dtype=np.int8) y = va.target.to_numpy(dtype=np.int8) scores = [] for seed in SEEDS: clf = SGDClassifier(loss="log_loss", alpha=1e-5, max_iter=200, tol=1e-3, random_state=seed, average=True) clf.fit(xtr, ytr) scores.append(clf.decision_function(xva)) pred = (np.mean(np.stack(scores), axis=0) >= 0).astype(np.int8) cwe_groups = [] for cwe, inds in va.groupby("cwe", sort=True).groups.items(): idx = np.asarray(list(inds), dtype=int) mask = np.zeros(len(va), dtype=bool) mask[idx] = True metric = metrics(mask, y, pred) cwe_groups.append({"cwe": cwe, "n_pairs": int(va.loc[idx, "pair_id"].nunique()), **metric}) # Function-length slices are descriptive quantiles of the VALID vulnerable # function lengths; they are not used to tune or select a model. vuln = va[va.target == 1] cuts = np.unique(np.quantile(vuln.vuln_lines, [0, .25, .5, .75, 1])) length_groups = [] for i in range(max(0, len(cuts) - 1)): lo, hi = cuts[i], cuts[i + 1] pair_sel = (va.vuln_lines >= lo) & ((va.vuln_lines <= hi) if i == len(cuts)-2 else (va.vuln_lines < hi)) mask = pair_sel.to_numpy() observed_lengths = va.loc[pair_sel, "vuln_lines"] length_groups.append({"min_vulnerable_lines": int(observed_lengths.min()), "max_vulnerable_lines": int(observed_lengths.max()), **metrics(mask, y, pred)}) overall = metrics(np.ones(len(va), dtype=bool), y, pred) result = { "method": "descriptive subgroup audit of fixed ten-seed char TF-IDF/SGD ensemble", "valid_pairs": int(va.pair_id.nunique()), "valid_rows": int(len(va)), "overall": overall, "cwe_slices": cwe_groups, "vulnerable_function_length_quartiles": length_groups, "protocol": { "model_seeds": list(SEEDS), "threshold": 0.0, "no_valid_model_or_threshold_selection": True, "test_read": False, "cwe_metrics_are_descriptive": True, "minimum_count_for_interpretation": 30, }, "provenance": { "corpus_sha256": hashlib.sha256(CORPUS.read_bytes()).hexdigest(), "train_split_sha256": hashlib.sha256(TRAIN.read_bytes()).hexdigest(), "valid_split_sha256": hashlib.sha256(VALID.read_bytes()).hexdigest(), }, "limitations": [ "CWE annotations are dataset labels and may be incomplete or noisy.", "Small CWE groups have wide uncertainty and are not suitable for ranking models.", "The VALID slices are descriptive and must not be used to tune model choices.", "Function length quartiles are defined on VALID solely for descriptive reporting.", ], } OUT_JSON.parent.mkdir(parents=True, exist_ok=True) OUT_JSON.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") rows = ["# Character baseline VALID subgroup audit", "", "Descriptive only; no feature, threshold, or model selection used VALID.", "", f"Overall: {overall['accuracy']:.3%} accuracy, Wilson 95% CI " f"{overall['wilson_95'][0]:.3%}–{overall['wilson_95'][1]:.3%}; " f"SAFE {overall['safe_accuracy']:.3%}, VULNERABLE {overall['vulnerable_accuracy']:.3%}.", "", "## CWE slices", "", "| CWE | pairs | SAFE acc. | VULNERABLE acc. | Overall acc. |", "|---|---:|---:|---:|---:|"] for item in cwe_groups: f = lambda x: "—" if x is None else f"{x:.1%}" rows.append(f"| {item['cwe']} | {item['n_pairs']} | {f(item['safe_accuracy'])} | " f"{f(item['vulnerable_accuracy'])} | {f(item['accuracy'])} |") rows += ["", "## Vulnerable-function length slices", "", "| Vulnerable-function lines | pairs | SAFE acc. | VULNERABLE acc. | Overall acc. |", "|---:|---:|---:|---:|---:|"] for item in length_groups: f = lambda x: "—" if x is None else f"{x:.1%}" rows.append(f"| {item['min_vulnerable_lines']}–{item['max_vulnerable_lines']} | " f"{item['n_rows']//2} | {f(item['safe_accuracy'])} | " f"{f(item['vulnerable_accuracy'])} | {f(item['accuracy'])} |") rows += ["", "Small groups are descriptive only; CWE labels may be incomplete/noisy. " "TEST was not read.", ""] OUT_MD.write_text("\n".join(rows)) print(json.dumps({"overall": overall, "cwe_groups": len(cwe_groups), "length_groups": len(length_groups), "artifacts": [str(OUT_JSON), str(OUT_MD)]}, indent=2)) if __name__ == "__main__": main()