docs: validate baseline and add deterministic regression coverage

Session 3 baseline validation (no beta/support-weighted code added).

- Two measured full-data original_protean conditions (X-IIoTID and 5G-NIDD,
  alpha=0.5, seed=0); frozen manifests reused and reload-verified.
- Partition consistency, per-class/per-client inspection, round-0, loss audit,
  inference validation, reproducibility and a deterministic end-to-end test.
- S_j participant rule remains an explicitly labelled REPRODUCTION ASSUMPTION;
  |S_j| < M quantified in 53/162 class-condition cells (32.7%), 15/18 conditions.
- Adds docs/baseline/BASELINE_VALIDATION.md, tests/test_baseline_validation.py,
  scripts/validate_sj_participant_rule.py and small analysis artifacts.
- Test suite: 46 passed.

planningawal/ and raw datasets unchanged.
This commit is contained in:
willtanoe committed 2026-09-24 20:42:56 +00:00
1 parent 374651a0fe
commit 16b3add2cf
9 files changed
+24265 -200

No files matched your search

+165
View File
@@ -0,0 +1,165 @@
#!/usr/bin/env python
"""Read-only validation of the baseline ``S_j`` prototype participant rule.
This is a **baseline-validation analysis**, not an experiment mode. It never
trains, never changes the baseline, and never fabricates prototypes for training.
It answers two questions:
1. Over the frozen partition manifests, how often is ``|S_j| < M`` (i.e. how
often does the paper's literal ``1/M`` become ambiguous), for which classes
and conditions?
2. Given a fixed round snapshot of ``C_ij`` and ``N_ij`` (no retraining), how far
is the current ``S_j`` mean from one hypothetical literal-``1/M`` reading that
replaces missing prototypes with the zero vector?
Outputs (analysis artifacts, clearly marked hypothetical):
- ``artifacts/analysis/sj_participant_rule_analysis.csv``
- ``artifacts/analysis/sj_vs_literal_1m_hypothetical_<dataset>_alpha<..>_seed<..>.csv``
Usage::
python scripts/validate_sj_participant_rule.py
python scripts/validate_sj_participant_rule.py \
--snapshot artifacts/analysis/xiiotid__alpha0.5__seed0__original_protean__prototype_snapshot.json
"""
from __future__ import annotations
import argparse
import csv
import json
from pathlib import Path
import numpy as np
REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_PARTITIONS = REPO_ROOT / "artifacts" / "partitions"
DEFAULT_ANALYSIS = REPO_ROOT / "artifacts" / "analysis"
def summarize_manifests(partition_dir: Path, analysis_dir: Path) -> None:
rows = []
n_cells = n_less = n_conditions = n_conditions_less = 0
for path in sorted(partition_dir.glob("*.json")):
m = json.loads(path.read_text())
support = m["support"]
M = int(m["num_clients"])
names = m["class_names"]
condition_less = 0
for j, name in enumerate(names):
column = [int(support[i][j]) for i in range(M)]
contributors = [i for i, v in enumerate(column) if v > 0]
missing = [i for i, v in enumerate(column) if v == 0]
n_cells += 1
if len(contributors) < M:
n_less += 1
condition_less += 1
rows.append(
{
"dataset": m["dataset"],
"alpha": m["alpha"],
"seed": m["seed"],
"M": M,
"class_id": j,
"class_name": name,
"S_j": len(contributors),
"missing_clients": "|".join(str(i) for i in missing),
"N_contrib_min": min([column[i] for i in contributors]) if contributors else 0,
"N_contrib_max": max([column[i] for i in contributors]) if contributors else 0,
"N_total": sum(column),
}
)
n_conditions += 1
if condition_less:
n_conditions_less += 1
out = analysis_dir / "sj_participant_rule_analysis.csv"
out.parent.mkdir(parents=True, exist_ok=True)
fields = list(rows[0].keys())
with out.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fields)
writer.writeheader()
writer.writerows(rows)
print(f"manifests scanned: {n_conditions}")
print(f"conditions with any |S_j| < M: {n_conditions_less}/{n_conditions}")
print(f"class-condition cells with |S_j| < M: {n_less}/{n_cells} "
f"({100.0 * n_less / n_cells:.1f}%)")
print(f"wrote {out}")
def snapshot_variants(snapshot_path: Path, analysis_dir: Path) -> None:
snap = json.loads(snapshot_path.read_text())
prototypes = snap["client_prototypes"]
supports = snap["client_supports"]
M = len(prototypes)
classes = sorted(
{int(c) for row in prototypes for c in row}
| {int(c) for row in supports for c in row}
)
dim = len(next(iter(prototypes[0].values())))
rows = []
for j in classes:
contributors = [
(i, np.asarray(prototypes[i][str(j)], dtype=np.float64), int(supports[i][str(j)]))
for i in range(M)
if str(j) in prototypes[i]
]
if not contributors:
continue
cbar_sj = np.mean([v for _, v, _ in contributors], axis=0)
# Hypothetical only: literal 1/M with missing prototypes replaced by zeros.
zero_sum = sum((v for _, v, _ in contributors), np.zeros(dim, dtype=np.float64))
cbar_zero = zero_sum / M
rows.append(
{
"class_id": j,
"M": M,
"S_j": len(contributors),
"missing_clients": "|".join(
str(i) for i in range(M) if i not in {i for i, _, _ in contributors}
),
"N_ij_contributors": "|".join(
str(int(supports[i][str(j)])) for i, _, _ in contributors
),
"displacement_Sj_vs_literal_1M_zero": float(np.linalg.norm(cbar_sj - cbar_zero)),
"mean_norm_Cbar_Sj": float(np.linalg.norm(cbar_sj)),
"mean_contributor_norm": float(
np.mean([np.linalg.norm(v) for _, v, _ in contributors])
),
"hypothetical": True,
}
)
tag = snapshot_path.name.replace("__prototype_snapshot.json", "")
out = analysis_dir / f"sj_vs_literal_1m_hypothetical_{tag}.csv"
fields = list(rows[0].keys())
with out.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fields)
writer.writeheader()
writer.writerows(rows)
print(f"snapshot round={snap.get('round')} M={M} dim={dim}; wrote {out}")
def parse_args(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--partitions-dir", default=str(DEFAULT_PARTITIONS))
parser.add_argument("--analysis-dir", default=str(DEFAULT_ANALYSIS))
parser.add_argument("--snapshot", default=None, help="optional round-0 prototype snapshot JSON")
return parser.parse_args(argv)
def main(argv=None) -> int:
args = parse_args(argv)
analysis_dir = Path(args.analysis_dir)
summarize_manifests(Path(args.partitions_dir), analysis_dir)
if args.snapshot:
snapshot_variants(Path(args.snapshot), analysis_dir)
return 0
if __name__ == "__main__":
raise SystemExit(main())