"""/prices · /prices/history · /prices/index — append-only pricing table (`valid_to` closes a row). API 1.1: `/prices` gains `family=`, `org=`, `modality=` filters and `sort=cheapest_frontier`; `/prices/index` becomes the AI Price Index (daily medians for all / frontier / open / embedding offers, cheapest frontier offer, current distribution, listings and delistings).""" from __future__ import annotations from typing import Any from fastapi import APIRouter, Query, Request from aiatlas.api.common import ( EVENT_COLS, EVENT_FROM, OPEN_CATEGORIES, PAGINATION, PRICE_COLS, PRICE_FROM, ApiError, Pagination, cached, change_event, deployment_row, page, price_row, resolve_id, ) from aiatlas.db import connection, fetch_all, fetch_one, fetch_val from aiatlas.services.frontier import FRONTIER_METHODOLOGY, frontier_model_ids router = APIRouter(prefix="/api/v1/prices", tags=["prices"]) SORTS = {"input": "p.input_per_mtok asc nulls last", "output": "p.output_per_mtok asc nulls last", "model": "m.canonical_name asc", "provider": "pv.canonical_name asc", "observed": "p.observed_at desc", "valid_from": "p.valid_from desc", "cheapest_frontier": "p.output_per_mtok asc nulls last"} BUCKETS = [(0, 0.1), (0.1, 0.5), (0.5, 1), (1, 2), (2, 5), (5, 10), (10, 20), (20, 50), (50, None)] @router.get("") @cached(300) async def list_prices(request: Request, p: Pagination = PAGINATION, model: str | None = None, provider: str | None = None, sort: str = "model", order: str = "", current: int = Query(1, ge=0, le=1), family: str | None = None, org: str | None = None, modality: str | None = None) -> dict[str, Any]: if sort not in SORTS: raise ApiError(400, f"sort must be one of {', '.join(SORTS)}") order_sql = SORTS[sort] if order == "desc" and " asc" in order_sql: order_sql = order_sql.replace(" asc", " desc") elif order == "asc" and " desc" in order_sql: order_sql = order_sql.replace(" desc", " asc") where = ["p.valid_to is null"] if current else ["true"] where.append("m.entity_type in ('model','artifact')") params: dict[str, Any] = {} frontier_note = None async with connection() as conn: if model: where.append("p.model_id = :model") params["model"] = await resolve_id(conn, model) if provider: where.append("p.provider_id = :provider") params["provider"] = await resolve_id(conn, provider) if family: where.append("(exists (select 1 from entities f where f.id = m.family_id and (f.slug = :family or f.id = :family or f.canonical_name ilike :family)) or m.attributes->>'family' ilike :family)") params["family"] = family if org: where.append("(mo.slug = :org or mo.id = :org or mo.canonical_name ilike :org)") params["org"] = org if modality: where.append("(m.attributes->'modalities' ? :modality or m.attributes->'modalities_input' ? :modality or m.attributes->'modalities_output' ? :modality)") params["modality"] = modality if sort == "cheapest_frontier": ids, _ = await frontier_model_ids(conn) where.append("p.model_id = any(cast(:frontier as text[]))") where.append("p.output_per_mtok > 0") params["frontier"] = sorted(ids) frontier_note = FRONTIER_METHODOLOGY where_sql = " and ".join(where) rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where {where_sql} order by {order_sql}, p.id limit :lim offset :off", lim=p.limit, off=p.offset, **params) total = await fetch_val(conn, f"select count(*) from {PRICE_FROM} where {where_sql}", **params) out = page([price_row(r) for r in rows], int(total or 0), p) if frontier_note: out["methodology"] = frontier_note return out @router.get("/history") @cached(300) async def price_history(request: Request, model: str | None = None, provider: str | None = None, limit: int = Query(2000, ge=1, le=5000)) -> dict[str, Any]: if not model and not provider: raise ApiError(400, "model or provider is required") where = ["true"] params: dict[str, Any] = {"lim": limit} async with connection() as conn: if model: where.append("p.model_id = :model") params["model"] = await resolve_id(conn, model) if provider: where.append("p.provider_id = :provider") params["provider"] = await resolve_id(conn, provider) rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where {' and '.join(where)} order by p.valid_from asc, p.id limit :lim", **params) return {"items": [price_row(r) for r in rows]} @router.get("/index") @cached(600) async def price_index(request: Request, days: int = Query(180, ge=7, le=1825)) -> dict[str, Any]: """AI Price Index: daily medians of live USD-per-1M-token prices, by universe (all offers / frontier / open-weight / embeddings).""" async with connection() as conn: frontier, composition = await frontier_model_ids(conn) open_ids = [r["id"] for r in await fetch_all(conn, "select id from entities where entity_type = 'model' and merged_into is null and attributes->>'openness' = any(cast(:o as text[]))", o=list(OPEN_CATEGORIES))] emb_ids = [r["id"] for r in await fetch_all(conn, """select id from entities where entity_type = 'model' and merged_into is null and (attributes->'modalities' ? 'embedding' or attributes->'modalities_output' ? 'embedding' or attributes->>'pipeline_tag' ilike '%embedding%')""")] series = await fetch_all(conn, """ with days as (select generate_series(((now() at time zone 'UTC') - make_interval(days => :d))::date::timestamp, (now() at time zone 'UTC')::date::timestamp, interval '1 day')::date as day), live as (select p.model_id, p.provider_id, p.input_per_mtok, p.output_per_mtok, p.valid_from, p.valid_to, p.model_id = any(cast(:frontier as text[])) as is_frontier, p.model_id = any(cast(:open as text[])) as is_open, p.model_id = any(cast(:emb as text[])) as is_emb from prices p join entities m on m.id = p.model_id where m.entity_type in ('model','artifact') and m.merged_into is null and ((p.input_per_mtok is not null and p.input_per_mtok > 0) or (p.output_per_mtok is not null and p.output_per_mtok > 0)) and p.valid_from < now()) select d.day, percentile_cont(0.5) within group (order by l.input_per_mtok) filter (where l.input_per_mtok > 0) as median_input, percentile_cont(0.5) within group (order by l.output_per_mtok) filter (where l.output_per_mtok > 0) as median_output, percentile_cont(0.5) within group (order by l.output_per_mtok) filter (where l.output_per_mtok > 0 and l.is_frontier) as median_frontier_output, percentile_cont(0.5) within group (order by l.input_per_mtok) filter (where l.input_per_mtok > 0 and l.is_frontier) as median_frontier_input, percentile_cont(0.5) within group (order by l.output_per_mtok) filter (where l.output_per_mtok > 0 and l.is_open) as median_open_output, percentile_cont(0.5) within group (order by l.input_per_mtok) filter (where l.input_per_mtok > 0 and l.is_emb) as median_embedding_input, min(l.input_per_mtok) filter (where l.input_per_mtok > 0) as min_input, max(l.input_per_mtok) as max_input, min(l.output_per_mtok) filter (where l.output_per_mtok > 0 and l.is_frontier) as min_frontier_output, count(distinct l.model_id) as models, count(l.model_id) as offers, count(distinct l.model_id) filter (where l.is_frontier) as frontier_models, count(l.model_id) filter (where l.is_frontier) as frontier_offers, count(distinct l.model_id) filter (where l.is_open) as open_models, count(distinct l.model_id) filter (where l.is_emb) as embedding_models from days d left join live l on l.valid_from < (d.day + 1)::timestamp at time zone 'UTC' and (l.valid_to is null or l.valid_to >= (d.day + 1)::timestamp at time zone 'UTC') group by d.day order by d.day""", d=days, frontier=sorted(frontier), open=open_ids, emb=emb_ids) movers = await fetch_all(conn, f"select {EVENT_COLS} from {EVENT_FROM} where ev.category = 'price' and ev.is_backfill = false and ev.occurred_at > now() - make_interval(days => :d) " f"order by ev.importance desc, ev.occurred_at desc limit 30", d=days) 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[])) " f"order by p.output_per_mtok asc, p.input_per_mtok asc nulls last limit 1", ids=sorted(frontier)) 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(frontier)) current_outputs = await fetch_all(conn, "select p.output_per_mtok as v from prices p join entities m on m.id = p.model_id where p.valid_to is null and p.output_per_mtok > 0 and m.merged_into is null") new_where = ("p.valid_from > now() - interval '30 days' and not exists (select 1 from prices q where q.model_id = p.model_id and q.provider_id = p.provider_id and q.valid_from < p.valid_from) " "and m.entity_type in ('model','artifact') and m.merged_into is null") gone_where = "p.valid_to > now() - interval '30 days' and not exists (select 1 from prices q where q.model_id = p.model_id and q.provider_id = p.provider_id and q.valid_to is null)" new_rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where {new_where} order by p.valid_from desc limit 50") new_n = await fetch_val(conn, f"select count(*) from {PRICE_FROM} where {new_where}") gone_rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where {gone_where} order by p.valid_to desc limit 50") gone_n = await fetch_val(conn, f"select count(*) from {PRICE_FROM} where {gone_where}") chg_rows = await fetch_all(conn, f"select {EVENT_COLS}, ev.occurred_at from {EVENT_FROM} where ev.category = 'price' and ev.is_backfill = false and ev.occurred_at > now() - interval '30 days' " f"order by ev.occurred_at desc limit 50") chg_n = await fetch_val(conn, "select count(*) from change_events where category = 'price' and is_backfill = false and occurred_at > now() - interval '30 days'") dist = [] vals = [float(r["v"]) for r in current_outputs] for lo, hi in BUCKETS: n = sum(1 for v in vals if v >= lo and (hi is None or v < hi)) dist.append({"from": lo, "to": hi, "label": f"${lo:g}–${hi:g}" if hi is not None else f"≥ ${lo:g}", "offers": n}) out_series = [] for r in series: out_series.append({"day": r["day"].isoformat(), "median_input": r["median_input"], "median_output": r["median_output"], "median_frontier_output": r["median_frontier_output"], "median_frontier_input": r["median_frontier_input"], "median_open_output": r["median_open_output"], "median_embedding_input": r["median_embedding_input"], "min_input": r["min_input"], "max_input": r["max_input"], "min_frontier_output": r["min_frontier_output"], "models": int(r["models"] or 0), "offers": int(r["offers"] or 0), "sample": {"models": int(r["models"] or 0), "offers": int(r["offers"] or 0), "frontier_models": int(r["frontier_models"] or 0), "frontier_offers": int(r["frontier_offers"] or 0), "open_models": int(r["open_models"] or 0), "embedding_models": int(r["embedding_models"] or 0)}}) return {"days": days, "series": out_series, "movers": [change_event(r) for r in movers], "cheapest_frontier": _cheap(cheapest), "cheapest_frontier_1m_context": _cheap(cheapest_1m), "distribution": {"metric": "output_per_mtok", "unit": "USD per 1M tokens", "buckets": dist, "offers": len(vals)}, "new_listings_30d": {"count": int(new_n or 0), "items": [deployment_row(r) for r in new_rows], "definition": "offers whose first row for (model, provider) opened in the last 30 days"}, "delistings_30d": {"count": int(gone_n or 0), "items": [deployment_row(r) for r in gone_rows], "definition": "offers closed in the last 30 days with no live successor for (model, provider)"}, "price_changes_30d": {"count": int(chg_n or 0), "items": [_price_event(r) for r in chg_rows], "definition": "price events (not back-filled) that occurred in the last 30 days"}, "frontier": {"composition": composition, "methodology": FRONTIER_METHODOLOGY}, "methodology": "Daily medians of live USD-per-1M-token prices across every provider offer valid at the end of each UTC day (offers with a zero or missing price are excluded " "from that median). `median_open_output` covers models with openness open-weights/open-source; `median_embedding_input` covers models whose modalities include " "embedding. Sample sizes are returned per day; a null median means no offer in that universe on that day. " + FRONTIER_METHODOLOGY, "note": "Daily medians of live USD-per-1M-token input/output prices across all provider offers valid at the end of each UTC day."} def _price_event(r: dict[str, Any]) -> dict[str, Any]: from aiatlas.api.routers.intelligence import price_delta ev = change_event(r) d = price_delta(ev.get("old_value"), ev.get("new_value")) ev["occurred_at"] = r.get("occurred_at") ev["percent_change"] = d["percent"] if d else None ev["price_delta"] = d ev["provider"] = (ev.get("meta") or {}).get("provider") return ev def _cheap(row: dict[str, Any] | None) -> dict[str, Any] | None: if not row: return None pr = price_row(row) return {"model": pr["model"], "provider": pr["provider"], "output": pr["output_per_mtok"], "input": pr["input_per_mtok"], "context_length": pr["context_length"], "price_id": pr["id"]} __all__ = ["BUCKETS", "router"]