#!/usr/bin/env python3 """ build_cn_indices.py ------------------- Build ComputeNav research indices from ComputeNav weekly price panels. Series: CN-TOKEN-SPOT — Floor / P25 / Median (USD / 1M blended) CN-FRONTIER — Fixed closed flagship basket geo-mean (USD) CN-OPEN — Fixed open-hosting basket geo-mean (USD) CN-ARB-SPREAD — Median cross-host relative range (%) CN-QA — Deferred stub in API only Methodology: handbook/compute/*.md Usage (from Web-GCPI Research repo root): python3 scripts/build_cn_indices.py """ from __future__ import annotations import argparse import csv import json import math import re import sys from collections import defaultdict from datetime import datetime, timezone from pathlib import Path REPO = Path(__file__).resolve().parent.parent DEFAULT_COMPUTENAV = REPO.parent / "Web-ComputeNav" BASE_DATE = "2026-06-20" SERIES_START_DATE = "2026-04-11" # drop sparse early panels (e.g. 2026-04-08) METHODOLOGY_VERSION = "0.3.0" INPUT_W, OUTPUT_W = 0.7, 0.3 PANEL_RE = re.compile(r"^prices_(\d{4}-\d{2}-\d{2})\.csv$") FRONTIER_CONFIG_PATH = REPO / "data" / "config" / "cn_frontier_seats.json" OPEN_GROUPS: list[str] = [ "deepseek-ai/DeepSeek-V3.2", "deepseek-ai/DeepSeek-R1", "Qwen/Qwen3.5-397B-A17B", "Qwen/Qwen3-235B-A22B-Instruct-2507", "meta-llama/Llama-3.3-70B-Instruct", "meta-llama/Llama-4-Maverick-17B-128E", "zai-org/GLM-5.1", "MiniMaxAI/MiniMax-M2.7", "openai/gpt-oss-120b", "mistralai/Mistral-Small-3.2-24B-Instruct-2506", "gemma-4-31B-it", "kimi-k2.5-0127", ] OPEN_MIN_GROUPS = 5 OPEN_PROVIDER_WHITELIST = frozenset( { "DeepInfra", "Together AI", "Together", "Fireworks AI", "Fireworks", "Groq", "SiliconFlow", "OpenRouter", } ) OPEN_OUTLIER_MULT = 5.0 ARB_MIN_GROUPS = 5 # soft; handbook prefers ≥20 def load_frontier_config(path: Path = FRONTIER_CONFIG_PATH) -> dict: with path.open(encoding="utf-8") as f: cfg = json.load(f) epochs = sorted(cfg["mapping_epochs"], key=lambda e: e["effective_from"]) cfg["mapping_epochs"] = epochs cfg["_seat_ids"] = [s["id"] for s in cfg["seats"]] cfg["_seat_labels"] = {s["id"]: s["label"] for s in cfg["seats"]} return cfg def frontier_map_for_date(cfg: dict, date: str) -> tuple[dict[str, str], dict]: """Return (seat_id → compare_group_id, epoch_meta) for panel date.""" chosen = cfg["mapping_epochs"][0] for ep in cfg["mapping_epochs"]: if ep["effective_from"] <= date: chosen = ep else: break return dict(chosen["map"]), chosen _FRONTIER_CFG: dict | None = None def frontier_cfg() -> dict: global _FRONTIER_CFG if _FRONTIER_CFG is None: _FRONTIER_CFG = load_frontier_config() return _FRONTIER_CFG def _utc_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _f(x: object) -> float | None: try: v = float(str(x).strip()) except (TypeError, ValueError): return None if not math.isfinite(v) or v <= 0: return None return v _EXCLUDED_MODALITIES = frozenset({"image", "video", "embedding"}) _TOKEN_UNITS = frozenset({"per_1m_tokens", "per_1m_token", "token", "tokens"}) def _is_chat_token_healthy(row: dict) -> bool: status = str(row.get("status") or "").strip().lower() if status != "healthy": return False mod = str(row.get("modality") or "").strip().lower() if mod in _EXCLUDED_MODALITIES: return False if mod and mod != "chat": return False unit = str(row.get("pricing_unit") or "").strip().lower() if not unit: return True if unit in _TOKEN_UNITS: return True if "token" in unit and "image" not in unit and "second" not in unit: return True return False def blended(row: dict) -> float | None: inp = _f(row.get("price_per_1m_input")) out = _f(row.get("price_per_1m_output")) if inp is None or out is None: return None return round(INPUT_W * inp + OUTPUT_W * out, 8) def percentile(sorted_vals: list[float], p: float) -> float: n = len(sorted_vals) if n == 1: return sorted_vals[0] k = (n - 1) * p lo = math.floor(k) hi = math.ceil(k) if lo == hi: return sorted_vals[int(k)] return sorted_vals[lo] * (hi - k) + sorted_vals[hi] * (k - lo) def geo_mean(vals: list[float]) -> float | None: if not vals: return None s = 0.0 for v in vals: if v <= 0 or not math.isfinite(v): return None s += math.log(v) return math.exp(s / len(vals)) def index_points(level: float, base_level: float | None) -> float | None: if base_level is None or base_level <= 0: return None return round(100.0 * level / base_level, 6) def discover_panels(prices_dirs: list[Path]) -> dict[str, Path]: found: dict[str, Path] = {} for d in prices_dirs: if not d.is_dir(): continue for p in sorted(d.glob("prices_*.csv")): m = PANEL_RE.match(p.name) if not m: continue date = m.group(1) if date not in found: found[date] = p return dict(sorted(found.items())) def load_eligible(path: Path) -> list[dict]: """Return eligible quote dicts with blended price and keys.""" out: list[dict] = [] with path.open(newline="", encoding="utf-8") as f: for row in csv.DictReader(f): if not _is_chat_token_healthy(row): continue p = blended(row) if p is None: continue group = str(row.get("compare_group_id") or "").strip() provider = str(row.get("compute_provider") or row.get("provider") or "").strip() source = str(row.get("model_source") or "").strip().lower() out.append( { "group": group, "provider": provider, "source": source, "price": p, } ) return out def provider_prices(quotes: list[dict]) -> dict[str, float]: """Median blended per compute_provider.""" by_p: dict[str, list[float]] = defaultdict(list) for q in quotes: if q["provider"]: by_p[q["provider"]].append(q["price"]) return {p: percentile(sorted(vs), 0.5) for p, vs in by_p.items() if vs} def spot_for_quotes(quotes: list[dict]) -> dict | None: prices = sorted(q["price"] for q in quotes) if not prices: return None return { "n_quotes": len(prices), "floor_usd": round(prices[0], 6), "p25_usd": round(percentile(prices, 0.25), 6), "median_usd": round(percentile(prices, 0.50), 6), } def frontier_for_quotes(quotes: list[dict], date: str, cfg: dict | None = None) -> dict | None: cfg = cfg or frontier_cfg() seat_map, epoch = frontier_map_for_date(cfg, date) min_seats = int(cfg.get("min_active_seats", 3)) by_g: dict[str, list[float]] = defaultdict(list) needed = set(seat_map.values()) for q in quotes: if q["group"] and q["group"] in needed: by_g[q["group"]].append(q["price"]) seat_prices: list[float] = [] active_roles: list[str] = [] active_groups: list[str] = [] mapping_bits: list[str] = [] for seat_id in cfg["_seat_ids"]: group = seat_map.get(seat_id) if not group: continue vals = by_g.get(group) mapping_bits.append(f"{seat_id}={group}") if not vals: continue seat_prices.append(percentile(sorted(vals), 0.5)) active_roles.append(seat_id) active_groups.append(group) if len(seat_prices) < min_seats: return None level = geo_mean(seat_prices) if level is None: return None return { "n_seats": len(seat_prices), "level_usd": round(level, 6), "active_seats": ",".join(active_roles), "active_groups": ",".join(active_groups), "seat_map": ";".join(mapping_bits), "mapping_epoch": epoch["effective_from"], "basket_version": cfg.get("basket_version", ""), } def _open_provider_ok(name: str) -> bool: if not name: return False if name in OPEN_PROVIDER_WHITELIST: return True # Loose match for Together / Fireworks naming drift low = name.lower() return any(w.lower() in low for w in ("deepinfra", "together", "fireworks", "groq", "siliconflow", "openrouter")) def open_for_quotes(quotes: list[dict]) -> dict | None: by_g: dict[str, list[dict]] = defaultdict(list) for q in quotes: if not q["group"] or q["group"] not in OPEN_GROUPS: continue if q["source"] and q["source"] != "open": continue if not _open_provider_ok(q["provider"]): continue by_g[q["group"]].append(q) group_prices: list[float] = [] active: list[str] = [] for group in OPEN_GROUPS: qs = by_g.get(group) or [] if not qs: continue # One price per provider, then drop >5× preliminary median outliers. pp = provider_prices(qs) if len(pp) < 2: continue vals = sorted(pp.values()) med0 = percentile(vals, 0.5) cleaned = [v for v in vals if v <= OPEN_OUTLIER_MULT * med0] if len(cleaned) < 2: continue # Re-check distinct providers after filter — use cleaned count as proxy if len(cleaned) < 2: continue group_prices.append(percentile(sorted(cleaned), 0.5)) active.append(group) if len(group_prices) < OPEN_MIN_GROUPS: return None level = geo_mean(group_prices) if level is None: return None return { "n_groups": len(group_prices), "level_usd": round(level, 6), "active_groups": ",".join(active), } def arb_for_quotes(quotes: list[dict]) -> dict | None: by_g: dict[str, list[dict]] = defaultdict(list) for q in quotes: if not q["group"]: continue by_g[q["group"]].append(q) rels: list[float] = [] ranges_usd: list[float] = [] for _g, qs in by_g.items(): pp = provider_prices(qs) if len(pp) < 2: continue vals = sorted(pp.values()) med = percentile(vals, 0.5) if med <= 0: continue rng = vals[-1] - vals[0] rels.append(rng / med) ranges_usd.append(rng) if len(rels) < ARB_MIN_GROUPS: return None rels.sort() ranges_usd.sort() s_rel = percentile(rels, 0.5) return { "n_groups": len(rels), "spread_pct": round(100.0 * s_rel, 4), "spread_range_usd": round(percentile(ranges_usd, 0.5), 6), } def attach_wow_and_index( rows: list[dict], level_key: str, index_key: str, ) -> list[dict]: base = next((r for r in rows if r["date"] == BASE_DATE), None) base_level = base[level_key] if base else (rows[0][level_key] if rows else None) out: list[dict] = [] prev: float | None = None for r in rows: level = r[level_key] wow = None if prev is None else round((level / prev - 1.0) * 100.0, 4) row = dict(r) row["wow_pct"] = wow row[index_key] = index_points(level, base_level) out.append(row) prev = level return out def build_all(panels: dict[str, Path]) -> dict[str, list[dict]]: spot_raw: list[dict] = [] frontier_raw: list[dict] = [] open_raw: list[dict] = [] arb_raw: list[dict] = [] for date, path in panels.items(): if date < SERIES_START_DATE: continue quotes = load_eligible(path) src_name = path.name spot = spot_for_quotes(quotes) if spot: spot_raw.append({"date": date, "source_file": src_name, **spot}) else: print(f"WARN: no Token Spot quotes in {path}", file=sys.stderr) fr = frontier_for_quotes(quotes, date) if fr: frontier_raw.append({"date": date, "source_file": src_name, **fr}) op = open_for_quotes(quotes) if op: open_raw.append({"date": date, "source_file": src_name, **op}) ar = arb_for_quotes(quotes) if ar: arb_raw.append({"date": date, "source_file": src_name, **ar}) # Token Spot keeps dedicated median_wow + multi index columns spot_out: list[dict] = [] base = next((r for r in spot_raw if r["date"] == BASE_DATE), None) base_med = base["median_usd"] if base else None base_floor = base["floor_usd"] if base else None base_p25 = base["p25_usd"] if base else None prev_med: float | None = None for r in spot_raw: med = r["median_usd"] wow = None if prev_med is None else round((med / prev_med - 1.0) * 100.0, 4) spot_out.append( { "date": r["date"], "n_quotes": r["n_quotes"], "floor_usd": r["floor_usd"], "p25_usd": r["p25_usd"], "median_usd": med, "median_wow_pct": wow, "index_floor": index_points(r["floor_usd"], base_floor), "index_p25": index_points(r["p25_usd"], base_p25), "index_median": index_points(med, base_med), "source_file": r["source_file"], } ) prev_med = med frontier_out = attach_wow_and_index(frontier_raw, "level_usd", "index_level") open_out = attach_wow_and_index(open_raw, "level_usd", "index_level") arb_out = attach_wow_and_index(arb_raw, "spread_pct", "index_level") return { "CN-TOKEN-SPOT": spot_out, "CN-FRONTIER": frontier_out, "CN-OPEN": open_out, "CN-ARB-SPREAD": arb_out, } def write_csv(path: Path, rows: list[dict], fields: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore") w.writeheader() for r in rows: w.writerow({k: "" if r.get(k) is None else r[k] for k in fields}) def _series_public(rows: list[dict], keys: list[str]) -> list[dict]: return [{k: r.get(k) for k in keys} for r in rows] def public_panel_labels(prices_dirs: list[Path]) -> list[str]: """Stable, non-absolute labels for published meta (no machine paths).""" labels: list[str] = [] cn_root = DEFAULT_COMPUTENAV.resolve() for p in prices_dirs: try: rel = p.resolve().relative_to(cn_root) labels.append(f"Web-ComputeNav/{rel.as_posix()}") except ValueError: labels.append(p.name or "external") return labels def write_json(path: Path, series_map: dict[str, list[dict]], prices_dirs: list[Path]) -> None: spot = series_map["CN-TOKEN-SPOT"] frontier = series_map["CN-FRONTIER"] open_b = series_map["CN-OPEN"] arb = series_map["CN-ARB-SPREAD"] fcfg = frontier_cfg() def latest_of(rows: list[dict]) -> dict | None: return rows[-1] if rows else None spot_l = latest_of(spot) fr_l = latest_of(frontier) op_l = latest_of(open_b) ar_l = latest_of(arb) fr_seats_ui: list[dict] = [] if fr_l is not None: smap, _ep = frontier_map_for_date(fcfg, fr_l["date"]) active_roles = set(str(fr_l.get("active_seats") or "").split(",")) - {""} for seat in fcfg["seats"]: sid = seat["id"] gid = smap.get(sid, "") fr_seats_ui.append( { "seat_id": sid, "label": seat["label"], "vendor": seat.get("vendor"), "compare_group_id": gid, "active": sid in active_roles, } ) payload = { "meta": { "generated_at": _utc_now(), "methodology_version": METHODOLOGY_VERSION, "license": "CC BY 4.0", "source": "GCPI Research (ComputeNav panels)", "base_date": BASE_DATE, "series_start_date": SERIES_START_DATE, "blended_weights": {"input": INPUT_W, "output": OUTPUT_W}, "prices_dirs": public_panel_labels(prices_dirs), "disclaimer": "Research indices. Not financial advice.", "constituents": { "CN-FRONTIER": { "basket_version": fcfg.get("basket_version"), "config": str(FRONTIER_CONFIG_PATH.relative_to(REPO)), "seats": fcfg["seats"], "mapping_epochs": fcfg["mapping_epochs"], }, "CN-OPEN": OPEN_GROUPS, }, }, "indices": { "CN-TOKEN-SPOT": { "id": "CN-TOKEN-SPOT", "label": "Token Spot Index", "status": "live", "headline_series": "median_usd", "unit_primary": "USD per 1M tokens (blended)", "frequency": "weekly", "methodology": "handbook/compute/CN-Token-Spot.md", "latest": None if spot_l is None else { "date": spot_l["date"], "floor_usd": spot_l["floor_usd"], "p25_usd": spot_l["p25_usd"], "median_usd": spot_l["median_usd"], "median_wow_pct": spot_l["median_wow_pct"], "n_quotes": spot_l["n_quotes"], "index_median": spot_l["index_median"], }, "series": _series_public( spot, [ "date", "n_quotes", "floor_usd", "p25_usd", "median_usd", "median_wow_pct", "index_floor", "index_p25", "index_median", ], ), }, "CN-FRONTIER": { "id": "CN-FRONTIER", "label": "Frontier Index", "status": "live", "headline_series": "level_usd", "unit_primary": "USD per 1M tokens (blended geo-mean)", "frequency": "weekly", "methodology": "handbook/compute/CN-Frontier-Basket.md", "basket_version": fcfg.get("basket_version"), "min_seats": int(fcfg.get("min_active_seats", 3)), "constituents": fr_seats_ui, "latest": None if fr_l is None else { "date": fr_l["date"], "level_usd": fr_l["level_usd"], "wow_pct": fr_l["wow_pct"], "n_seats": fr_l["n_seats"], "index_level": fr_l["index_level"], "active_seats": fr_l.get("active_seats"), "active_groups": fr_l.get("active_groups"), "seat_map": fr_l.get("seat_map"), "mapping_epoch": fr_l.get("mapping_epoch"), "basket_version": fr_l.get("basket_version") or fcfg.get("basket_version"), }, "series": _series_public( frontier, [ "date", "n_seats", "level_usd", "wow_pct", "index_level", "active_seats", "active_groups", "seat_map", "mapping_epoch", "basket_version", ], ), }, "CN-OPEN": { "id": "CN-OPEN", "label": "Open Index", "status": "live", "headline_series": "level_usd", "unit_primary": "USD per 1M tokens (blended geo-mean)", "frequency": "weekly", "methodology": "handbook/compute/CN-Open-Basket.md", "constituents": OPEN_GROUPS, "min_groups": OPEN_MIN_GROUPS, "latest": None if op_l is None else { "date": op_l["date"], "level_usd": op_l["level_usd"], "wow_pct": op_l["wow_pct"], "n_groups": op_l["n_groups"], "index_level": op_l["index_level"], "active_groups": op_l.get("active_groups"), }, "series": _series_public( open_b, [ "date", "n_groups", "level_usd", "wow_pct", "index_level", "active_groups", ], ), }, "CN-ARB-SPREAD": { "id": "CN-ARB-SPREAD", "label": "Arbitrage Index", "status": "live", "headline_series": "spread_pct", "unit_primary": "percent (median relative cross-host range)", "frequency": "weekly", "methodology": "handbook/compute/CN-Arbitrage-Spread.md", "latest": None if ar_l is None else { "date": ar_l["date"], "spread_pct": ar_l["spread_pct"], "spread_range_usd": ar_l["spread_range_usd"], "wow_pct": ar_l["wow_pct"], "n_groups": ar_l["n_groups"], "index_level": ar_l["index_level"], }, "series": _series_public( arb, [ "date", "n_groups", "spread_pct", "spread_range_usd", "wow_pct", "index_level", ], ), }, "CN-QA": { "id": "CN-QA", "label": "Quality-Adj Index", "status": "deferred", "headline_series": None, "unit_primary": "USD / capability unit", "frequency": "weekly", "methodology": "handbook/compute/CN-Quality-Adjusted.md", "latest": None, "series": [], "note": "Official levels deferred until a frozen capability source is joined to the panel.", }, }, } path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def main() -> int: ap = argparse.ArgumentParser(description="Build CN research indices from ComputeNav panels.") ap.add_argument("--prices-dir", action="append", type=Path, default=None) ap.add_argument("--json-out", type=Path, default=REPO / "api" / "v1" / "indices.json") ap.add_argument( "--mirror-json", type=Path, default=DEFAULT_COMPUTENAV / "api" / "v1" / "cn_indices.json", ) ap.add_argument("--no-mirror", action="store_true") ap.add_argument( "--out-dir", type=Path, default=REPO / "data" / "processed", help="Directory for per-index CSV outputs.", ) args = ap.parse_args() if args.prices_dir: prices_dirs = args.prices_dir else: root = DEFAULT_COMPUTENAV prices_dirs = [root / "data" / "processed", root / "data" / "history"] panels = discover_panels(prices_dirs) if not panels: print("ERROR: no prices_YYYY-MM-DD.csv found in:", prices_dirs, file=sys.stderr) return 1 series_map = build_all(panels) if not series_map["CN-TOKEN-SPOT"]: print("ERROR: no Token Spot rows.", file=sys.stderr) return 1 out_dir: Path = args.out_dir write_csv( out_dir / "cn_token_spot.csv", series_map["CN-TOKEN-SPOT"], [ "date", "n_quotes", "floor_usd", "p25_usd", "median_usd", "median_wow_pct", "index_floor", "index_p25", "index_median", "source_file", ], ) write_csv( out_dir / "cn_frontier.csv", series_map["CN-FRONTIER"], [ "date", "n_seats", "level_usd", "wow_pct", "index_level", "active_seats", "active_groups", "seat_map", "mapping_epoch", "basket_version", "source_file", ], ) write_csv( out_dir / "cn_open.csv", series_map["CN-OPEN"], [ "date", "n_groups", "level_usd", "wow_pct", "index_level", "active_groups", "source_file", ], ) write_csv( out_dir / "cn_arb_spread.csv", series_map["CN-ARB-SPREAD"], [ "date", "n_groups", "spread_pct", "spread_range_usd", "wow_pct", "index_level", "source_file", ], ) write_json(args.json_out, series_map, prices_dirs) if not args.no_mirror and args.mirror_json: write_json(args.mirror_json, series_map, prices_dirs) print(f"Wrote mirror {args.mirror_json}") for code, rows in series_map.items(): if not rows: print(f"{code}: no publishable weeks") continue latest = rows[-1] if code == "CN-TOKEN-SPOT": print( f"{code} weeks={len(rows)} latest={latest['date']} " f"median_usd={latest['median_usd']} n={latest['n_quotes']}" ) elif code == "CN-ARB-SPREAD": print( f"{code} weeks={len(rows)} latest={latest['date']} " f"spread_pct={latest['spread_pct']} n_groups={latest['n_groups']}" ) else: print( f"{code} weeks={len(rows)} latest={latest['date']} " f"level_usd={latest['level_usd']}" ) print(f"Wrote CSVs under {out_dir}") print(f"Wrote {args.json_out}") # Keep site/ self-contained for static hosting (api + data + handbook). scripts_dir = Path(__file__).resolve().parent if str(scripts_dir) not in sys.path: sys.path.insert(0, str(scripts_dir)) try: from sync_site_bundle import main as sync_site sync_site() except Exception as exc: # noqa: BLE001 — non-fatal for research rebuilds print(f"WARN: site bundle sync skipped ({exc})", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())