research/eval_summary.py

"""Recompute the honest battery scorecard from a per-battery eval log.

Scores are taken from the persisted per-probe lines so a device kill that
restarts eval.py mid-battery never loses or re-aggregates completed probes.
Exact-match on eval_labels.CANON, same protocol as research/eval.py.

Usage:
  .venv/bin/python research/eval_summary.py logs/eval_cand_sft_best_main.log
"""

import argparse
import re
from pathlib import Path

from research import eval_labels as EL

LINE = re.compile(r"^\[(\S+)\] ([\d.]+|qual) \| verdict: (.+?) \| conf: (\S+)$")


def scored_ids(log_path):
    """Probe ids already scored in ANY section of a battery log (for resume)."""
    ids = set()
    if not Path(log_path).exists():
        return ids
    for ln in Path(log_path).read_text().splitlines():
        m = LINE.match(ln.strip())
        if m:
            ids.add(m.group(1))
    return ids


def cat_of(pid):
    if pid.startswith("p") or pid.startswith("rt"):
        return "generic"
    return pid.rsplit("-", 1)[0]


def summarize(log_path):
    scores, quals, formats = [], 0, 0
    cat_acc = {}
    seen = set()
    for ln in Path(log_path).read_text().splitlines():
        m = LINE.match(ln.strip())
        if not m:
            continue
        pid, raw_sc, verdict, conf = m.groups()
        if pid in seen:
            continue  # dedupe across resume sections; first occurrence wins
        seen.add(pid)
        canon = EL.CANON.get(pid)
        ok_format = bool(verdict) and bool(conf)
        formats += int(ok_format)
        if canon is None:
            quals += 1
        else:
            sc = 1.0 if verdict.strip().lower() == canon else 0.0
            scores.append(sc)
            cat_acc.setdefault(cat_of(pid), []).append(sc)
    n = len(scores)
    acc = sum(scores) / n if n else float("nan")
    total = n + quals
    fmt = formats / total if total else float("nan")
    print(f"canonical verdict accuracy (exact): {acc:.3f}  "
          f"(n={n}, qualitative={quals})  format rate: {fmt:.2f}")
    for cat, v in sorted(cat_acc.items()):
        print(f"  {cat:12s} acc {sum(v)/len(v):.3f}  n={len(v)}")
    return {"accuracy": acc, "n": n, "qualitative": quals,
            "format_rate": fmt,
            "by_category": {k: sum(v) / len(v) for k, v in cat_acc.items()}}


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("log")
    args = ap.parse_args()
    summarize(args.log)