Blame for export_router_run.py

#!/usr/bin/env python3
"""Export the Router Benchmark board from the paper's generated stats.

Reads stats.json, collapse.dat, and isolation.dat. Does not retype results.
Writes runs/router1_2026-10-26/ for the Hugging Face Space and the website importer.

Usage (from AutoBench-Leaderboard-live):
  python export_router_run.py
"""

from __future__ import annotations

import argparse
import csv
import json
import math
import re
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path


# Frozen answer matrix this board is scored on. Identity of the published
# model run, not a result copied out of the paper tables.
MATRIX_DATE = "2026-08-29"
MATRIX_MODELS = 20
COLLAPSE_THRESHOLD = 0.90

# Official AutoBench Router Index. beta is locked at 0.2.
# Quality is the share of the gap from the seed-7 random floor to a perfect answer.
# Cost is a log score between fixed anchors of $0.00001 and $1 per question.
ARI_BETA = 0.2
ARI_DH_RAND = 0.646825
ARI_C_MIN = 1e-5
ARI_C_MAX = 1.0

# key, public label, vendor, setting, is_baseline, collapse.dat label
PASSES = [
    ("oracle", "Oracle", "Baseline", "oracle", 1, None),
    ("cheapest_adequate", "Cheapest-adequate", "Baseline", "cheapest adequate", 1, None),
    ("best_static_single", "Always-Sol", "Baseline", "always Sol", 1, None),
    ("always_luna", "Always-Luna", "Baseline", "always Luna", 1, None),
    ("random", "Random", "Baseline", "random", 1, "Random"),
    ("always_cheapest", "Always-cheapest", "Baseline", "always cheapest", 1, None),
    ("orca_constrained_cheapest", "Orca cheapest", "OrcaRouter", "cheapest", 0, "Orca cheap"),
    ("orca_constrained_balanced", "Orca balanced", "OrcaRouter", "balanced", 0, "Orca bal"),
    ("orca_constrained_quality", "Orca quality", "OrcaRouter", "quality", 0, "Orca qual"),
    ("azure_cost", "Azure cost", "Azure Model Router", "cost", 0, "Azure cost"),
    ("azure_balanced", "Azure balanced", "Azure Model Router", "balanced", 0, "Azure bal"),
    ("azure_quality", "Azure quality", "Azure Model Router", "quality", 0, "Azure qual"),
    ("nadirclaw", "NadirClaw", "NadirClaw", "as shipped", 0, "NadirClaw"),
    ("brick", "Brick", "Brick", "as shipped", 0, "Brick"),
    ("cascade", "Cascade", "Cascade", "as shipped", 0, "Cascade"),
    ("nd_subset_k0", "ND k0", "Not Diamond", "k0", 0, "ND k0"),
    ("nd_subset_k4", "ND k4", "Not Diamond", "k4", 0, "ND k4"),
    ("nd_subset_k5", "ND k5", "Not Diamond", "k5", 0, "ND k5"),
    ("nd_native_k5", "ND native", "Not Diamond", "native", 0, "ND native"),
    ("or_auto", "Auto low", "OpenRouter", "low", 0, "Auto low"),
    ("or_auto_medium", "Auto medium", "OpenRouter", "medium", 0, "Auto med"),
    ("or_auto_high", "Auto high", "OpenRouter", "high", 0, "Auto high"),
    ("or_auto_max", "Auto max", "OpenRouter", "max", 0, "Auto max"),
]

BOARD_FIELDS = [
    "pass_key",
    "label",
    "vendor",
    "setting",
    "n",
    "accuracy",
    "ci_low",
    "ci_high",
    "cost",
    "cost_ci_low",
    "cost_ci_high",
    "top_share",
    "collapsed",
    "is_baseline",
    "display_order",
    "ari",
]


def router_index(signal: float, cost: float) -> float:
    """ARI at the official beta. Returns one decimal, matching the published table."""
    quality = max(0.0, (signal - ARI_DH_RAND) / (1.0 - ARI_DH_RAND))
    if cost <= 0 or quality == 0:
        return 0.0
    span = math.log2(ARI_C_MAX) - math.log2(ARI_C_MIN)
    cost_score = (math.log2(ARI_C_MAX) - math.log2(cost)) / span
    cost_score = min(1.0, max(0.0, cost_score))
    if cost_score == 0:
        return 0.0
    value = (
        100.0
        * (1.0 + ARI_BETA)
        * quality
        * cost_score
        / (ARI_BETA * quality + cost_score)
    )
    return float(
        Decimal(f"{value:.10f}").quantize(Decimal("0.1"), rounding=ROUND_HALF_UP)
    )


def parse_collapse(path: Path) -> dict[str, float]:
    shares: dict[str, float] = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("idx"):
            continue
        match = re.match(r"^\d+\s+([0-9.]+)\s+\{(.+)\}$", line)
        if not match:
            raise SystemExit(f"Unparsed collapse row: {line}")
        shares[match.group(2)] = float(match.group(1))
    return shares


