File size: 8,838 Bytes
646ccfc | 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 | """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()
|