(2/n) Organize Benchmark Execution Into Responsibility-Owned Packages

judgearena/artifacts/__init__.py

@@ -0,0 +1,16 @@
+"""Run artifacts and reproducibility metadata."""
+
+from judgearena.artifacts.metadata import to_jsonable, write_run_metadata
+from judgearena.artifacts.run import (
+    prepare_run_directory,
+    safe_filename,
+    write_run_metadata_safely,
+)
+
+__all__ = [
+    "prepare_run_directory",
+    "safe_filename",
+    "to_jsonable",
+    "write_run_metadata",
+    "write_run_metadata_safely",
+]

judgearena/repro.py → judgearena/artifacts/metadata.py

@@ -28,7 +28,7 @@ def _utc_now() -> datetime:
     return datetime.now(UTC)
 
 
-def _to_jsonable(value: Any) -> Any:
+def to_jsonable(value: Any) -> Any:
     """Convert arbitrary objects into JSON-safe values."""
     if value is None or isinstance(value, (str, int, bool)):
         return value
@@ -40,22 +40,22 @@ def _to_jsonable(value: Any) -> Any:
     if isinstance(value, Path):
         return str(value)
     if isinstance(value, dict):
-        return {str(k): _to_jsonable(v) for k, v in value.items()}
+        return {str(k): to_jsonable(v) for k, v in value.items()}
     if isinstance(value, (list, tuple, set)):
-        return [_to_jsonable(v) for v in value]
+        return [to_jsonable(v) for v in value]
     # numpy / pandas scalars usually expose .item()
     item = getattr(value, "item", None)
     if callable(item):
         try:
-            return _to_jsonable(item())
+            return to_jsonable(item())
         except Exception:
             pass
     return str(value)
 
 
 def _stable_json_dumps(value: Any) -> str:
     return json.dumps(
-        _to_jsonable(value), sort_keys=True, separators=(",", ":"), ensure_ascii=False
+        to_jsonable(value), sort_keys=True, separators=(",", ":"), ensure_ascii=False
     )
 
 
@@ -71,7 +71,7 @@ def _hash_normalized_set_sha256(values: list[Any] | None) -> str | None:
 
     normalized_by_key: dict[str, Any] = {}
     for value in values:
-        normalized = _to_jsonable(value)
+        normalized = to_jsonable(value)
         normalized_by_key[_stable_json_dumps(normalized)] = normalized
 
     normalized_values = [normalized_by_key[key] for key in sorted(normalized_by_key)]
