"""Intelligence surfaces (API 1.1) — deterministic, no LLM, no composite scores: /frontier · /pareto · /pulse · /open · /find-a-model · /run-locally · /time-machine.""" from __future__ import annotations import statistics from collections import defaultdict from datetime import UTC, datetime, timedelta from datetime import time as dtime from typing import Any from fastapi import APIRouter, Query, Request from aiatlas.api.common import ( CLAIM_COLS, CLAIM_FROM, DOWNLOADABLE_CATEGORIES, ENTITY_COLS, ENTITY_FROM, EVENT_COLS, EVENT_FROM, OPEN_CATEGORIES, PRICE_COLS, PRICE_FROM, ApiError, cached, change_event, csv, deployment_row, entity_summary, num_expr, openness_values, parse_date, resolve_entity, resolve_id, ) from aiatlas.db import connection, fetch_all, fetch_one, fetch_val from aiatlas.ontology.licenses import LICENSES, normalize_license from aiatlas.services import hardware_fit as hf from aiatlas.services.finder import USE_CASES, find_models from aiatlas.services.frontier import ( FRONTIER_METHODOLOGY, all_primary_groups, benchmark_meta, frontier_model_ids, group_rows, group_summary, leader_at, leaderboard_rows, load_results, primary_group, rank_rows, ) from aiatlas.services.pareto import pareto_frontier router = APIRouter(prefix="/api/v1", tags=["intelligence"]) PARAMS = num_expr("e.attributes->>'parameter_count'") CONTEXT = num_expr("e.attributes->>'context_length'") QUALITY_BENCHMARKS = ("artificial-analysis-intelligence-index", "gpqa") PARETO_X = {"output_price": "cheapest current output price (USD / 1M tokens)", "input_price": "cheapest current input price (USD / 1M tokens)", "parameter_count": "total parameters", "context_length": "context window (tokens)", "memory_estimate": "ESTIMATED memory at 4-bit, 8K context (GB)"} def _f(v: Any) -> float | None: if v is None or isinstance(v, bool): return None try: return float(v) except (TypeError, ValueError): return None def price_delta(old: Any, new: Any) -> dict[str, Any] | None: """PRICE_CHANGED events carry `{input_per_mtok, output_per_mtok}` dicts (or a bare number). Returns % changes (output first) or None.""" def pick(v: Any, k: str) -> float | None: return _f(v.get(k)) if isinstance(v, dict) else (_f(v) if k == "output_per_mtok" else None) out: dict[str, Any] = {} for k, label in (("output_per_mtok", "output"), ("input_per_mtok", "input")): o, n = pick(old, k), pick(new, k) if o and n is not None: out[f"{label}_percent"] = round((n - o) / o * 100, 2) out[f"{label}_from"], out[f"{label}_to"] = o, n if not out: return None out["percent"] = out.get("output_percent", out.get("input_percent")) return out def _mods(attrs: dict[str, Any]) -> set[str]: out: set[str] = set() for k in ("modalities", "modalities_input", "modalities_output"): v = attrs.get(k) if isinstance(v, list): out |= {str(x).lower() for x in v} return {("image" if m == "vision" else "document" if m == "pdf" else m) for m in out} async def _cheapest_prices(conn: Any, model_ids: list[str] | None = None) -> dict[str, dict[str, Any]]: where = "p.valid_to is null" + (" and p.model_id = any(cast(:ids as text[]))" if model_ids is not None else "") rows = await fetch_all(conn, f"""select p.model_id, min(p.output_per_mtok) filter (where p.output_per_mtok > 0) as min_output, min(p.input_per_mtok) filter (where p.input_per_mtok > 0) as min_input, count(distinct p.provider_id) as providers, max(p.context_length) as max_ctx from prices p where {where} group by 1""", ids=model_ids) return {r["model_id"]: r for r in rows} async def _cheapest_offer(conn: Any, model_id: str, field: str = "output") -> dict[str, Any] | None: col = "p.output_per_mtok" if field == "output" else "p.input_per_mtok" row = await fetch_one(conn, f"select {PRICE_COLS} from {PRICE_FROM} where p.model_id = :id and p.valid_to is null and {col} > 0 order by {col} asc limit 1", id=model_id) return deployment_row(row) if row else None # ------------------------------------------------------------------------------------------------------------------ /frontier @router.get("/frontier") @cached(300) async def frontier(request: Request, limit: int = Query(12, ge=1, le=50)) -> dict[str, Any]: now = datetime.now(UTC) async with connection() as conn: groups = await all_primary_groups(conn) fids, composition = await frontier_model_ids(conn) major = await fetch_all(conn, f""" select distinct on (e.id) {EVENT_COLS}, ev.occurred_at, ev.is_backfill from {EVENT_FROM} left join sources s on s.id = ev.source_id where ev.event_type in ('NEW_MODEL','RELEASE') and ev.importance >= 3 and e.entity_type = 'model' and e.merged_into is null and coalesce(s.tier, 2) <= 2 and (ev.is_backfill = false or e.attributes->>'release_date' >= :since) order by e.id, ev.occurred_at desc""", since=(now - timedelta(days=60)).date().isoformat()) major.sort(key=lambda r: (str((r.get("e_attributes") or {}).get("release_date") or ""), r["occurred_at"]), reverse=True) cheapest = await fetch_one(conn, f"select {PRICE_COLS} from {PRICE_FROM} where p.valid_to is null and p.output_per_mtok > 0 and p.model_id = any(cast(:ids as text[])) order by p.output_per_mtok asc limit 1", ids=sorted(fids)) cheapest_1m = await fetch_one(conn, f"""select {PRICE_COLS} from {PRICE_FROM} where p.valid_to is null and p.output_per_mtok > 0 and p.model_id = any(cast(:ids as text[])) and coalesce(p.context_length, case when m.attributes->>'context_length' ~ '^[0-9]+$' then (m.attributes->>'context_length')::bigint end) >= 1000000 order by p.output_per_mtok asc limit 1""", ids=sorted(fids)) ctx_rows = await fetch_all(conn, f"select {ENTITY_COLS}, {CONTEXT} as ctx from {ENTITY_FROM} where e.entity_type = 'model' and e.merged_into is null and {CONTEXT} is not null order by {CONTEXT} desc, e.canonical_name limit :lim", lim=limit) open_rows = await fetch_all(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.entity_type = 'model' and e.merged_into is null and e.attributes->>'openness' = any(cast(:o as text[]))", o=list(OPEN_CATEGORIES)) movements = await fetch_all(conn, f"""select {EVENT_COLS}, ev.occurred_at from {EVENT_FROM} where ev.is_backfill = false and ev.occurred_at > now() - interval '30 days' and (ev.event_type in ('BENCHMARK_UPDATED','BENCHMARK_LEADER_CHANGED','NEW_BENCHMARK_LEADER') or ev.event_type = 'PRICE_CHANGED') order by ev.occurred_at desc limit 500""") all_prices = await _cheapest_prices(conn) # ranks per model across primary groups ranks: dict[str, dict[str, int]] = defaultdict(dict) for bid, g in groups.items(): for r in rank_rows(g["rows"], g["higher_is_better"]): ranks[r["model_id"]][g["benchmark"]["slug"]] = r["rank"] bench_frontier = [] agentic = [] for bid, g in sorted(groups.items(), key=lambda kv: -kv[1]["n"]): lb = leaderboard_rows(g) if g["n"] >= 20 and lb: leader, second = lb[0], (lb[1] if len(lb) > 1 else None) bench_frontier.append({"benchmark": {k: g["benchmark"].get(k) for k in ("id", "slug", "name", "category")}, "group": group_summary(g), "leader": leader, "second": second, "gap": round(leader["score"] - second["score"], 3) if second else None}) if (g["benchmark"].get("category") or "").lower() == "agentic" and lb: agentic.append({"benchmark": {k: g["benchmark"].get(k) for k in ("id", "slug", "name")}, "group": group_summary(g), "leaders": lb[:5]}) open_frontier = [] for r in open_rows: attrs = r["attributes"] or {} rk = ranks.get(r["id"], {}) best = min(rk.values()) if rk else None open_frontier.append({"model": entity_summary(r), "best_rank": best, "best_rank_on": min(rk, key=rk.get) if rk else None, "parameter_count": _f(attrs.get("parameter_count")), "context_length": _f(attrs.get("context_length")), "ranks": dict(sorted(rk.items())), "_k": (best if best is not None else 10_000, -(_f(attrs.get("parameter_count")) or 0))}) open_frontier.sort(key=lambda x: x["_k"]) for x in open_frontier: x.pop("_k") # efficiency frontier: quality (index or GPQA primary group) vs cheapest output price quality_group = None for slug in QUALITY_BENCHMARKS: quality_group = next((g for g in groups.values() if g["benchmark"]["slug"] == slug), None) if quality_group: break eff_points = [] if quality_group: for r in rank_rows(quality_group["rows"], quality_group["higher_is_better"]): pr = all_prices.get(r["model_id"]) if pr and pr["min_output"] is not None: eff_points.append({"id": r["model_id"], "model": {"id": r["model_id"], "slug": r["model_slug"], "name": r["model_name"]}, "x": float(pr["min_output"]), "y": float(r["score"]), "rank": r["rank"], "trust_level": r["trust_level"]}) eff_front = set(pareto_frontier(eff_points, maximize_y=quality_group["higher_is_better"] if quality_group else True)) multimodal = [] for mid, rk in ranks.items(): if min(rk.values()) > 10: continue row = next((r for g in groups.values() for r in g["rows"] if r["model_id"] == mid), None) if not row: continue mods = _mods(row.get("model_attrs") or {}) if len(mods) >= 3: multimodal.append({"model": {"id": mid, "slug": row["model_slug"], "name": row["model_name"]}, "modalities": sorted(mods), "top10_on": sorted(b for b, k in rk.items() if k <= 10)}) moves = [] for r in movements: ev = change_event(r) if ev["event_type"] == "PRICE_CHANGED": d = price_delta(ev.get("old_value"), ev.get("new_value")) if d and d["percent"] is not None and abs(d["percent"]) >= 20: ev["percent_change"] = d["percent"] ev["price_delta"] = d ev["provider"] = (ev.get("meta") or {}).get("provider") moves.append(ev) else: moves.append(ev) return { "latest_major_models": [change_event(r) | {"occurred_at": r["occurred_at"], "is_backfill": r["is_backfill"]} for r in major[:limit]], "benchmark_frontier": bench_frontier, "price_frontier": {"cheapest_output": _deploy_or_none(cheapest), "cheapest_output_1m_context": _deploy_or_none(cheapest_1m), "frontier_models": len(fids), "composition": composition}, "context_frontier": [{"model": entity_summary(r), "context_length": r["ctx"]} for r in ctx_rows], "open_weight_frontier": {"items": open_frontier[:limit], "dimensions": ["best_rank", "parameter_count", "context_length"], "note": "sorted by best benchmark rank then parameters; no composite"}, "efficiency_frontier": {"quality": {"benchmark": quality_group["benchmark"]["slug"], "group": group_summary(quality_group)} if quality_group else None, "x": "cheapest current output price (USD / 1M tokens)", "points": [{**p, "pareto": p["id"] in eff_front} for p in sorted(eff_points, key=lambda p: p["x"])], "frontier": sorted(eff_front)}, "agentic_frontier": agentic, "multimodal_frontier": sorted(multimodal, key=lambda x: (-len(x["top10_on"]), x["model"]["name"] or ""))[:limit], "recent_frontier_movements": moves[:50], "generated_at": now, "methodology": FRONTIER_METHODOLOGY + " benchmark_frontier lists the primary comparability group of every benchmark with ≥ 20 current results; efficiency_frontier is the " "Pareto set (maximise quality score, minimise cheapest current output price); recent movements are non-backfill benchmark events and price moves ≥ 20% in 30 days. " "Nothing here is a composite ranking.", } def _deploy_or_none(row: dict[str, Any] | None) -> dict[str, Any] | None: return deployment_row(row) if row else None # ------------------------------------------------------------------------------------------------------------------ /pareto @router.get("/pareto") @cached(300) async def pareto(request: Request, benchmark: str = Query(...), x: str = Query("output_price"), y: str = Query("score"), metric: str | None = None, config_key: str | None = None, org: str | None = None, family: str | None = None, openness: str | None = None) -> dict[str, Any]: if x == "latency": raise ApiError(400, "x=latency is not available: AI Atlas does not store latency measurements (nothing is estimated for it)") if x not in PARETO_X: raise ApiError(400, f"x must be one of {', '.join(PARETO_X)}") if y != "score": raise ApiError(400, "y must be 'score'") async with connection() as conn: bench = await resolve_entity(conn, benchmark, ("benchmark",), aliases=True) rows = await load_results(conn, benchmark_ids=[bench["id"]]) groups = list(group_rows(rows).values()) g = None if metric or config_key: cands = [gg for gg in groups if (not metric or gg["metric"] == metric.lower()) and (not config_key or gg["config_key"] == config_key)] g = max(cands, key=lambda gg: gg["model_count"]) if cands else None else: g = primary_group(bench.get("attributes"), groups) if not g: return {"benchmark": entity_summary(bench), "points": [], "frontier": [], "groups": [group_summary(x) for x in groups], "note": "no current results in the requested group"} ranked = rank_rows(g["rows"], g["higher_is_better"]) ids = [r["model_id"] for r in ranked] prices = await _cheapest_prices(conn, ids) if x in ("output_price", "input_price") else {} offers: dict[str, dict[str, Any]] = {} if x in ("output_price", "input_price"): col = "p.output_per_mtok" if x == "output_price" else "p.input_per_mtok" for pr in await fetch_all(conn, f"select distinct on (p.model_id) {PRICE_COLS} from {PRICE_FROM} where p.valid_to is null and {col} > 0 and p.model_id = any(cast(:ids as text[])) order by p.model_id, {col} asc", ids=ids): offers[pr["m_id"]] = deployment_row(pr) org_id = await resolve_id(conn, org) if org else None fam_ids: set[str] | None = None if family: fr = await fetch_all(conn, "select e.id from entities e left join entities f on f.id = e.family_id where e.entity_type = 'model' and (f.slug = :f or f.id = :f or e.attributes->>'family' ilike :f)", f=family) fam_ids = {r["id"] for r in fr} open_vals = set(openness_values(openness)) if openness else None points = [] for r in ranked: attrs = r.get("model_attrs") or {} if org_id and r.get("organization_id") != org_id: continue if fam_ids is not None and r["model_id"] not in fam_ids: continue if open_vals and str(attrs.get("openness") or "") not in open_vals: continue xv: float | None provider = None if x == "output_price": pr = prices.get(r["model_id"]) xv = _f(pr["min_output"]) if pr else None provider = (offers.get(r["model_id"]) or {}).get("provider") elif x == "input_price": pr = prices.get(r["model_id"]) xv = _f(pr["min_input"]) if pr else None provider = (offers.get(r["model_id"]) or {}).get("provider") elif x == "parameter_count": xv = _f(attrs.get("parameter_count")) elif x == "context_length": xv = _f(attrs.get("context_length")) else: pc = hf.parameter_count(attrs) xv = hf.estimate_memory_gb(pc, "4bit", 8192) if pc else None if xv is None: continue points.append({"id": r["model_id"], "model": {"id": r["model_id"], "slug": r["model_slug"], "name": r["model_name"], "organization": r.get("org_name"), "openness": attrs.get("openness")}, "x": xv, "y": float(r["score"]), "rank": r["rank"], "trust_level": r["trust_level"], "config": {k: v for k, v in (r.get("config") or {}).items() if k in ("reasoning_effort", "reasoning", "variant", "evaluator")}, "context_length": _f(attrs.get("context_length")), "parameter_count": _f(attrs.get("parameter_count")), "release_date": attrs.get("release_date"), **({"provider": provider} if provider else {}), **({"estimated": True} if x == "memory_estimate" else {})}) front = pareto_frontier(points, maximize_y=g["higher_is_better"]) fset = set(front) for p in points: p["pareto"] = p["id"] in fset return {"benchmark": entity_summary(bench), "group": group_summary(g), "groups": [group_summary(gg) for gg in groups], "x": {"key": x, "label": PARETO_X[x]}, "y": {"key": "score", "label": f"{g['metric']} on {bench['canonical_name']}"}, "points": sorted(points, key=lambda p: p["x"]), "frontier": front, "methodology": f"Points are the best current row per canonical model in comparability group '{g['label']}'. Price = cheapest current offer across providers " f"(the provider shown). Pareto frontier maximises the score and minimises x; exact ties are all kept. memory_estimate is an estimate (see /methodology)."} # ------------------------------------------------------------------------------------------------------------------ /pulse @router.get("/pulse") @cached(120) async def pulse(request: Request, days: int = Query(7, ge=1, le=90)) -> dict[str, Any]: now = datetime.now(UTC) since = now - timedelta(days=days) async with connection() as conn: c = await fetch_one(conn, f""" with ev as (select ev.*, e.entity_type as et, e.attributes as attrs from change_events ev left join entities e on e.id = ev.entity_id where ev.is_backfill = false and ev.occurred_at > :since) select (select count(distinct entity_id) from ev where event_type = 'NEW_MODEL' and et = 'model') as new_models, (select count(distinct entity_id) from ev where event_type = 'NEW_MODEL' and et = 'model' and attrs->>'openness' in ('open-weights','open-source')) as new_open_models, (select count(*) from entities where entity_type = 'artifact' and merged_into is null and first_seen_at > :since) as new_artifacts, (select count(distinct entity_id) from ev where event_type = 'NEW_PAPER') as new_papers, (select count(*) from ev where event_type = 'PROVIDER_LISTED') as provider_listings, (select count(*) from ev where event_type = 'PROVIDER_DELISTED') as provider_delistings, (select count(*) from ev where event_type = 'PRICE_CHANGED') as price_changes, (select count(distinct entity_id) from ev where event_type = 'NEW_MODEL' and et = 'model' and {num_expr("attrs->>'context_length'")} >= 1000000) as new_models_1m_context, (select count(*) from ev where event_type = 'DOCUMENT_CHANGED') as documents_changed, (select count(distinct d.source_id) from snapshots s join documents d on d.id = s.document_id where s.observed_at > :since) as sources_observed, (select count(*) from ev where event_type <> 'DOCUMENT_CHANGED') as events_total""", since=since) pct = await fetch_all(conn, f"select {EVENT_COLS}, ev.occurred_at from {EVENT_FROM} where ev.is_backfill = false and ev.occurred_at > :since and ev.event_type = 'PRICE_CHANGED' " f"order by ev.occurred_at desc limit 500", since=since) ctx_models = await fetch_all(conn, f"""select distinct on (e.id) {ENTITY_COLS} from {EVENT_FROM} where ev.is_backfill = false and ev.occurred_at > :since and ev.event_type = 'NEW_MODEL' and e.entity_type = 'model' and e.merged_into is null and {CONTEXT} >= 1000000 order by e.id limit 100""", since=since) rows = await load_results(conn, current_only=False) meta = await benchmark_meta(conn) moves: list[float] = [] price_items: list[dict[str, Any]] = [] for r in pct: ev = change_event(r) d = price_delta(ev.get("old_value"), ev.get("new_value")) if d and d["percent"] is not None: moves.append(float(d["percent"])) price_items.append({"id": ev["id"], "summary": ev["summary"], "model": ev["entity"], "provider": (ev.get("meta") or {}).get("provider"), "provider_id": (ev.get("meta") or {}).get("provider_id"), "occurred_at": r.get("occurred_at"), "percent_change": d["percent"] if d else None, "delta": d, "source_url": ev.get("source_url")}) by_bench: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in rows: by_bench[r["benchmark_id"]].append(r) new_leaders = [] for bid, brows in by_bench.items(): m = meta.get(bid, {}) la, lb = leader_at(brows, since, m.get("attributes")), leader_at(brows, now, m.get("attributes")) if lb and (la is None or la["model"]["id"] != lb["model"]["id"]): new_leaders.append({"benchmark": {"id": bid, "slug": m.get("slug"), "name": m.get("name")}, "previous": la, "current": lb}) c = c or {} counters = { "new_models": {"value": int(c.get("new_models") or 0), "definition": "canonical models with a NEW_MODEL event that occurred in the window (not back-filled)"}, "new_open_weight_models": {"value": int(c.get("new_open_models") or 0), "definition": "subset of new_models with openness open-weights / open-source"}, "new_artifacts": {"value": int(c.get("new_artifacts") or 0), "definition": "artifact entities (checkpoints, quantisations, conversions) first seen in the window"}, "new_papers": {"value": int(c.get("new_papers") or 0), "definition": "papers with a NEW_PAPER event that occurred in the window"}, "provider_listings": {"value": int(c.get("provider_listings") or 0), "definition": "PROVIDER_LISTED events in the window"}, "provider_delistings": {"value": int(c.get("provider_delistings") or 0), "definition": "PROVIDER_DELISTED events in the window"}, "price_changes": {"value": int(c.get("price_changes") or 0), "median_percent": round(statistics.median(moves), 2) if moves else None, "items": price_items[:50], "definition": "PRICE_CHANGED events in the window; % change = output price (else input) new vs old from the event's old/new values; median over events with both values"}, "new_models_1m_context": {"value": int(c.get("new_models_1m_context") or 0), "items": [entity_summary(m) for m in ctx_models], "definition": "new_models whose context_length is at least 1 000 000 tokens"}, "new_benchmark_leaders": {"value": len(new_leaders), "items": new_leaders, "definition": "benchmarks whose primary-group leader (computed from results observed by each date) changed over the window"}, "documents_changed": {"value": int(c.get("documents_changed") or 0), "definition": "DOCUMENT_CHANGED events in the window"}, "sources_observed": {"value": int(c.get("sources_observed") or 0), "definition": "distinct sources with at least one snapshot taken in the window"}, "events_total": {"value": int(c.get("events_total") or 0), "definition": "all non-backfill events (excluding source-document changes) that occurred in the window"}, } return {"days": days, "since": since, "until": now, "counters": counters, "note": "Deterministic counters over events that OCCURRED in the window and are not back-filled history; each counter carries its own definition."} # ------------------------------------------------------------------------------------------------------------------ /open @router.get("/open") @cached(300) async def open_models(request: Request, sort: str = Query("release", pattern="^(release|params|context|rank|name|downloads)$"), license: str | None = None, min_params: float | None = Query(None, ge=0), max_params: float | None = Query(None, ge=0), min_context: int | None = Query(None, ge=0), modality: str | None = None, days: int | None = Query(None, ge=1, le=3650), limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0), openness: str | None = None) -> dict[str, Any]: """Open-weight / open-source / restricted-weights canonical models with licence permissions, observed dimensions, best results, estimated hardware fit.""" where = ["e.entity_type = 'model'", "e.merged_into is null", "e.attributes->>'openness' = any(cast(:o as text[]))"] params: dict[str, Any] = {"o": openness_values(openness) if openness else list(DOWNLOADABLE_CATEGORIES)} if license: from aiatlas.api.routers.models import license_match_sql, license_params where.append(license_match_sql()) params.update(license_params(license)) if min_params is not None: where.append(f"{PARAMS} >= :minp") params["minp"] = float(min_params) if max_params is not None: where.append(f"{PARAMS} <= :maxp") params["maxp"] = float(max_params) if min_context is not None: where.append(f"{CONTEXT} >= :minc") params["minc"] = float(min_context) if modality: where.append("(e.attributes->'modalities' ? :mod or e.attributes->'modalities_input' ? :mod or e.attributes->'modalities_output' ? :mod)") params["mod"] = modality if days: where.append("(e.attributes->>'release_date' >= :since or e.first_seen_at > now() - make_interval(days => :days))") params["since"] = (datetime.now(UTC) - timedelta(days=days)).date().isoformat() params["days"] = days order = {"release": "e.attributes->>'release_date' desc nulls last", "params": f"{PARAMS} desc nulls last", "context": f"{CONTEXT} desc nulls last", "name": "e.canonical_name asc", "downloads": num_expr("e.attributes->>'metric.downloads'") + " desc nulls last", "rank": "e.canonical_name asc"}[sort] where_sql = " and ".join(where) async with connection() as conn: rows = await fetch_all(conn, f"select {ENTITY_COLS}, e.family_id from {ENTITY_FROM} where {where_sql} order by {order}, e.id limit :lim offset :off", lim=limit if sort != "rank" else 2000, off=0 if sort == "rank" else offset, **params) total = await fetch_val(conn, f"select count(*) from {ENTITY_FROM} where {where_sql}", **params) groups = await all_primary_groups(conn) prices = await _cheapest_prices(conn, [r["id"] for r in rows]) summary = await fetch_one(conn, """select jsonb_object_agg(k, n) as by_cat from (select attributes->>'openness' as k, count(*) as n from entities where entity_type = 'model' and merged_into is null and attributes->>'openness' in ('open-weights','open-source','restricted-weights','restricted') group by 1) x""") lic_rows = await fetch_all(conn, """select coalesce(attributes->>'license_key', attributes->>'license') as raw, count(*) as n from entities where entity_type = 'model' and merged_into is null and attributes->>'openness' in ('open-weights','open-source','restricted-weights','restricted') and (attributes ? 'license' or attributes ? 'license_key') group by 1""") # released in the last 30 days (release_date when known; first_seen_at only for models without any release date) new_30d = await fetch_val(conn, """select count(*) from entities where entity_type = 'model' and merged_into is null and attributes->>'openness' in ('open-weights','open-source','restricted-weights','restricted') and case when attributes->>'release_date' ~ '^\\d{4}' then left(attributes->>'release_date', 10) >= to_char((now() at time zone 'UTC') - interval '30 days', 'YYYY-MM-DD') else first_seen_at > now() - interval '30 days' end""") ranks: dict[str, dict[str, int]] = defaultdict(dict) for g in groups.values(): for r in rank_rows(g["rows"], g["higher_is_better"]): ranks[r["model_id"]][g["benchmark"]["slug"]] = r["rank"] items = [] for r in rows: attrs = r["attributes"] or {} key = attrs.get("license_key") or normalize_license(attrs.get("license")) info = LICENSES.get(key) if key else None rk = ranks.get(r["id"], {}) best = sorted(rk.items(), key=lambda kv: kv[1])[:3] pr = prices.get(r["id"]) items.append({"model": entity_summary(r), "licence": {**info.as_dict(), "raw": attrs.get("license")} if info else {"key": None, "raw": attrs.get("license"), "note": "not classified"}, "dimensions": {"parameter_count": _f(attrs.get("parameter_count")), "active_parameter_count": _f(attrs.get("active_parameter_count")), "context_length": _f(attrs.get("context_length")), "modalities": sorted(_mods(attrs)), "release_date": attrs.get("release_date"), "openness": attrs.get("openness"), "downloads": _f(attrs.get("metric.downloads"))}, "best_results": [{"benchmark": b, "rank": k} for b, k in best], "best_rank": best[0][1] if best else None, "hardware_fit": {"4bit_64gb": hf.fit_detailed(attrs, 64, quant="4bit", context=8192), "8bit_128gb": hf.fit_detailed(attrs, 128, quant="8bit", context=8192), "estimated": True}, "providers": int(pr["providers"]) if pr else 0, "cheapest_output_per_mtok": _f(pr["min_output"]) if pr else None}) if sort == "rank": items.sort(key=lambda x: (x["best_rank"] if x["best_rank"] is not None else 10_000, x["model"]["name"] or "")) items = items[offset:offset + limit] lic_counts: dict[str, int] = defaultdict(int) for lr in lic_rows: k = lr["raw"] if lr["raw"] in LICENSES else normalize_license(lr["raw"]) lic_counts[k or f"raw:{lr['raw']}"] += int(lr["n"]) return {"items": items, "total": int(total or 0), "limit": limit, "offset": offset, "summary": {"by_category": (summary or {}).get("by_cat") or {}, "by_license_top": [{"key": k, "label": LICENSES[k].label if k in LICENSES else k, "models": n} for k, n in sorted(lic_counts.items(), key=lambda kv: -kv[1])[:12]], "new_30d": int(new_30d or 0), "new_30d_definition": "release_date within the last 30 days (first_seen_at only when no release date is known)"}, "note": "Universe = canonical models whose weights can be downloaded (open-weights, open-source, restricted-weights). hardware_fit values are ESTIMATES (see /methodology); " "best_results are ranks inside each benchmark's primary comparability group — no composite score."} # ------------------------------------------------------------------------------------------------------------------ /find-a-model @router.get("/find-a-model") @cached(300) async def find_a_model(request: Request, use_case: str | None = Query(None), deployment: str = Query("any", pattern="^(local|api|any)$"), memory_gb: float | None = Query(None, gt=0, le=100000), quant: str = Query("4bit"), context_min: int | None = Query(None, ge=0), license: str = Query("any", pattern="^(commercial|any)$"), openness: str | None = None, max_input_price: float | None = Query(None, ge=0), max_output_price: float | None = Query(None, ge=0), modalities: str | None = None, limit: int = Query(30, ge=1, le=100)) -> dict[str, Any]: if use_case and use_case not in USE_CASES: raise ApiError(400, f"use_case must be one of {', '.join(USE_CASES)}") async with connection() as conn: out = await find_models(conn, use_case=use_case, deployment=deployment, memory_gb=memory_gb, quant=quant, context_min=context_min, license=license, openness=openness_values(openness) if openness else None, max_input_price=max_input_price, max_output_price=max_output_price, modalities=csv(modalities), limit=limit) if deployment != "local": ids = [m["model"]["id"] for m in out["matches"]] rows = await fetch_all(conn, f"""select * from (select {PRICE_COLS}, row_number() over (partition by p.model_id order by p.output_per_mtok asc nulls last) as rn from {PRICE_FROM} where p.valid_to is null and p.model_id = any(cast(:ids as text[]))) x where rn <= 5""", ids=ids) by: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in rows: by[r["m_id"]].append(deployment_row(r)) for m in out["matches"]: m["deployments"] = by.get(m["model"]["id"], []) return out # ------------------------------------------------------------------------------------------------------------------ /run-locally @router.get("/run-locally") @cached(300) async def run_locally(request: Request, memory_gb: float = Query(..., gt=0, le=100000), gpu_count: int = Query(1, ge=1, le=8), quant: str = Query("4bit"), context: int = Query(8192, ge=0, le=10_000_000), batch: int = Query(1, ge=1, le=256), platform: str = Query("any", pattern="^(apple|nvidia|amd|any)$"), use_case: str | None = None, limit: int = Query(60, ge=1, le=300), openness: str | None = None) -> dict[str, Any]: """Canonical models (and their compatible artifacts) whose ESTIMATED footprint fits `memory_gb × gpu_count`.""" if gpu_count not in (1, 2, 4, 8): raise ApiError(400, "gpu_count must be 1, 2, 4 or 8") q = hf.normalize_quant(quant) fmt_pref = {"apple": ("mlx", "gguf"), "nvidia": ("gguf", "awq", "gptq", "fp8", "nvfp4", "int4", "int8"), "amd": ("gguf", "mxfp4", "int8"), "any": ()}[platform] where = ["e.entity_type = 'model'", "e.merged_into is null", "e.attributes ? 'parameter_count'", "e.attributes->>'openness' = any(cast(:o as text[]))"] params: dict[str, Any] = {"o": openness_values(openness) if openness else list(DOWNLOADABLE_CATEGORIES)} async with connection() as conn: rows = await fetch_all(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where {' and '.join(where)} order by {PARAMS} desc nulls last limit 3000", **params) arts = await fetch_all(conn, f""" select a.canonical_id as model_id, {ENTITY_COLS} from entities e join lateral (select e.canonical_id) a on true left join entities eo on eo.id = e.organization_id where e.entity_type = 'artifact' and e.merged_into is null and e.canonical_id is not null union all select r.object_id as model_id, {ENTITY_COLS} from relations r join entities e on e.id = r.subject_id left join entities eo on eo.id = e.organization_id where r.predicate = 'quantized_from' and r.valid_to is null and e.merged_into is null and (e.attributes->>'is_quantized' = 'true' or e.attributes ? 'quant_format')""") groups = await all_primary_groups(conn) if use_case else {} cat_models: set[str] | None = None if use_case: cat = {"coding": "coding", "reasoning": "reasoning", "agentic": "agentic", "vision": "multimodal", "math": "math"}.get(use_case, use_case) cat_models = {r["model_id"] for g in groups.values() if (g["benchmark"].get("category") or "").lower() == cat for r in g["rows"]} art_by: dict[str, list[dict[str, Any]]] = defaultdict(list) for a in arts: art_by[a["model_id"]].append(a) items = [] for r in rows: if cat_models is not None and r["id"] not in cat_models: continue attrs = r["attributes"] or {} f = hf.fit_detailed(attrs, memory_gb, quant=q, context=context, batch=batch, gpu_count=gpu_count) if not f: continue compatible = [] for a in art_by.get(r["id"], []): aa = a["attributes"] or {} fmt = str(aa.get("quant_format") or "").lower() if fmt_pref and fmt and fmt not in fmt_pref: continue size = hf.file_size_gb(aa) af = hf.fit_detailed({**attrs, **{k: v for k, v in aa.items() if k in ("num_hidden_layers", "num_key_value_heads", "num_attention_heads", "head_dim", "hidden_size")}}, memory_gb, quant=q, context=context, batch=batch, gpu_count=gpu_count, observed_size_gb=size) compatible.append({"artifact": entity_summary(a), "quant_format": fmt or None, "file_size_gb": size, "weights_source": "observed" if size else "estimated", "fit": af}) compatible.sort(key=lambda x: (not (x["fit"] or {}).get("fits", False), x["file_size_gb"] or 1e9)) items.append({"model": entity_summary(r), "fit": f, "artifacts": compatible[:8], "artifact_count": len(art_by.get(r["id"], []))}) fits = [i for i in items if i["fit"]["fits"]] fits.sort(key=lambda i: -(i["fit"].get("parameter_count") or 0)) return {"inputs": {"memory_gb": memory_gb, "gpu_count": gpu_count, "total_memory_gb": memory_gb * gpu_count, "quant": q, "context": context, "batch": batch, "platform": platform, "use_case": use_case}, "estimated": True, "assumptions": hf.ASSUMPTIONS, "counts": {"fits": len(fits), "evaluated": len(items)}, "items": fits[:limit], "note": "Every figure is an ESTIMATE: weights = params × bytes/param (× 1.15 overhead) unless an artifact's observed file size is available; KV cache uses architecture metadata when known, " + ("else 0.5 GB per 8K tokens × batch. Multi-GPU sums device memory and ignores interconnect." if gpu_count > 1 else "else 0.5 GB per 8K tokens × batch.")} # ------------------------------------------------------------------------------------------------------------------ /time-machine @router.get("/time-machine") @cached(300) async def time_machine(request: Request, date: str = Query(..., description="YYYY-MM-DD"), scope: str = Query("models", pattern="^(models|prices|benchmarks|hardware|all)$"), limit: int = Query(50, ge=1, le=300)) -> dict[str, Any]: d = parse_date(date, "date") assert d is not None at = datetime.combine(d, dtime.max, UTC) out: dict[str, Any] = {"date": d.isoformat(), "scope": scope} async with connection() as conn: first = await fetch_val(conn, "select min(first_seen_at) from entities") reconstructed = first is not None and at < first out["first_entity_at"] = first out["reconstructed"] = bool(reconstructed) out["note"] = (f"AI Atlas observation history starts at {first.isoformat() if first else 'n/a'}. " + ("This date is earlier: the state is RECONSTRUCTED from claims with effective dates and from release dates — not from direct observation." if reconstructed else "The state is taken from claims, prices and results as they were known at the end of that UTC day.")) if scope in ("models", "all"): rows = await fetch_all(conn, f"""select {ENTITY_COLS}, (e.first_seen_at <= :at) as observed_then from {ENTITY_FROM} where e.entity_type = 'model' and e.merged_into is null and (e.first_seen_at <= :at or (e.attributes->>'release_date' ~ '^\\d{{4}}' and left(e.attributes->>'release_date', 10) <= :d)) order by e.attributes->>'release_date' desc nulls last, e.canonical_name limit :lim""", at=at, d=d.isoformat(), lim=limit) ids = [r["id"] for r in rows] claims = await fetch_all(conn, f"""select distinct on (c.entity_id, c.property) c.entity_id, {CLAIM_COLS} from {CLAIM_FROM} where c.entity_id = any(cast(:ids as text[])) and c.status <> 'retracted' and coalesce(c.effective_at, c.valid_from) <= :at and (c.valid_to is null or c.valid_to > :at) order by c.entity_id, c.property, c.tier, c.valid_from desc""", ids=ids, at=at) if ids else [] by: dict[str, dict[str, Any]] = defaultdict(dict) for c in claims: by[c["entity_id"]][c["property"]] = c["value"] total = await fetch_val(conn, """select count(*) from entities e where e.entity_type = 'model' and e.merged_into is null and (e.first_seen_at <= :at or (e.attributes->>'release_date' ~ '^\\d{4}' and left(e.attributes->>'release_date', 10) <= :d))""", at=at, d=d.isoformat()) out["models"] = {"items": [{"model": entity_summary(r), "attributes_as_of": by.get(r["id"], {}), "observed_then": bool(r["observed_then"]), "reconstructed": not r["observed_then"]} for r in rows], "total": int(total or 0), "limit": limit} if scope in ("prices", "all"): prs = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where p.valid_from <= :at and (p.valid_to is null or p.valid_to > :at) order by p.output_per_mtok nulls last limit :lim", at=at, lim=limit) n = await fetch_val(conn, "select count(*) from prices p where p.valid_from <= :at and (p.valid_to is null or p.valid_to > :at)", at=at) out["prices"] = {"items": [deployment_row(p) for p in prs], "total": int(n or 0), "note": "offers whose validity interval covers the date (price rows are append-only)"} if scope in ("benchmarks", "all"): rows_r = await load_results(conn, current_only=False) meta = await benchmark_meta(conn) by_b: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in rows_r: by_b[r["benchmark_id"]].append(r) leaders = [] for bid, brows in by_b.items(): la = leader_at(brows, at, meta.get(bid, {}).get("attributes")) if la: leaders.append({"benchmark": {"id": bid, "slug": meta.get(bid, {}).get("slug"), "name": meta.get(bid, {}).get("name")}, "leader": la}) out["benchmarks"] = {"leaders": sorted(leaders, key=lambda x: x["benchmark"]["name"] or ""), "note": "leaders from results observed by the date (evaluation dates are not used: a result is known only once observed)"} if scope in ("hardware", "all"): hw = await fetch_all(conn, f"""select {ENTITY_COLS}, (e.first_seen_at <= :at) as observed_then from {ENTITY_FROM} where e.entity_type = 'hardware' and e.merged_into is null and (e.first_seen_at <= :at or (e.attributes->>'release_date' ~ '^\\d{{4}}' and left(e.attributes->>'release_date', 7) <= :d)) order by e.attributes->>'release_date' desc nulls last limit :lim""", at=at, d=d.isoformat()[:7], lim=limit) out["hardware"] = {"items": [{"hardware": entity_summary(r), "reconstructed": not r["observed_then"]} for r in hw]} return out __all__ = ["PARETO_X", "router"]