tinymind-native-colab-handoff/bundle/evaluation/global_leaderboards.py

"""Fetch global LLM leaderboard snapshots into local evidence cache."""

from __future__ import annotations

from datetime import datetime, timezone
import json
from pathlib import Path
from typing import Callable, Iterable, Any


ARENA_ENDPOINT = "https://api.wulong.dev/arena-ai-leaderboards/v1/leaderboard"
ARENA_NAMES = ("text", "code", "vision", "search", "text-to-image", "text-to-video")
HF_DATASET = "open-llm-leaderboard/results"


def fetch_arena_leaderboard(name: str, timeout: float = 30.0) -> dict:
    import httpx

    response = httpx.get(ARENA_ENDPOINT, params={"name": name}, timeout=timeout)
    response.raise_for_status()
    return response.json()


def fetch_hf_open_llm_results(limit: int = 100) -> list[dict]:
    try:
        from datasets import load_dataset

        ds = load_dataset(HF_DATASET, split="train", streaming=True)
        rows: list[dict] = []
        for row in ds:
            rows.append({key: _json_safe(value) for key, value in dict(row).items()})
            if len(rows) >= max(1, int(limit)):
                break
        return rows
    except Exception:
        import httpx

        response = httpx.get(
            "https://datasets-server.huggingface.co/rows",
            params={"dataset": HF_DATASET, "config": "default", "split": "train", "offset": 0, "length": max(1, int(limit))},
            timeout=60.0,
        )
        response.raise_for_status()
        payload = response.json()
        return [_json_safe(row.get("row", row)) for row in payload.get("rows", [])]


def _json_safe(value: Any) -> Any:
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, dict):
        return {str(k): _json_safe(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_json_safe(v) for v in value]
    return str(value)


def _arena_rows(payload: dict) -> list[dict]:
    if isinstance(payload.get("leaderboard"), list):
        return payload["leaderboard"]
    if isinstance(payload.get("data"), list):
        return payload["data"]
    if isinstance(payload.get("rows"), list):
        return payload["rows"]
    if isinstance(payload.get("models"), list):
        return payload["models"]
    if isinstance(payload, list):
        return payload
    return []


def build_global_leaderboard_cache(
    out_dir: str | Path,
    arena_names: Iterable[str] = ("text", "code", "vision"),
    hf_limit: int = 50,
    offline: bool = False,
    arena_client: Callable[[str], dict] | None = None,
    hf_client: Callable[[int], list[dict]] | None = None,
) -> dict:
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    arena_client = arena_client or fetch_arena_leaderboard
    hf_client = hf_client or fetch_hf_open_llm_results

    arena: dict[str, dict] = {}
    for name in arena_names:
        if offline:
            arena[name] = {"status": "offline", "rows": [], "raw": None, "source_url": f"{ARENA_ENDPOINT}?name={name}"}
            continue
        try:
            raw = arena_client(name)
            rows = _arena_rows(raw)
            arena[name] = {"status": "ok", "rows": rows, "raw": _json_safe(raw), "source_url": f"{ARENA_ENDPOINT}?name={name}"}
        except Exception as exc:
            arena[name] = {"status": "error", "rows": [], "raw": None, "error": f"{type(exc).__name__}: {exc}", "source_url": f"{ARENA_ENDPOINT}?name={name}"}

    if offline:
        hf = {"status": "offline", "rows": [], "dataset": HF_DATASET}
    else:
        try:
            rows = hf_client(int(hf_limit))
            hf = {"status": "ok", "rows": rows, "dataset": HF_DATASET, "limit": int(hf_limit)}
        except Exception as exc:
            hf = {"status": "error", "rows": [], "dataset": HF_DATASET, "limit": int(hf_limit), "error": f"{type(exc).__name__}: {exc}"}

    report = {
        "schema_version": "tinymind-global-leaderboard-cache-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "sources": {
            "arena_ai_snapshot": {
                "endpoint": ARENA_ENDPOINT,
                "names": list(arena_names),
                "note": "Unofficial structured snapshot API for Arena AI/LMSYS-style leaderboards.",
            },
            "huggingface_open_llm": {
                "dataset": HF_DATASET,
                "note": "Hugging Face dataset-backed technical benchmark results.",
            },
        },
        "arena": arena,
        "huggingface": hf,
        "cache_gate": {
            "passed": any(v.get("status") == "ok" for v in arena.values()) or hf.get("status") == "ok" or offline,
            "offline": bool(offline),
        },
        "claim_gate": {
            "world_rank_claim_allowed": False,
            "reason": "These are reference leaderboards/caches. TinyMind needs its own listed results before rank claims.",
        },
    }
    cache_path = out / "global_leaderboards_cache.json"
    json_path = out / "global_leaderboards_report.json"
    md_path = out / "global_leaderboards_report.md"
    report["cache_path"] = str(cache_path)
    report["json_path"] = str(json_path)
    report["markdown_path"] = str(md_path)
    cache_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    md_path.write_text(_markdown(report), encoding="utf-8")
    return report


def _markdown(report: dict) -> str:
    lines = [
        "# TinyMind Global Leaderboard Cache",
        "",
        f"- Cache gate: {report['cache_gate']['passed']}",
        f"- Offline: {report['cache_gate']['offline']}",
        f"- World rank claim allowed: {report['claim_gate']['world_rank_claim_allowed']}",
        "",
        "## Arena AI",
        "",
    ]
    for name, payload in report["arena"].items():
        lines.append(f"- {name}: {payload['status']} rows={len(payload.get('rows', []))}")
    lines.extend(["", "## Hugging Face Open LLM", "", f"- status: {report['huggingface']['status']} rows={len(report['huggingface'].get('rows', []))}", ""])
    return "\n".join(lines)