@@ -199,7 +199,7 @@ def _build_dataset_statistics(input_payloads: dict[str, Any] | None) -> dict[str
         return {}
     summary: dict[str, Any] = {}
     for key, payload in input_payloads.items():
-        normalized = _to_jsonable(payload)
+        normalized = to_jsonable(payload)
         count = len(normalized) if hasattr(normalized, "__len__") else None
         summary[f"{key}_count"] = count
     return summary
@@ -210,7 +210,7 @@ def _compact_results(results: dict[str, Any] | None) -> dict[str, Any]:
 
     Keep summary stats but avoid embedding large per-sample arrays.
     """
-    payload = _to_jsonable(results or {})
+    payload = to_jsonable(results or {})
     if not isinstance(payload, dict):
         return {"value": payload}
 
@@ -251,7 +251,7 @@ def write_run_metadata(
             "duration_sec": duration_sec,
         },
         "entrypoint": entrypoint,
-        "run": _to_jsonable(run),
+        "run": to_jsonable(run),
         "results": _compact_results(results),
         "dataset_statistics": _build_dataset_statistics(input_payloads),
         "environment": {
@@ -268,7 +268,7 @@ def write_run_metadata(
 
     instruction_indices = None
     if input_payloads and "instruction_index" in input_payloads:
-        raw_indices = _to_jsonable(input_payloads["instruction_index"])
+        raw_indices = to_jsonable(input_payloads["instruction_index"])
         if isinstance(raw_indices, list):
             instruction_indices = raw_indices
     instruction_indices_hash = _hash_normalized_set_sha256(instruction_indices)
@@ -289,5 +289,5 @@ def write_run_metadata(
 
     metadata_path = output_path / metadata_filename
     with open(metadata_path, "w", encoding="utf-8") as f:
-        json.dump(_to_jsonable(metadata), f, indent=2, allow_nan=False)
+        json.dump(to_jsonable(metadata), f, indent=2, allow_nan=False)
     return metadata_path

judgearena/artifacts/run.py

@@ -0,0 +1,45 @@
+"""Shared lifecycle helpers for benchmark result directories."""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from judgearena.artifacts.metadata import write_run_metadata
+from judgearena.log import attach_file_handler, get_logger, make_run_log_path
+
+if TYPE_CHECKING:
+    from judgearena.config import RunConfig
+
+logger = get_logger(__name__)
+
+
+def safe_filename(value: str) -> str:
+    """Return a filesystem-safe model, task, or benchmark name."""
+    slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
+    return slug or "value"
+
+
+def prepare_run_directory(
+    cfg: RunConfig, path: str | Path, *, attach_log: bool = True
+) -> Path:
+    """Create a result directory and persist its resolved configuration."""
+    output_dir = Path(path)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    if attach_log and not cfg.run.no_log_file:
+        attach_file_handler(make_run_log_path(output_dir))
+
+    from judgearena.config import dump_config
+
+    dump_config(cfg, output_dir / "config.yaml")
+    return output_dir
+
+
+def write_run_metadata_safely(**kwargs):
+    """Write reproducibility metadata without discarding completed results."""
+    try:
+        return write_run_metadata(**kwargs)
+    except OSError as exc:
+        logger.warning("Failed to write run metadata: %s", exc)
+        return None

judgearena/battles.py

@@ -16,7 +16,7 @@
 import numpy as np
 import pandas as pd
 
-from judgearena.repro import _to_jsonable
+from judgearena.artifacts import to_jsonable
 
 # Winners accepted by compute_bradley_terry.
 WINNERS = frozenset({"model_a", "model_b", "tie", "tie (bothbad)"})
@@ -76,7 +76,7 @@ class RatingEntry:
 class Leaderboard:
     """Per-model ratings (mean + bootstrap CI) plus the run metadata that produced them.
 
-    Named distinctly from :class:`judgearena.estimate_elo_ratings.EloReport`,
+    Named distinctly from :class:`judgearena.benchmarks.elo.runner.EloReport`,
     which is the console/``results-*.json`` run report; this is the narrower
     ``elo_ratings.json`` leaderboard artifact with per-model CIs.
     """
@@ -90,7 +90,7 @@ class Leaderboard:
 
     def write(self, path: str | Path) -> None:
         with Path(path).open("w") as f:
-            json.dump(_to_jsonable(asdict(self)), f, indent=2, allow_nan=False)
+            json.dump(to_jsonable(asdict(self)), f, indent=2, allow_nan=False)
 
 
 def summarize_bootstrap(

judgearena/benchmarks/__init__.py

@@ -0,0 +1 @@
+"""Benchmark-specific runners and evaluation logic."""

judgearena/benchmarks/elo/__init__.py

@@ -0,0 +1 @@
+"""Arena-anchored ELO benchmark implementation."""

judgearena/benchmarks/elo/rating.py

@@ -0,0 +1,189 @@
+"""Pure sampling and Bradley-Terry rating logic for arena-anchored ELO."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+
+import numpy as np
+import pandas as pd
+from sklearn.linear_model import LogisticRegression
+
+
+def winner_to_pref(winner: str) -> float | None:
+    """Convert a hard winner label to a continuous preference value."""
+    if winner == "model_a":
+        return 0.0
+    if winner == "model_b":
+        return 1.0
+    if winner in ("tie", "tie (bothbad)"):
+        return 0.5
+    return None
+
+
+def _is_nan_pref(pref) -> bool:
+    return pref is None or (isinstance(pref, float) and np.isnan(pref))
+
+
+def fit_bradley_terry(
+    df: pd.DataFrame,
+    pref_col: str = "pref",
+    scale: float = 400,
+    base: float = 10,
+    init_rating: float = 1000,
+    baseline_model: str | None = None,
+    baseline_rating: float = 1000,
+) -> dict[str, float]:
+    """Fit Bradley-Terry ratings via weighted logistic regression."""
+    df = df.dropna(subset=[pref_col])
+    if df.empty:
+        return {}
+
+    grouped = (
+        df.groupby(["model_a", "model_b", pref_col]).size().reset_index(name="count")
+    )
+    all_models = sorted(set(grouped["model_a"]) | set(grouped["model_b"]))
+    models = pd.Series(np.arange(len(all_models)), index=all_models)
+
+    model_a = grouped["model_a"].map(models).to_numpy()
+    model_b = grouped["model_b"].map(models).to_numpy()
+    prefs = grouped[pref_col].to_numpy(dtype=float)
+    counts = grouped["count"].to_numpy(dtype=float)
+    n = len(grouped)
+
+    design = np.zeros((2 * n, len(models)))
+    top = np.arange(n)
+    bottom = n + top
+    design[top, model_a] = np.log(base)
+    design[top, model_b] = -np.log(base)
+    design[bottom, model_a] = np.log(base)
+    design[bottom, model_b] = -np.log(base)
+
+    labels = np.concatenate([np.ones(n), np.zeros(n)])
+    sample_weights = np.concatenate([(1.0 - prefs) * counts, prefs * counts])
+    if sample_weights.sum() == 0:
+        return {}
+
+    model = LogisticRegression(fit_intercept=False, C=1e10, tol=1e-6, max_iter=1000)
+    model.fit(design, labels, sample_weight=sample_weights)
+    ratings = scale * model.coef_[0] + init_rating
+    if baseline_model is not None and baseline_model in models.index:
+        ratings += baseline_rating - ratings[models[baseline_model]]
+    return dict(pd.Series(ratings, index=models.index))
+
+
+def _sample_fingerprint(sampled: pd.DataFrame) -> str:
+    rows = [
+        {
+            "index": int(index) if isinstance(index, int | np.integer) else str(index),
+            "question_id": str(row["question_id"]),
+            "model_a": str(row["model_a"]),
+            "model_b": str(row["model_b"]),
+        }
+        for index, row in sampled.iterrows()
+    ]
+    return hashlib.sha256(json.dumps(rows, sort_keys=True).encode()).hexdigest()
+
+
+def select_seeded_random_arena_battles(
+    df_battles: pd.DataFrame,
+    *,
+    n_battles: int,
+    seed: int,
+) -> tuple[pd.DataFrame, dict[str, object]]:
+    """Select a shared random battle panel for outside-model ELO estimation."""
+    sampled = df_battles.sample(
+        n=min(n_battles, len(df_battles)), random_state=seed, replace=False
+    )
+    metadata: dict[str, object] = {
+        "sampling_mode": "seeded_random",
+        "random_seed": seed,
+        "requested_rows": n_battles,
+        "sampled_rows": len(sampled),
+        "sampled_original_indices": [
+            int(index) if isinstance(index, int | np.integer) else str(index)
+            for index in sampled.index
+        ],
+        "sampled_question_ids": [
+            str(value) for value in sampled["question_id"].tolist()
+        ],
+        "sample_fingerprint": _sample_fingerprint(sampled),
+    }
+    return sampled.reset_index(drop=True), metadata
+
+
+def sampling_cache_token(
+    sampling_metadata: dict[str, object],
+    *,
+    n_instructions: int | None,
+    n_instructions_per_language: int | None,
+) -> str:
+    if sampling_metadata.get("sampling_mode") == "seeded_random":
+        return (
+            "seeded-random_"
+            f"{sampling_metadata['requested_rows']}_"
+            f"seed-{sampling_metadata['random_seed']}_"
+            f"{str(sampling_metadata['sample_fingerprint'])[:12]}"
+        )
+    return f"head_{n_instructions}_{n_instructions_per_language}"
+
+
+def prefs_to_battle_results(
+    prefs,
+    our_model_is_position_a,
+    opponent_models,
+    model_name: str,
+    *,
+    judge_model: str | None = None,
+    question_ids=None,
+) -> pd.DataFrame:
+    """Map position-oriented preferences into model-name-level battle rows."""
+    records = []
+    for pref, is_pos_a, opponent in zip(
+        prefs, our_model_is_position_a, opponent_models, strict=True
+    ):
+        if _is_nan_pref(pref) or pref == 0.5:
+            winner = "tie"
+        elif pref < 0.5:
+            winner = "model_a"
+        else:
+            winner = "model_b"
+
+        if is_pos_a:
+            record = {
+                "model_a": model_name,
+                "model_b": opponent,
+                "winner": winner,
+                "pref": pref,
+            }
+        else:
+            record = {
+                "model_a": opponent,
+                "model_b": model_name,
+                "winner": winner,
+                "pref": None if _is_nan_pref(pref) else pref,
+            }
+        record["pref_hard"] = winner_to_pref(winner)
+        records.append(record)
+
+    frame = pd.DataFrame(records)
+    frame["source"] = "llm-judge"
+    frame["judge_model"] = judge_model
+    if question_ids is not None:
+        frame["question_id"] = question_ids
+    return frame
+
+
+def arena_anchor_battles(df_arena_all: pd.DataFrame) -> pd.DataFrame:
+    """Build deterministic human-anchor battles from a loaded arena frame."""
+    frame = df_arena_all.loc[:, ["model_a", "model_b", "winner"]].copy()
+    counts = pd.concat([frame["model_a"], frame["model_b"]]).value_counts()
+    well_represented = set(counts[counts >= 500].index)
+    frame = frame[
+        frame["model_a"].isin(well_represented)
+        & frame["model_b"].isin(well_represented)
+    ]
+    frame["pref"] = frame["winner"].map(winner_to_pref)
+    frame["pref_hard"] = frame["pref"]
+    frame["source"] = "human"
+    return frame