research: add post-hoc statistical audit and defense preparation aids
- scripts/posthoc_statistical_audit.py + docs/research/POSTHOC_STATISTICAL_AUDIT.md: read-only paired audit of the completed grid (exact sign-flip / Wilcoxon, selection stability, per-class/per-client summaries); writes only posthoc__*. - kisikisithesis/: robustness audit and examiner-preparation document with deterministic read-only checks (robustness_checks.py). - artifacts/analysis/posthoc__*.csv: audit outputs (6 tables). - artifacts/analysis: track the 4 remaining beta0-snapshot effect/weight CSVs (xiiotid a0.5 seeds 1-2), completing the tracked beta0 set. - .gitignore: codify that non-beta0 prototype snapshots stay untracked.
This commit is contained in:
willtanoe committed
1 parent
34bab45865
commit
111adb272e
15 files changed
+2431
No files matched your search
@@ -0,0 +1,313 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Post-hoc statistical audit of the completed PROTEAN experiment grid.
|
||||
|
||||
Read-only post-hoc analysis: no training, no partition regeneration, no
|
||||
modification of any existing artifact. Reads the audited comparison tables and
|
||||
per-run metadata under ``results/`` and writes NEW derived tables with unique
|
||||
``posthoc__`` names under ``artifacts/analysis/``.
|
||||
|
||||
Computes:
|
||||
* paired global per-seed differences (original_protean vs selected-beta
|
||||
support_weighted) and condition-level descriptive summaries;
|
||||
* exploratory paired tests (exact sign-flip permutation, exact Wilcoxon
|
||||
signed-rank, paired t 95% CI) with explicit small-n limitations;
|
||||
* validation beta-selection stability (pooled vs per-seed);
|
||||
* per-class and per-client descriptive summaries.
|
||||
|
||||
Run:
|
||||
.venv/bin/python scripts/posthoc_statistical_audit.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import itertools
|
||||
import json
|
||||
import math
|
||||
import statistics
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
ANALYSIS = ROOT / "artifacts" / "analysis"
|
||||
RESULTS = ROOT / "results"
|
||||
|
||||
DATASETS = ["xiiotid", "5gnidd"]
|
||||
ALPHAS = [0.25, 0.50, 0.75]
|
||||
SEEDS = [0, 1, 2]
|
||||
METRICS = ["rare_class_macro_f1", "rare_class_recall", "macro_f1",
|
||||
"weighted_f1", "accuracy"]
|
||||
SELECTED = {("xiiotid", 0.25): 0.0, ("xiiotid", 0.5): 0.75, ("xiiotid", 0.75): 0.5,
|
||||
("5gnidd", 0.25): 0.0, ("5gnidd", 0.5): 1.0, ("5gnidd", 0.75): 0.5}
|
||||
T_ZERO = 1e-6 # numerical "effectively zero" for sign counts
|
||||
T_PRACTICAL = 0.02 # practical negligible band (matches deep comparison)
|
||||
T975_DF2 = 4.302652729 # t_{0.975, df=2}
|
||||
|
||||
|
||||
def read_csv(p: Path) -> list[dict]:
|
||||
with p.open() as fh:
|
||||
return list(csv.DictReader(fh))
|
||||
|
||||
|
||||
def write_csv(name: str, rows: list[dict]) -> None:
|
||||
# Guard: this script may only ever write its own posthoc__ outputs, and may
|
||||
# overwrite them so the analysis is exactly reproducible on rerun. It can
|
||||
# never touch any pre-existing analysis artifact.
|
||||
if not name.startswith("posthoc__"):
|
||||
raise SystemExit(f"refusing to write non-posthoc artifact: {name}")
|
||||
path = ANALYSIS / name
|
||||
with path.open("w", newline="") as fh:
|
||||
w = csv.DictWriter(fh, fieldnames=list(rows[0].keys()))
|
||||
w.writeheader()
|
||||
w.writerows(rows)
|
||||
print("wrote", path)
|
||||
|
||||
|
||||
def mean(v):
|
||||
return statistics.mean(v)
|
||||
|
||||
|
||||
def std_sample(v):
|
||||
return statistics.stdev(v) if len(v) > 1 else 0.0
|
||||
|
||||
|
||||
def median(v):
|
||||
return statistics.median(v)
|
||||
|
||||
|
||||
# --- exploratory exact tests ------------------------------------------------
|
||||
def signflip_p(deltas):
|
||||
"""Exact two-sided sign-flip permutation test on the mean (n<=~20)."""
|
||||
d = list(deltas)
|
||||
obs = abs(sum(d))
|
||||
if not d or all(x == 0 for x in d):
|
||||
return 1.0
|
||||
total = 0
|
||||
ge = 0
|
||||
for signs in itertools.product((1, -1), repeat=len(d)):
|
||||
s = abs(sum(sign * x for sign, x in zip(signs, d)))
|
||||
total += 1
|
||||
if s >= obs - 1e-12:
|
||||
ge += 1
|
||||
return ge / total
|
||||
|
||||
|
||||
def wilcoxon_exact_p(deltas):
|
||||
"""Exact two-sided Wilcoxon signed-rank (min-rank convention), zeros dropped."""
|
||||
d = [x for x in deltas if abs(x) > T_ZERO]
|
||||
n = len(d)
|
||||
if n == 0:
|
||||
return 1.0
|
||||
order = sorted(range(n), key=lambda i: abs(d[i]))
|
||||
ranks = [0.0] * n
|
||||
i = 0
|
||||
while i < n:
|
||||
j = i
|
||||
while j + 1 < n and abs(d[order[j + 1]]) == abs(d[order[i]]):
|
||||
j += 1
|
||||
avg = (i + j) / 2 + 1
|
||||
for k in range(i, j + 1):
|
||||
ranks[order[k]] = avg
|
||||
i = j + 1
|
||||
w_pos = sum(ranks[i] for i in range(n) if d[i] > 0)
|
||||
w_min_obs = min(w_pos, sum(ranks) - w_pos)
|
||||
dist = defaultdict(int)
|
||||
for signs in itertools.product((1, -1), repeat=n):
|
||||
wp = sum(ranks[i] for i in range(n) if signs[i] > 0)
|
||||
dist[min(wp, sum(ranks) - wp)] += 1
|
||||
le = sum(c for w, c in dist.items() if w <= w_min_obs + 1e-12)
|
||||
return min(1.0, 2 * le / (2 ** n))
|
||||
|
||||
|
||||
def paired_t_ci(deltas):
|
||||
d = list(deltas)
|
||||
n = len(d)
|
||||
m = mean(d)
|
||||
s = std_sample(d)
|
||||
half = T975_DF2 * s / math.sqrt(n) if n > 1 else 0.0
|
||||
return m - half, m + half
|
||||
|
||||
|
||||
# --- pairing verification ---------------------------------------------------
|
||||
def manifests(ds, alpha, seed, beta):
|
||||
o = json.loads((RESULTS / f"{ds}__alpha{alpha:g}__seed{seed}__original_protean"
|
||||
/ "metadata.json").read_text())
|
||||
s = json.loads((RESULTS / f"{ds}__alpha{alpha:g}__seed{seed}__support_weighted__beta{beta:g}"
|
||||
/ "metadata.json").read_text())
|
||||
return o, s
|
||||
|
||||
|
||||
def main() -> None:
|
||||
per_seed = read_csv(ANALYSIS / "deepcmp__global_per_seed.csv")
|
||||
|
||||
# 1. Verify pairing and selected beta, rebuild a per-seed table. ----------
|
||||
rebuilt = []
|
||||
pair_ok = True
|
||||
for ds in DATASETS:
|
||||
for alpha in ALPHAS:
|
||||
beta = SELECTED[(ds, alpha)]
|
||||
for seed in SEEDS:
|
||||
o, s = manifests(ds, alpha, seed, beta)
|
||||
ok = (o["dataset"] == s["dataset"] == ds
|
||||
and abs(float(o["alpha"]) - alpha) < 1e-9
|
||||
and int(o["seed"]) == int(s["seed"]) == seed
|
||||
and o["manifest_content_hash"] == s["manifest_content_hash"]
|
||||
and abs(float(s["beta"]) - beta) < 1e-9)
|
||||
pair_ok = pair_ok and ok
|
||||
for m in METRICS:
|
||||
row = next(r for r in per_seed
|
||||
if r["dataset"] == ds and abs(float(r["alpha"]) - alpha) < 1e-9
|
||||
and int(r["seed"]) == seed and r["metric"] == m)
|
||||
rebuilt.append({
|
||||
"dataset": ds, "alpha": alpha, "seed": seed,
|
||||
"selected_beta": beta, "metric": m,
|
||||
"original": row["original"], "proposed": row["proposed"],
|
||||
"delta": row["delta"],
|
||||
"manifest_content_hash": o["manifest_content_hash"],
|
||||
"pairing_verified": ok})
|
||||
if not pair_ok:
|
||||
raise SystemExit("pairing verification FAILED")
|
||||
print("pairing verified for all 18 condition-seed cells (manifest hash match)")
|
||||
write_csv("posthoc__paired_global_per_seed.csv", rebuilt)
|
||||
|
||||
# 2. Condition-level descriptive summary. -------------------------------
|
||||
by_cond = []
|
||||
tests = []
|
||||
for ds in DATASETS:
|
||||
for alpha in ALPHAS:
|
||||
beta = SELECTED[(ds, alpha)]
|
||||
for m in METRICS:
|
||||
rows = [r for r in rebuilt if r["dataset"] == ds
|
||||
and abs(r["alpha"] - alpha) < 1e-9 and r["metric"] == m]
|
||||
base = [float(r["original"]) for r in rows]
|
||||
prop = [float(r["proposed"]) for r in rows]
|
||||
dlt = [float(r["delta"]) for r in rows]
|
||||
npos = sum(1 for d in dlt if d > T_ZERO)
|
||||
nneg = sum(1 for d in dlt if d < -T_ZERO)
|
||||
nzero = sum(1 for d in dlt if abs(d) <= T_ZERO)
|
||||
by_cond.append({
|
||||
"dataset": ds, "alpha": alpha, "selected_beta": beta, "metric": m,
|
||||
"original_seed0": base[0], "original_seed1": base[1], "original_seed2": base[2],
|
||||
"proposed_seed0": prop[0], "proposed_seed1": prop[1], "proposed_seed2": prop[2],
|
||||
"delta_seed0": dlt[0], "delta_seed1": dlt[1], "delta_seed2": dlt[2],
|
||||
"delta_mean": mean(dlt), "delta_std_sample": std_sample(dlt),
|
||||
"delta_median": median(dlt), "delta_min": min(dlt), "delta_max": max(dlt),
|
||||
"baseline_std_sample": std_sample(base),
|
||||
"proposed_std_sample": std_sample(prop),
|
||||
"n_pos": npos, "n_neg": nneg, "n_zero": nzero,
|
||||
"same_sign_all_seeds": (npos == 3 or nneg == 3)})
|
||||
lo, hi = paired_t_ci(dlt)
|
||||
tests.append({
|
||||
"dataset": ds, "alpha": alpha, "selected_beta": beta, "metric": m,
|
||||
"n_pairs": len(dlt),
|
||||
"deltas": "|".join(f"{d:.6f}" for d in dlt),
|
||||
"signflip_exact_p_two_sided": signflip_p(dlt),
|
||||
"wilcoxon_exact_p_two_sided": wilcoxon_exact_p(dlt),
|
||||
"paired_t_ci95_low": lo, "paired_t_ci95_high": hi,
|
||||
"min_attainable_signflip_p": 2 / (2 ** len(dlt)),
|
||||
"note": ("exploratory; 3 seeds; no inferential power; "
|
||||
"beta=0 conditions are numerical, not a treatment contrast")})
|
||||
write_csv("posthoc__paired_global_by_condition.csv", by_cond)
|
||||
write_csv("posthoc__paired_test_results.csv", tests)
|
||||
|
||||
# 3. Beta-selection stability. -----------------------------------------
|
||||
stability = []
|
||||
for ds in DATASETS:
|
||||
for alpha in ALPHAS:
|
||||
vs = read_csv(ANALYSIS / f"{ds}__alpha{alpha:g}__multiseed__validation_per_seed.csv")
|
||||
scores = defaultdict(dict) # seed -> beta -> score
|
||||
for r in vs:
|
||||
scores[int(r["seed"])][float(r["beta"])] = float(r["val_rare_macro_f1_per_client_mean"])
|
||||
per_seed_beta = {s: max(scores[s], key=lambda b: (scores[s][b], -b)) for s in scores}
|
||||
pooled = {b: mean([scores[s][b] for s in SEEDS]) for b in sorted(next(iter(scores.values())))}
|
||||
pooled_beta = max(pooled, key=lambda b: (pooled[b], -b))
|
||||
ordered = sorted(pooled.values(), reverse=True)
|
||||
agree = sum(1 for s in SEEDS if abs(per_seed_beta[s] - pooled_beta) < 1e-9)
|
||||
second_beta = sorted((b for b in pooled if abs(b - pooled_beta) > 1e-9),
|
||||
key=lambda b: -pooled[b])[0]
|
||||
stability.append({
|
||||
"dataset": ds, "alpha": alpha,
|
||||
"pooled_selected_beta": pooled_beta,
|
||||
"per_seed_beta_s0": per_seed_beta[0],
|
||||
"per_seed_beta_s1": per_seed_beta[1],
|
||||
"per_seed_beta_s2": per_seed_beta[2],
|
||||
"n_seeds_agree_pooled": agree,
|
||||
"pooled_score_selected": pooled[pooled_beta],
|
||||
"second_best_beta": second_beta,
|
||||
"second_best_score": pooled[second_beta],
|
||||
"margin_best_minus_second": pooled[pooled_beta] - pooled[second_beta],
|
||||
"candidate_score_range": ordered[0] - ordered[-1],
|
||||
"test_used_for_selection": False})
|
||||
write_csv("posthoc__selection_stability.csv", stability)
|
||||
|
||||
# 4. Per-class summary. --------------------------------------------------
|
||||
pc = read_csv(ANALYSIS / "deepcmp__per_class.csv")
|
||||
grp = defaultdict(list)
|
||||
for r in pc:
|
||||
grp[(r["dataset"], r["class_name"])].append(float(r["delta_f1"]))
|
||||
per_class = []
|
||||
for (ds, name), vals in sorted(grp.items(), key=lambda kv: -mean(kv[1])):
|
||||
per_class.append({
|
||||
"dataset": ds, "class_name": name, "n_obs": len(vals),
|
||||
"mean_delta_f1": mean(vals), "median_delta_f1": median(vals),
|
||||
"min_delta_f1": min(vals), "max_delta_f1": max(vals),
|
||||
"n_positive": sum(1 for v in vals if v > 0),
|
||||
"n_negative": sum(1 for v in vals if v < 0)})
|
||||
write_csv("posthoc__per_class_summary.csv", per_class)
|
||||
|
||||
# 5. Per-client summary (descriptive only). ------------------------------
|
||||
cl = read_csv(ANALYSIS / "deepcmp__per_client.csv")
|
||||
by_condition = defaultdict(list)
|
||||
for r in cl:
|
||||
by_condition[(r["dataset"], float(r["alpha"]))].append(r)
|
||||
per_client = []
|
||||
for (ds, alpha), rows in sorted(by_condition.items()):
|
||||
d = [float(r["delta_rare_f1"]) for r in rows]
|
||||
per_client.append({
|
||||
"dataset": ds, "alpha": alpha, "selected_beta": SELECTED[(ds, alpha)],
|
||||
"n_client_seed_obs": len(d),
|
||||
"n_improved_gt_0.02": sum(1 for x in d if x > T_PRACTICAL),
|
||||
"n_degraded_lt_-0.02": sum(1 for x in d if x < -T_PRACTICAL),
|
||||
"n_unchanged": sum(1 for x in d if abs(x) <= T_PRACTICAL),
|
||||
"mean_delta": mean(d), "median_delta": median(d),
|
||||
"min_delta": min(d), "max_delta": max(d),
|
||||
"std_sample_within_condition": std_sample(d)})
|
||||
write_csv("posthoc__per_client_summary.csv", per_client)
|
||||
|
||||
# --- console summary ----------------------------------------------------
|
||||
print("\n=== paired rare-class macro-F1 (delta = proposed - original) ===")
|
||||
for r in by_cond:
|
||||
if r["metric"] != "rare_class_macro_f1":
|
||||
continue
|
||||
print(f" {r['dataset']:8} a{r['alpha']:<4} beta={r['selected_beta']:<4} "
|
||||
f"mean={float(r['delta_mean']):+.4f} std={float(r['delta_std_sample']):.4f} "
|
||||
f"base_std={float(r['baseline_std_sample']):.4f} prop_std={float(r['proposed_std_sample']):.4f} "
|
||||
f"pos/neg/zero={r['n_pos']}/{r['n_neg']}/{r['n_zero']} same_sign={r['same_sign_all_seeds']}")
|
||||
print("\n=== exact tests (rare-class macro-F1) ===")
|
||||
for r in tests:
|
||||
if r["metric"] != "rare_class_macro_f1":
|
||||
continue
|
||||
print(f" {r['dataset']:8} a{r['alpha']:<4} beta={r['selected_beta']:<4} "
|
||||
f"signflip_p={r['signflip_exact_p_two_sided']:.4f} "
|
||||
f"wilcoxon_p={r['wilcoxon_exact_p_two_sided']:.4f} "
|
||||
f"tCI=[{r['paired_t_ci95_low']:+.4f},{r['paired_t_ci95_high']:+.4f}] "
|
||||
f"(min signflip p={r['min_attainable_signflip_p']:.4f})")
|
||||
print("\n=== selection stability ===")
|
||||
for r in stability:
|
||||
print(f" {r['dataset']:8} a{r['alpha']:<4} pooled={r['pooled_selected_beta']:<4} "
|
||||
f"per-seed={r['per_seed_beta_s0']}/{r['per_seed_beta_s1']}/{r['per_seed_beta_s2']} "
|
||||
f"agree={r['n_seeds_agree_pooled']}/3 margin={float(r['margin_best_minus_second']):.4f} "
|
||||
f"range={float(r['candidate_score_range']):.4f}")
|
||||
print("\n=== S_j audit ===")
|
||||
sj = read_csv(ANALYSIS / "sj_participant_rule_analysis.csv")
|
||||
lt = sum(1 for r in sj if int(r["S_j"]) < int(r["M"]))
|
||||
print(f" |S_j| < M cells: {lt} / {len(sj)}")
|
||||
print("\n=== per-class extremes (mean delta F1) ===")
|
||||
print(" top +:", ", ".join(f"{r['dataset']}/{r['class_name']} {float(r['mean_delta_f1']):+.4f}"
|
||||
for r in per_class[:4]))
|
||||
print(" top -:", ", ".join(f"{r['dataset']}/{r['class_name']} {float(r['mean_delta_f1']):+.4f}"
|
||||
for r in per_class[-4:]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user