| """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}) |
|
|
| |
| |
| 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() |
|
|
