#!/usr/bin/env python3 """Deterministic, read-only robustness checks for the PROTEAN thesis. Run from the repository root: .venv/bin/python kisikisithesis/robustness_checks.py Nothing is written and no experimental artifact is modified. The script: * exercises the *actual* aggregation implementation on tiny synthetic inputs (formula invariants); * recomputes direct aggregation effects on persisted round-0 prototype snapshots (no training, no snapshot mutation). All values are printed deterministically so the thesis report can quote them. """ from __future__ import annotations import json import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from protean_research.aggregation.prototype_aggregation import ( # noqa: E402 AggregationMode, aggregate_prototypes, aggregate_class_prototypes, normalized_support_weights, validate_beta, BETA_CANDIDATES, ) from protean_research.analysis.direct_prototype import ( # noqa: E402 client_weight_rows, prototype_effect_rows, ) ANALYSIS = ROOT / "artifacts" / "analysis" def section(title: str) -> None: print("\n" + "=" * 74) print(title) print("=" * 74) def check(label: str, ok: bool) -> None: print(f" [{'PASS' if ok else 'FAIL'}] {label}") def synthetic_invariants() -> None: section("A. Synthetic formula invariants (actual implementation)") p = [[1.0, 0.0], [0.0, 2.0], [3.0, 3.0]] s = [5.0, 10.0, 20.0] eq = [sum(v[d] for v in p) / len(p) for d in range(2)] b0 = aggregate_prototypes(p, AggregationMode.SUPPORT_WEIGHTED, supports=s, beta=0.0) print(f" beta=0 aggregate = {b0}") print(f" arithmetic mean = {eq}") check("beta=0 equals the arithmetic mean over S_j", all(abs(a - b) < 1e-12 for a, b in zip(b0, eq))) w1 = normalized_support_weights(s, 1.0) b1 = aggregate_prototypes(p, AggregationMode.SUPPORT_WEIGHTED, supports=s, beta=1.0) closed = [sum(w * v[d] for w, v in zip(w1, p)) for d in range(2)] print(f" beta=1 weights = {[round(w, 6) for w in w1]}") print(f" beta=1 aggregate = {b1}") check("beta=1 equals the support-weighted closed form", all(abs(a - b) < 1e-12 for a, b in zip(b1, closed))) w05 = normalized_support_weights(s, 0.5) b05 = aggregate_prototypes(p, AggregationMode.SUPPORT_WEIGHTED, supports=s, beta=0.5) print(f" beta=0.5 weights = {[round(w, 6) for w in w05]}") print(f" beta=0.5 aggregate = {b05}") single = aggregate_prototypes([[7.0, 9.0]], AggregationMode.SUPPORT_WEIGHTED, supports=[3.0], beta=1.0) check("single contributor is beta-invariant", single == [7.0, 9.0]) equal = [aggregate_prototypes(p, AggregationMode.SUPPORT_WEIGHTED, supports=[4.0, 4.0, 4.0], beta=b) for b in BETA_CANDIDATES] check("equal support is beta-invariant", all(g == equal[0] for g in equal)) for b in BETA_CANDIDATES: tot = sum(normalized_support_weights(s, b)) if abs(tot - 1.0) > 1e-12: check(f"weights sum to 1 at beta={b}", False) break else: check("normalized weights sum to 1 for every beta candidate", True) # zero-support exclusion and missing-class handling through the S_j builder contributions = {0: [([1.0], 5.0), ([2.0], 0.0), (None, 7.0)], 1: [(None, 0.0)]} agg = aggregate_class_prototypes(contributions, AggregationMode.SUPPORT_WEIGHTED, beta=1.0) print(f" S_j builder output = {agg}") check("N_ij=0 and missing prototypes are dropped from S_j", agg.get(0) == [1.0]) check("a class with no valid contributor is omitted", 1 not in agg) dom = normalized_support_weights([1.0, 1000.0], 1.0) print(f" dominance weights [1,1000] at beta=1 = {[round(w, 6) for w in dom]}") check("a single dominant client can dominate the aggregate at beta=1", dom[1] > 0.999) rejected = 0 for bad in (-0.1, 1.1, None, "x", float("inf")): try: validate_beta(bad) except (ValueError, TypeError): rejected += 1 check("invalid beta values are rejected", rejected == 5) def snapshot_sensitivity(tag: str, path: Path, names: dict) -> None: section(f"B. Direct aggregation effect on a persisted snapshot: {tag}") snap = json.loads(path.read_text()) print(f" snapshot: {path.relative_to(ROOT)} (round {snap['round']})") rows = prototype_effect_rows(snap, betas=[0.0, 0.5, 1.0], class_names=names) weights = client_weight_rows(snap, betas=[0.0, 0.25, 0.5, 0.75, 1.0], class_names=names) by_class = {} for r in rows: by_class.setdefault(r["class_id"], {})[r["beta"]] = r maxw = {} for w in weights: if w["beta"] == 1.0: cur = maxw.get(w["class_id"]) if cur is None or w["weight"] > cur["weight"]: maxw[w["class_id"]] = w print(f" {'class':16} {'|S_j|':>5} {'sup_min':>8} {'sup_max':>9} " f"{'disp@0.5':>9} {'disp@1.0':>9} {'maxW@1.0':>9}") for cid in sorted(by_class): r = by_class[cid] n = r[0.0]["n_contributors"] d05 = r[0.5]["displacement_from_beta0"] d10 = r[1.0]["displacement_from_beta0"] mw = maxw.get(cid, {}).get("weight", float("nan")) name = (r[0.0]["class_name"] or cid) print(f" {str(name):16} {n:>5} {r[0.0]['support_min']:>8.1f} " f"{r[0.0]['support_max']:>9.1f} {d05:>9.4f} {d10:>9.4f} {mw:>9.4f}") partial = [cid for cid in by_class if by_class[cid][0.0]["n_contributors"] < 10] full = [cid for cid in by_class if by_class[cid][0.0]["n_contributors"] == 10] if partial: md = sum(by_class[c][1.0]["displacement_from_beta0"] for c in partial) / len(partial) print(f" mean disp@1.0, |S_j| None: synthetic_invariants() snapshot_sensitivity( "5G-NIDD, alpha=0.25, seed=0 (beta=0 snapshot)", ANALYSIS / "5gnidd__alpha0.25__seed0__support_weighted__beta0__prototype_snapshot.json", {i: n for i, n in enumerate( ["Benign", "HTTPFlood", "ICMPFlood", "SYNFlood", "SYNScan", "SlowrateDoS", "TCPConnectScan", "UDPFlood", "UDPScan"])}) snapshot_sensitivity( "X-IIoTID, alpha=0.25, seed=0 (beta=0 snapshot)", ANALYSIS / "xiiotid__alpha0.25__seed0__support_weighted__beta0__prototype_snapshot.json", {i: n for i, n in enumerate( ["C&C", "Exfiltration", "Exploitation", "Lateral_movement", "Normal", "RDOS", "Reconnaissance", "Tampering", "Weaponization"])}) print("\nREAD-ONLY CHECKS COMPLETE (no files written, no snapshots modified).") if __name__ == "__main__": main()