src/graphtrace/report.py

"""Markdown tables and plots from ``metrics.json``. Numbers come only from the run directory."""

from __future__ import annotations

import json
import math
from pathlib import Path

HEADLINE = [
    ("correct", "Accuracy"),
    ("path_all_valid", "Path valid (all)"),
    ("path_completeness", "Path completeness"),
    ("citation_rate", "Citation rate"),
    ("grounding_precision", "Grounding precision"),
    ("false_premise_rejected", "False-premise rejection"),
    ("conflict_flagged", "Conflict flagged"),
    ("cost_usd", "Cost / item ($)"),
]


def fmt(cell: dict, pct: bool = True) -> str:
    m = cell.get("mean")
    if m is None or (isinstance(m, float) and math.isnan(m)):
        return "n/a"
    if pct:
        return f"{100 * m:.1f} [{100 * cell['low']:.1f}, {100 * cell['high']:.1f}]"
    return f"{m:.4f}"


def _table(title: str, block: dict, keys: list[tuple[str, str]]) -> list[str]:
    lines = [title, "", "| Method | n | errors | " + " | ".join(h for _, h in keys) + " |", "|---" * (len(keys) + 3) + "|"]
    for m, agg in block.items():
        cells = [fmt(agg[k], pct=(k != "cost_usd")) for k, _ in keys]
        lines.append(f"| {m} | {agg['n']} | {agg.get('errors', 0)} | " + " | ".join(cells) + " |")
    return lines + [""]


def markdown(metrics: dict) -> str:
    cats = ", ".join(metrics.get("headline_categories", []))
    lines = [f"Evaluation judge: **{metrics.get('judge_status', 'unknown')}**.", ""]
    headline_keys = [kv for kv in HEADLINE if kv[0] != "conflict_flagged"]
    lines += _table(f"## Headline ({cats}; mean with 95% bootstrap CI; percentages)", metrics.get("headline", metrics["methods"]), headline_keys)
    if metrics.get("conflict_insufficient"):
        ci_keys = [("correct", "Accuracy"), ("conflict_flagged", "Conflict flagged (C1)"), ("insufficient_abstained", "Abstained (C2)")]
        lines += _table(
            "## Conflict / insufficient evidence (reported separately: the GraphTrace verifier shares the C1 agreement rule)",
            metrics["conflict_insufficient"],
            ci_keys,
        )
    lines += _table("## All items", metrics["methods"], HEADLINE)
    lines += ["", "## Accuracy by category", ""]
    cats = sorted({c for v in metrics["by_category"].values() for c in v})
    lines.append("| Method | " + " | ".join(cats) + " |")
    lines.append("|---" * (len(cats) + 1) + "|")
    for m, by in metrics["by_category"].items():
        lines.append(f"| {m} | " + " | ".join(fmt(by[c]["correct"]) if c in by else "n/a" for c in cats) + " |")
    lines += ["", "## Calibration", "", "| Method | ECE | Brier | AUROC | AURC |", "|---|---|---|---|---|"]
    for m, c in metrics["calibration"].items():
        lines.append(f"| {m} | {c['ece']:.3f} | {c['brier']:.3f} | {c['auroc']:.3f} | {c['aurc']:.3f} |")
    if metrics.get("paired_vs_reference"):
        lines += [
            "",
            "## Paired differences vs reference on headline items (method - reference)",
            "",
            "| Method | Metric | n | diff | 95% CI | p |",
            "|---|---|---|---|---|---|",
        ]
        for m, tests in metrics["paired_vs_reference"].items():
            for k, t in tests.items():
                if t["n"]:
                    lines.append(f"| {m} | {k} | {t['n']} | {100 * t['diff']:+.1f} | [{100 * t['low']:+.1f}, {100 * t['high']:+.1f}] | {t['p_value']:.4f} |")
    return "\n".join(lines) + "\n"


def plots(metrics: dict, out_dir: Path) -> list[Path]:
    import matplotlib

    matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    paths = []
    fig, ax = plt.subplots(figsize=(5, 4))
    for m, c in metrics["calibration"].items():
        pts = c["risk_coverage"]
        if pts:
            ax.plot([p[0] for p in pts], [p[1] for p in pts], label=m)
    ax.set_xlabel("Coverage")
    ax.set_ylabel("Risk (error rate)")
    ax.set_title("Risk-coverage")
    ax.legend()
    fig.tight_layout()
    p = out_dir / "risk_coverage.png"
    fig.savefig(p, dpi=150)
    plt.close(fig)
    paths.append(p)

    cats = sorted({c for v in metrics["by_category"].values() for c in v})
    methods = list(metrics["by_category"])
    fig, ax = plt.subplots(figsize=(7, 4))
    w = 0.8 / max(1, len(methods))
    for i, m in enumerate(methods):
        vals = [metrics["by_category"][m].get(c, {}).get("correct", {}).get("mean", float("nan")) for c in cats]
        ax.bar([j + i * w for j in range(len(cats))], vals, width=w, label=m)
    ax.set_xticks([j + 0.4 - w / 2 for j in range(len(cats))])
    ax.set_xticklabels(cats, rotation=15)
    ax.set_ylabel("Accuracy")
    ax.set_ylim(0, 1)
    ax.legend()
    fig.tight_layout()
    p = out_dir / "accuracy_by_category.png"
    fig.savefig(p, dpi=150)
    plt.close(fig)
    paths.append(p)
    return paths


def write_report(run_dir: str | Path, with_plots: bool = True) -> Path:
    run_dir = Path(run_dir)
    metrics = json.loads((run_dir / "metrics.json").read_text(encoding="utf-8"))
    text = f"# Results: {run_dir.name}\n\n" + markdown(metrics)
    if with_plots:
        try:
            for p in plots(metrics, run_dir):
                text += f"\n![{p.stem}]({p.name})\n"
        except ImportError:
            text += "\n(plots skipped: install the 'report' extra)\n"
    out = run_dir / "report.md"
    out.write_text(text, encoding="utf-8")
    return out