thesis-prototype-agregation/kisikisithesis/robustness_checks.py

#!/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|<M classes ({len(partial)}): {md:.4f}")
    if full:
        md = sum(by_class[c][1.0]["displacement_from_beta0"] for c in full) / len(full)
        print(f"  mean disp@1.0, |S_j|=M classes ({len(full)}): {md:.4f}")


def main() -> 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()