def parse_axis(path: Path) -> list[str]:
    text = path.read_text(encoding="utf-8")
    match = re.search(r"xticklabels=\{([^}]+)\}", text)
    if not match:
        raise SystemExit(f"No xticklabels in {path}")
    return [label.strip() for label in match.group(1).split(",")]


def parse_isolation(path: Path, axis: list[str]) -> list[dict]:
    rows = []
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("x "):
            continue
        x_s, y_s, rate_s = line.split()
        x, y = int(x_s), int(y_s)
        rows.append(
            {
                "axis_a": x,
                "axis_b": y,
                "pass_a": axis[x],
                "pass_b": axis[y],
                "rate": float(rate_s),
            }
        )
    return rows


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--stats",
        type=Path,
        default=Path(
            r"C:\DEV\AutoBench\autobench\docs\projects\router-benchmark-paper\generated\stats.json"
        ),
    )
    parser.add_argument(
        "--out",
        type=Path,
        default=Path(__file__).resolve().parent / "runs" / "router1_2026-10-26",
    )
    args = parser.parse_args()

    generated = args.stats.parent
    stats = json.loads(args.stats.read_text(encoding="utf-8"))
    marginal = stats["marginal"]
    shares = parse_collapse(generated / "data" / "collapse.dat")
    axis = parse_axis(generated / "data" / "isolation_axis.tex")
    same_pick = parse_isolation(generated / "data" / "isolation.dat", axis)
    isolation = stats["isolation_summary"]

    board = []
    for order, (key, label, vendor, setting, is_baseline, collapse_label) in enumerate(PASSES):
        if key not in marginal:
            raise SystemExit(f"Missing marginal key: {key}")
        row = marginal[key]
        top_share = ""
        collapsed = 0
        if collapse_label is not None:
            if collapse_label not in shares:
                raise SystemExit(f"Missing collapse label: {collapse_label}")
            top_share = shares[collapse_label]
            collapsed = 1 if top_share >= COLLAPSE_THRESHOLD else 0
        board.append(
            {
                "pass_key": key,
                "label": label,
                "vendor": vendor,
                "setting": setting,
                "n": int(row["n"]),
                "accuracy": row["dh_pct"],
                "ci_low": row["dh_ci"][0],
                "ci_high": row["dh_ci"][1],
                "cost": row["cost"],
                "cost_ci_low": row["cost_ci"][0],
                "cost_ci_high": row["cost_ci"][1],
                "top_share": top_share,
                "collapsed": collapsed,
                "is_baseline": is_baseline,
                "display_order": order,
                "ari": router_index(float(row["dh"]), float(row["cost"])),
            }
        )

    headline = [row for row in board if row["is_baseline"] == 0]
    collapsed_n = sum(row["collapsed"] for row in headline)
    vendors = sorted({row["vendor"] for row in headline})
    questions = int(marginal["oracle"]["n"])

    metadata = {
        "run_id": "router1_2026-10-26",
        "run_type": "router",
        "title": "AutoBench Router Benchmark 1 - October 2026",
        "date": "2026-10-26",
        "run_version": "2026-10-26",
        "is_latest": False,
        "description": (
            "Commercial routers and baselines graded on a frozen answer matrix. "
            "Accuracy is the share of questions where the chosen model is statistically "
            "as good as the best available answer."
        ),
        "report_url": "https://autobench.org/reports/do-llm-routers-pay-for-themselves",
        "blog_url": "https://autobench.org/reports/do-llm-routers-pay-for-themselves.pdf",
        "matrix_date": MATRIX_DATE,
        "matrix_models": MATRIX_MODELS,
        "questions": questions,
        "same_pick_axis": axis,
        "facts": {
            "headline_passes": len(headline),
            "collapsed_passes": collapsed_n,
            "collapse_threshold": COLLAPSE_THRESHOLD,
            "commercial_routers": len(vendors),
            "cross_vendor_pairs": isolation["n_cross_vendor"],
            "median_same_pick": isolation["median_same_pick"],
            "questions": questions,
            "matrix_models": MATRIX_MODELS,
            "matrix_date": MATRIX_DATE,
            "ari_beta": ARI_BETA,
        },
    }

    args.out.mkdir(parents=True, exist_ok=True)
    (args.out / "metadata.json").write_text(
        json.dumps(metadata, indent=2) + "\n", encoding="utf-8"
    )
    with (args.out / "router_board.csv").open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=BOARD_FIELDS)
        writer.writeheader()
        writer.writerows(board)
    with (args.out / "same_pick.csv").open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(
            handle, fieldnames=["axis_a", "axis_b", "pass_a", "pass_b", "rate"]
        )
        writer.writeheader()
        writer.writerows(same_pick)

    print(f"Wrote {args.out}")
    print(
        f"headline={len(headline)} collapsed={collapsed_n} "
        f"vendors={len(vendors)} ({', '.join(vendors)}) "
        f"pairs={isolation['n_cross_vendor']} median={isolation['median_same_pick']} "
        f"questions={questions} same_pick_cells={len(same_pick)}"
    )


if __name__ == "__main__":
    main()