6krrt/leaderboard.py

#!/usr/bin/env python3
"""Import curated leaderboard priors into the ``proficiency`` table.

Reads ``leaderboards.yaml`` (family -> category -> score), writes
``leaderboard_score`` for every catalog row in each family, and re-blends.

Run manually, or after editing the YAML:
    python leaderboard.py
    python leaderboard.py --check     # report coverage, write nothing

The coverage report is the point of running this on a schedule. NeuralWatt
adds models; a newly listed family has no prior and no self-eval history, so
it sits at the neutral 0.5 and is indistinguishable from a model that was
measured and found average. Naming those families is what turns "quietly
unmeasured" into "needs a prior".
"""

from __future__ import annotations

import argparse
import sqlite3
import sys
from pathlib import Path

import yaml

from config import RouterConfig, load_config
from proficiency_store import set_leaderboard

LEADERBOARDS_PATH = "leaderboards.yaml"


def load_priors(path: str | Path, cfg: RouterConfig) -> dict[str, dict[str, float]]:
    """Parse and validate the curated file into {family: {category: score}}.

    Validation is strict rather than forgiving: a typo'd category silently
    dropped would look exactly like a benchmark that does not cover it, and
    an out-of-range score would skew the min-max normalization downstream.
    """
    raw = yaml.safe_load(Path(path).read_text()) or {}
    families = raw.get("families") or {}
    if not isinstance(families, dict):
        raise ValueError("leaderboards.yaml: 'families' must be a mapping")

    allowed = set(cfg.proficiency.categories)
    priors: dict[str, dict[str, float]] = {}

    for family, entry in families.items():
        if not isinstance(entry, dict):
            raise ValueError(f"leaderboards.yaml: {family!r} must be a mapping")
        scores = entry.get("scores") or {}
        if not isinstance(scores, dict):
            raise ValueError(f"leaderboards.yaml: {family!r}.scores must be a mapping")

        unknown = set(scores) - allowed
        if unknown:
            raise ValueError(
                f"leaderboards.yaml: {family!r} has categories not in "
                f"proficiency.categories: {sorted(unknown)}"
            )
        for category, score in scores.items():
            if not isinstance(score, (int, float)) or not (0.0 <= score <= 1.0):
                raise ValueError(
                    f"leaderboards.yaml: {family!r}.{category} must be a number "
                    f"in 0..1, got {score!r}"
                )
        if scores:
            priors[family] = {c: float(s) for c, s in scores.items()}

    return priors


def active_families(conn: sqlite3.Connection, cfg: RouterConfig) -> dict[str, list[str]]:
    """Routable families mapped to their catalog rows."""
    placeholders = ",".join("?" * len(cfg.routing.allowed_access_levels))
    rows = conn.execute(
        f"""
        SELECT base_model_id, model_id FROM models
         WHERE availability = 'active'
           AND access_level IN ({placeholders})
         ORDER BY base_model_id, model_id
        """,
        tuple(cfg.routing.allowed_access_levels),
    ).fetchall()
    out: dict[str, list[str]] = {}
    for base, model_id in rows:
        out.setdefault(base, []).append(model_id)
    return out


def report_coverage(
    families: dict[str, list[str]], priors: dict[str, dict[str, float]]
) -> list[str]:
    """Print per-family coverage; return the families with no prior at all."""
    missing = []
    print(f"{'family':22s}{'rows':>6}{'categories with a prior':>26}")
    for family, model_ids in sorted(families.items()):
        scores = priors.get(family, {})
        if not scores:
            missing.append(family)
            print(f"{family:22s}{len(model_ids):>6}{'— none —':>26}")
        else:
            print(f"{family:22s}{len(model_ids):>6}{len(scores):>26}")
    return missing


def apply_priors(
    conn: sqlite3.Connection,
    cfg: RouterConfig,
    families: dict[str, list[str]],
    priors: dict[str, dict[str, float]],
) -> int:
    """Write each family's prior onto every catalog row in that family."""
    written = 0
    for family, model_ids in families.items():
        scores = priors.get(family)
        if not scores:
            continue
        for model_id in model_ids:
            for category, score in scores.items():
                set_leaderboard(conn, cfg, model_id, "neuralwatt", category, score)
                written += 1
    conn.commit()
    return written


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--file", default=LEADERBOARDS_PATH)
    ap.add_argument(
        "--check", action="store_true", help="report coverage without writing"
    )
    args = ap.parse_args()

    cfg = load_config("config.yaml")
    try:
        priors = load_priors(args.file, cfg)
    except (ValueError, FileNotFoundError) as e:
        print(f"{e}", file=sys.stderr)
        return 1

    conn = sqlite3.connect(cfg.database.path)
    families = active_families(conn, cfg)

    missing = report_coverage(families, priors)

    if not args.check:
        written = apply_priors(conn, cfg, families, priors)
        print(f"\nwrote {written} leaderboard scores")
    conn.close()

    if missing:
        print(
            f"\nWARNING: {len(missing)} active famil"
            f"{'y has' if len(missing) == 1 else 'ies have'} no leaderboard prior: "
            f"{', '.join(missing)}",
            file=sys.stderr,
        )
        print(
            "  Until self-eval accumulates, these score the neutral 0.5 — "
            "indistinguishable from a model measured and found average.\n"
            f"  Add real, sourced figures to {args.file}.",
            file=sys.stderr,
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())