scripts/v06_d_severity_compare.py

"""v0.6 issue A Phase 1 — D-sev diagnostic.

For each evaluable event, compare predicted-node severity bucket
distributions between L2 (v0.5 issue #68 baseline) and L3-B (v0.5 issue
#70 末态). Output table per event × (domain × severity) with
(L2_count, L3_count, delta) triples so we can spot whether severity
drift is uniform / domain-specific / random.

Reads:
- data/evaluation/gold/_v8_archive/{event_id}_seed{seed}.json (L2)
- data/evaluation/gold/{event_id}_seed{seed}.json             (L3-B current)

Writes:
- stdout (table)
- /tmp/v06_d_sev.csv (machine-readable per-event-domain-severity row)

Usage:
    PYTHONPATH=. python scripts/v06_d_severity_compare.py | tee /tmp/v06_d_sev.log
"""

from __future__ import annotations
import csv
import json
from collections import Counter, defaultdict
from pathlib import Path

L2_DIR = Path("data/evaluation/gold/_v8_archive")
L3_DIR = Path("data/evaluation/gold")
OUT_CSV = Path("/tmp/v06_d_sev.csv")


def _load_severity_counts(cache_dir: Path) -> dict[str, Counter]:
    """Return {event_id: Counter((domain, severity))}, averaged across seeds."""
    per_event: dict[str, Counter] = defaultdict(Counter)
    for path in sorted(cache_dir.glob("*_seed*.json")):
        # Filename: {event_id}_seed{seed}.json
        stem = path.stem
        if "_seed" not in stem:
            continue
        event_id = stem.rsplit("_seed", 1)[0]
        try:
            data = json.loads(path.read_text())
        except (json.JSONDecodeError, OSError):
            continue
        chain = data.get("predicted_chain") or {}
        for node in chain.get("cascade_events") or []:
            per_event[event_id][(node.get("domain", "?"), node.get("severity", "?"))] += 1
    return per_event


def compare(l2_dir: Path = L2_DIR, l3_dir: Path = L3_DIR) -> None:
    l2 = _load_severity_counts(l2_dir)
    l3 = _load_severity_counts(l3_dir)

    all_events = sorted(set(l2) | set(l3))
    rows = []

    print(f"\n=== D-sev: predicted (domain × severity) bucket counts (L2 vs L3-B) ===")
    print(f"{'event_id':<18} {'domain':<22} {'severity':<12} {'L2':>4} {'L3':>4} {'Δ':>5}")

    for eid in all_events:
        keys = sorted(set(l2.get(eid, Counter())) | set(l3.get(eid, Counter())))
        for (dom, sev) in keys:
            c2 = l2.get(eid, Counter())[(dom, sev)]
            c3 = l3.get(eid, Counter())[(dom, sev)]
            delta = c3 - c2
            print(f"{eid:<18} {dom:<22} {sev:<12} {c2:>4} {c3:>4} {delta:>+5}")
            rows.append({
                "event_id": eid, "domain": dom, "severity": sev,
                "l2_count": c2, "l3_count": c3, "delta": delta,
            })

    if rows:
        with OUT_CSV.open("w", newline="", encoding="utf-8") as fh:
            writer = csv.DictWriter(fh, fieldnames=list(rows[0].keys()))
            writer.writeheader()
            writer.writerows(rows)
        print(f"\nWrote {len(rows)} rows to {OUT_CSV}")


def main() -> None:
    compare()


if __name__ == "__main__":
    main()