1"""Public leaderboards → append-only `benchmark_results` (tier 2). One connector class per leaderboard domain (the SDK binds one2`source_key` per connector, and each board has its own rate limit and provenance) — module kept as `leaderboards`.34 * aider_leaderboard raw polyglot_leaderboard.yml from the aider repository → `aider-polyglot` (pass_rate_2 + well-formed rate).5 * swebench_leaderboard https://www.swebench.com/ embeds `<script id="leaderboard-data">` (5 boards: Verified, Lite, Test, Multimodal,6 Multilingual) → results keyed by (model, benchmark, config {system, system_org, …}).7 * livebench_leaderboard https://livebench.ai/ is a React app; its bundle lists the release dates and fetches `./table_<date>.csv` +8 `./categories_<date>.json` (plus a model metadata map with display names / HF links). The connector follows9 index → bundle → categories → table and emits category averages + the overall mean (LiveBench's own aggregation).10 * artificial_analysis https://artificialanalysis.ai/leaderboards/models — Next.js RSC flight payload (`self.__next_f.push`) with the full11 model list: Intelligence Index (config {version}) + component evaluations mapped to registry benchmarks.1213Model names on leaderboards are free text → `EntityRef("model", name, aliases=[…])`; the resolver merges by normalized alias when the14model exists and otherwise creates a low-tier entity.15"""16from __future__ import annotations1718import csv19import io20import json21import re22from datetime import UTC, date, datetime23from statistics import fmean24from typing import Any2526import yaml2728from aiatlas.registry import load, org_by_hf, org_ref, organizations29from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext30from aiatlas.sdk.extract.dates import parse_datetime31from aiatlas.sdk.facts import EntityRef, Facts, Target32from aiatlas.sdk.fetch import FetchResult333435def benchmark_ref(key: str) -> EntityRef | None:36 b = next((b for b in load("benchmarks") if b["key"] == key), None)37 if not b:38 return None39 return EntityRef(entity_type="benchmark", name=b["name"], identifiers={"registry_benchmark": key}, aliases=list(b.get("aliases", [])), slug_hint=key)404142def _slug(s: str) -> str:43 return re.sub(r"[^a-z0-9]+", "", s.lower())444546def org_from_name(facts: Facts, *names: str | None) -> EntityRef | None:47 """Registry organization by hf_org / key / name / alias (punctuation-insensitive: 'Z AI' == 'Z.ai'); models map to the lab, not the holding."""48 for n in names:49 if not n:50 continue51 low = n.strip().lower()52 known = org_by_hf(low)53 key = known["key"] if known else None54 if not key:55 wanted = _slug(low)56 for k, o in organizations().items():57 if wanted in {_slug(k), _slug(o["name"]), *(_slug(a) for a in o.get("aliases", []))}:58 key = k59 break60 if key:61 if key == "meta" and "meta-ai" in organizations():62 key = "meta-ai"63 ref = org_ref(key)64 facts.entities.append(ref)65 return ref66 return None676869def _dt(v: Any) -> datetime | None:70 if isinstance(v, datetime):71 return v if v.tzinfo else v.replace(tzinfo=UTC)72 if isinstance(v, date):73 return datetime(v.year, v.month, v.day, tzinfo=UTC)74 return parse_datetime(v) if isinstance(v, str) else None757677class _Leaderboard(BaseConnector):78 version = "1"79 parser_version = "1"80 interval_seconds = 12 * 360081 min_interval_seconds = 6 * 360082 max_interval_seconds = 3 * 8640083 tier = 284 priority = 285 concurrency = 1868788# ================================================================================================ Aider polyglot89class AiderLeaderboardConnector(_Leaderboard):90 name = "aider_leaderboard"91 label = "Aider polyglot coding leaderboard"92 description = "polyglot_leaderboard.yml (raw GitHub file of the aider repo): pass rate, well-formed rate, cost, edit format per run."93 source_key = "aider.chat"94 rate_per_min = 1095 expected_min_records = 3096 URL = "https://raw.githubusercontent.com/Aider-AI/aider/main/aider/website/_data/polyglot_leaderboard.yml"9798 async def discover(self, ctx: RunContext) -> list[Target]:99 return [Target(url=self.URL, doc_type="leaderboard", key="aider", min_bytes=1000, priority=1, meta={"content_type": "text/yaml"})]100101 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:102 facts = Facts()103 bench = benchmark_ref("aider-polyglot")104 if bench is None:105 return facts106 facts.entities.append(bench)107 try:108 rows = yaml.safe_load(res.text) or []109 except yaml.YAMLError:110 return facts111 for row in rows:112 if not isinstance(row, dict) or not row.get("model") or row.get("pass_rate_2") is None:113 continue114 name = str(row["model"]).strip()115 # the `--model` id is NOT an alias: many display-name variants share one API id, which would make every alias lookup116 # ambiguous (the resolver then creates a new entity per run). It is kept in the result config instead.117 ref = facts.entity("model", name[:200])118 evaluated = _dt(row.get("date"))119 config = {k: row.get(k) for k in ("edit_format", "command", "test_cases", "versions", "dirname") if row.get(k) is not None}120 m = re.search(r"--model\s+(\S+)", str(row.get("command") or ""))121 if m:122 config["api_model_id"] = m.group(1)123 config["date"] = evaluated.date().isoformat() if evaluated else None124 if row.get("total_cost") is not None:125 config["total_cost_usd"] = row["total_cost"]126 if row.get("seconds_per_case") is not None:127 config["seconds_per_case"] = row["seconds_per_case"]128 facts.result(model=ref, benchmark=bench, score=float(row["pass_rate_2"]), metric="pass_rate_2", unit="%", config=config, evaluated_at=evaluated,129 source_url="https://aider.chat/docs/leaderboards/")130 if row.get("percent_cases_well_formed") is not None:131 facts.result(model=ref, benchmark=bench, score=float(row["percent_cases_well_formed"]), metric="percent_cases_well_formed", unit="%",132 config=config, evaluated_at=evaluated, source_url="https://aider.chat/docs/leaderboards/")133 facts.document_entity = bench134 facts.document_title = "Aider polyglot leaderboard data"135 return facts136137138# ================================================================================================ SWE-bench139BOARD_KEYS = {"verified": "swe-bench-verified", "lite": "swe-bench-lite", "test": "swe-bench-full", "full": "swe-bench-full",140 "multimodal": "swe-bench-multimodal", "multilingual": "swe-bench-multilingual"}141142143class SweBenchLeaderboardConnector(_Leaderboard):144 name = "swebench_leaderboard"145 label = "SWE-bench leaderboards (Verified, Lite, Full, Multimodal, Multilingual)"146 description = "Embedded leaderboard JSON of swebench.com: % resolved per (system, model) submission with dates and cost."147 source_key = "swebench.com"148 rate_per_min = 10149 expected_min_records = 50150151 async def discover(self, ctx: RunContext) -> list[Target]:152 return [Target(url="https://www.swebench.com/", doc_type="leaderboard", key="swebench", min_bytes=20000, priority=1)]153154 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:155 facts = Facts()156 html = parsed.html157 if not html:158 return facts159 data = html.embedded_json.get("leaderboard-data")160 if not isinstance(data, list):161 return facts162 for board in data:163 key = BOARD_KEYS.get(str(board.get("name") or "").lower())164 bench = benchmark_ref(key) if key else None165 if bench is None:166 continue167 facts.entities.append(bench)168 for r in board.get("results") or []:169 self._row(facts, bench, board.get("name"), r)170 facts.document_title = html.title or "SWE-bench leaderboards"171 return facts172173 def _row(self, facts: Facts, bench: EntityRef, board: str, r: dict[str, Any]) -> None:174 model_name = (r.get("model_display") or "").strip()175 resolved = r.get("resolved")176 if not model_name or not isinstance(resolved, (int, float)):177 return178 tags = [t for t in r.get("tags") or [] if isinstance(t, str)]179 model_tags = [t.split(":", 1)[1].strip() for t in tags if t.lower().startswith("model:")]180 org = org_from_name(facts, r.get("model_org"))181 ref = facts.entity("model", model_name[:200], organization=org) # SWE-bench's "Model:" tag goes to config, not aliases (see aider)182 evaluated = _dt(r.get("date"))183 config = {"board": board, "system": r.get("agent") or r.get("name"), "system_org": r.get("agent_org"), "open_source_system": r.get("os_system"),184 "checked_by_swebench": r.get("checked"), "submission": r.get("folder"), "date": evaluated.date().isoformat() if evaluated else None}185 if model_tags:186 config["model_tag"] = model_tags[0]187 if r.get("reasoning_effort"):188 config["reasoning_effort"] = r["reasoning_effort"]189 if isinstance(r.get("cost"), (int, float)):190 config["total_cost_usd"] = r["cost"]191 if isinstance(r.get("instance_cost"), (int, float)):192 config["cost_per_instance_usd"] = r["instance_cost"]193 facts.result(model=ref, benchmark=bench, score=float(resolved), metric="resolved", unit="%", config={k: v for k, v in config.items() if v is not None},194 evaluated_at=evaluated, source_url="https://www.swebench.com/")195196197# ================================================================================================ LiveBench198RELEASES = re.compile(r'\[("20\d\d-\d\d-\d\d"(?:,"20\d\d-\d\d-\d\d")*)\]')199MODEL_META = re.compile(r'"([\w.:/-]+)":\{((?:[^{}]|\{[^{}]*\})*?displayName:"(?:[^{}]|\{[^{}]*\})*)\}')200META_FIELD = re.compile(r'(url|huggingface|organization|displayName):"([^"]*)"')201202203class LiveBenchLeaderboardConnector(_Leaderboard):204 name = "livebench_leaderboard"205 label = "LiveBench leaderboard"206 description = "Latest LiveBench release table (CSV + categories JSON discovered from the site bundle): category averages and overall mean."207 source_key = "livebench.ai"208 rate_per_min = 10209 expected_min_records = 20210 BASE = "https://livebench.ai/"211212 async def discover(self, ctx: RunContext) -> list[Target]:213 return [Target(url=self.BASE, doc_type="listing", key="livebench_index", min_bytes=200, priority=1)]214215 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:216 facts = Facts()217 key = target.key or ""218 if key == "livebench_index" and parsed.html:219 script = next((s.attributes.get("src") for s in parsed.html.css("script[src]") if "main." in (s.attributes.get("src") or "")), None)220 if script:221 facts.follow(self.BASE + script.lstrip("./"), doc_type="script", key="livebench_js", min_bytes=10000, priority=1, meta={"content_type": "text/javascript"})222 elif key == "livebench_js":223 js = res.text224 dates = sorted({d for m in RELEASES.finditer(js) for d in json.loads("[" + m.group(1) + "]")})225 if not dates:226 ctx.log.warning("livebench bundle: no release list found")227 return facts228 latest = self.config.get("release") or dates[-1]229 models = {}230 for m in MODEL_META.finditer(js):231 fields = dict(META_FIELD.findall(m.group(2)))232 models[m.group(1)] = {"url": fields.get("url"), "huggingface": fields.get("huggingface"), "organization": fields.get("organization"),233 "display_name": fields.get("displayName")}234 tag = latest.replace("-", "_")235 facts.follow(f"{self.BASE}categories_{tag}.json", doc_type="data", key="livebench_categories", min_bytes=20, priority=1,236 meta={"release": latest, "models": models, "content_type": "application/json"})237 elif key == "livebench_categories":238 cats = parsed.json if parsed.kind == "json" else None239 release = target.meta.get("release")240 if isinstance(cats, dict) and release:241 facts.follow(f"{self.BASE}table_{release.replace('-', '_')}.csv", doc_type="leaderboard", key="livebench_table", min_bytes=200, priority=1,242 meta={"release": release, "categories": cats, "models": target.meta.get("models") or {}, "content_type": "text/csv"})243 elif key == "livebench_table":244 self._table(facts, target, res.text)245 return facts246247 def _table(self, facts: Facts, target: Target, text: str) -> None:248 bench = benchmark_ref("livebench")249 if bench is None:250 return251 facts.entities.append(bench)252 release = target.meta.get("release")253 cats: dict[str, list[str]] = target.meta.get("categories") or {}254 meta: dict[str, dict[str, Any]] = target.meta.get("models") or {}255 evaluated = _dt(release)256 for row in csv.DictReader(io.StringIO(text)):257 mid = (row.get("model") or "").strip()258 if not mid:259 continue260 info = meta.get(mid) or {}261 name = info.get("display_name") or mid262 # no `hf_repo` identifier here: several LiveBench ids (effort variants, MLX conversions) share one display name and would263 # glue unrelated hub repositories onto a single entity; the link is kept as a claim on the leaderboard row's model264 org = org_from_name(facts, info.get("organization"))265 ref = facts.entity("model", name[:200], organization=org, aliases=[mid])266 hf = info.get("huggingface") or ""267 m = re.search(r"huggingface\.co/([\w.-]+/[\w.-]+)", hf)268 if m:269 facts.claim(ref, "livebench_hf_link", f"https://huggingface.co/{m.group(1)}")270 scores = {k: float(v) for k, v in row.items() if k and k != "model" and v not in (None, "", "-") and _is_number(v)}271 cat_avgs: dict[str, float] = {}272 for cat, tasks in cats.items():273 vals = [scores[t] for t in tasks if t in scores]274 if vals:275 cat_avgs[cat] = round(fmean(vals), 3)276 if not cat_avgs:277 continue278 base_cfg = {"release": release, "livebench_model_id": mid, "aggregation": "mean of category averages; category = mean of its subtasks"}279 facts.result(model=ref, benchmark=bench, score=round(fmean(cat_avgs.values()), 3), metric="global_average", unit="%", config=base_cfg,280 evaluated_at=evaluated, source_url=self.BASE)281 for cat, avg in cat_avgs.items():282 facts.result(model=ref, benchmark=bench, score=avg, metric=f"category:{cat}", unit="%", config={"release": release, "livebench_model_id": mid, "subtasks": cats[cat]},283 evaluated_at=evaluated, source_url=self.BASE)284 facts.document_entity = bench285 facts.document_title = f"LiveBench {release}"286287288def _is_number(v: str) -> bool:289 try:290 float(v)291 return True292 except ValueError:293 return False294295296# ================================================================================================ Artificial Analysis297AA_EVALS = {"gpqa": ("gpqa", "accuracy", "GPQA Diamond"), "hle": ("humanitys-last-exam", "accuracy", None), "aime25": ("aime-2025", "accuracy", None),298 "livecodebench": ("livecodebench", "pass@1", None), "scicode": ("scicode", "accuracy", None), "ifbench": ("ifbench", "accuracy", None),299 "mmmuPro": ("mmmu-pro", "accuracy", None), "tau2": ("tau2-bench", "pass^1", None), "terminalbenchV40": ("terminal-bench", "accuracy", "v4.0"),300 "terminalbenchV21": ("terminal-bench", "accuracy", "v2.1"), "terminalbenchHard": ("terminal-bench", "accuracy", "hard")}301PUSH = re.compile(r'self\.__next_f\.push\(\[1,"(.*?)"\]\)\s*</script>', re.DOTALL)302303304class ArtificialAnalysisConnector(_Leaderboard):305 name = "artificial_analysis"306 label = "Artificial Analysis — Intelligence Index and component evaluations"307 description = "Model leaderboard page (RSC payload): Intelligence Index with version, component evaluations, openness and release dates."308 source_key = "artificialanalysis.ai"309 rate_per_min = 6310 expected_min_records = 100311 URL = "https://artificialanalysis.ai/leaderboards/models"312313 async def discover(self, ctx: RunContext) -> list[Target]:314 return [Target(url=self.URL, doc_type="leaderboard", key="aa_models", min_bytes=50000, priority=1)]315316 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:317 facts = Facts()318 models, version = parse_rsc_models(res.text)319 if not models:320 ctx.log.warning("artificial analysis: no model objects in the RSC payload")321 return facts322 index = benchmark_ref("artificial-analysis-intelligence-index")323 assert index324 facts.entities.append(index)325 benches = {k: benchmark_ref(k) for k, *_ in AA_EVALS.values()}326 for slug, m in models.items():327 name = (m.get("name") or slug).strip()328 creator = m.get("creator") or {}329 creator_name = creator.get("name") or m.get("modelCreatorName")330 org = org_from_name(facts, creator.get("slug"), creator_name)331 if org is None and creator_name:332 org = facts.entity("company", creator_name, identifiers={"artificial_analysis_creator": creator.get("slug") or _slug(creator_name)})333 ref = facts.entity("model", name[:200], identifiers={"artificial_analysis": slug}, organization=org,334 aliases=[a for a in {slug, m.get("shortName")} if a and a != name])335 if m.get("releaseDate"):336 facts.claim(ref, "release_date", str(m["releaseDate"])[:10])337 if isinstance(m.get("isOpenWeights"), bool):338 facts.claim(ref, "openness", "open-weights" if m["isOpenWeights"] else "proprietary")339 if isinstance(m.get("isReasoning"), bool):340 facts.claim(ref, "reasoning", m["isReasoning"])341 if m.get("deprecated") is True:342 facts.claim(ref, "status", "deprecated")343 if isinstance(m.get("contextWindowTokens"), int) and m["contextWindowTokens"] > 0:344 facts.claim(ref, "context_length", m["contextWindowTokens"], unit="tokens")345 if isinstance(m.get("medianOutputTokensPerSecond"), (int, float)):346 facts.claim(ref, "metric.aa_median_output_tokens_per_second", round(float(m["medianOutputTokensPerSecond"]), 1))347 ii = m.get("intelligenceIndex")348 if isinstance(ii, (int, float)):349 cfg = {"version": version, "estimated": bool(m.get("intelligenceIndexIsEstimated")), "aa_slug": slug}350 facts.result(model=ref, benchmark=index, score=round(float(ii), 2), metric="index", unit="", config=cfg, source_url=self.URL)351 for field, (bkey, metric, variant) in AA_EVALS.items():352 v = m.get(field)353 bench = benches.get(bkey)354 if bench is None or not isinstance(v, (int, float)):355 continue356 if bench not in facts.entities:357 facts.entities.append(bench)358 cfg = {"evaluator": "Artificial Analysis", "index_version": version, "aa_slug": slug}359 if variant:360 cfg["variant"] = variant361 facts.result(model=ref, benchmark=bench, score=round(float(v) * 100, 2), metric=metric, unit="%", config=cfg, source_url=self.URL)362 facts.document_entity = index363 facts.document_title = f"Artificial Analysis models — Intelligence Index v{version}" if version else "Artificial Analysis models"364 return facts365366367def parse_rsc_models(html: str) -> tuple[dict[str, dict[str, Any]], str | None]:368 """Concatenate the Next.js flight chunks, parse each `id:json` line and collect every object with `slug` + `intelligenceIndex`."""369 chunks = []370 for p in PUSH.findall(html):371 try:372 chunks.append(json.loads('"' + p + '"'))373 except json.JSONDecodeError:374 continue375 blob = "".join(chunks)376 version_m = re.search(r"Intelligence Index v(\d+(?:\.\d+)?)", blob)377 models: dict[str, dict[str, Any]] = {}378 for line in blob.split("\n"):379 _, sep, payload = line.partition(":")380 if not sep or "intelligenceIndex" not in payload:381 continue382 try:383 data = json.loads(payload)384 except json.JSONDecodeError:385 continue386 stack = [data]387 while stack:388 o = stack.pop()389 if isinstance(o, dict):390 if "intelligenceIndex" in o and isinstance(o.get("slug"), str):391 models.setdefault(o["slug"], o)392 else:393 stack.extend(o.values())394 elif isinstance(o, list):395 stack.extend(o)396 return models, (version_m.group(1) if version_m else None)397398399CONNECTORS = [AiderLeaderboardConnector, SweBenchLeaderboardConnector, LiveBenchLeaderboardConnector, ArtificialAnalysisConnector]400