"""ccusage-cost-dashboard-explained.py - re-run the pricing and grouping rules we read in ccusage at a pinned commit. What it does Downloads (once, into --cache) four inputs pinned by commit: - ccusage/ccusage @ 43af93d21c8f: the two Claude Code test transcripts under apps/ccusage/test/fixtures/claude/ and the embedded models.dev price snapshot rust/crates/ccusage-core/src/models-dev-pricing.json; - BerriAI/litellm @ ac98cabae5ff: model_prices_and_context_window.json (the file ccusage fetches from main). Then re-implements, in plain Python, four rules from that commit and runs them: 1. cost modes (crates/ccusage-core/src/cost.rs): display = stored costUSD or 0; auto = stored costUSD when present, else priced from tokens; calculate = always priced from tokens; 2. the token formula: input, output, 5-minute cache writes at the cache-write rate, 1-hour cache writes at 2x the input rate, cache reads; and the two long-context semantics in the same file (a models.dev tier re-prices the WHOLE request once its context passes the tier size; LiteLLM *_above_200k rates apply only to the tokens of one bucket beyond 200,000); 3. de-duplication (adapters/claude/src/lib.rs, usage_dedupe_hash): key = message id + request id; with no request id, add the session id, and add the timestamp unless the message id starts with "msg_"; when two lines collide the one with more tokens is kept (sidechain replays are not modelled here); 4. 5-hour blocks (crates/ccusage/src/blocks.rs): a block starts at the first entry's timestamp floored to the hour and closes when an entry is more than 5 h after the block start or after the previous entry. This is a reading aid, not ccusage: it prices from the models.dev snapshot alone (ccusage layers LiteLLM, the snapshot and built-in schedules), it ignores the fast-mode multiplier (one fixture line is marked fast and is priced at 1x here), and the duplicate, tier and block cases are constructed inputs, labelled so. Inputs none required; --cache DIR (default ./ccusage_inputs), --csv PATH (default ./ccusage-cost-dashboard-explained.csv) Run python ccusage-cost-dashboard-explained.py Needs Python 3.9+, standard library only. Network on first run (raw.githubusercontent.com, which serves no robots.txt rules); requests send a descriptive User-Agent and are spaced one second apart. """ from __future__ import annotations import argparse import csv import datetime as dt import json import os import sys import time import urllib.request UA = "VibeCodeTextbook research script (contact@vibecodetextbook.com)" CCUSAGE_SHA = "43af93d21c8f3b61b57e0c961f68b491aa2e3add" LITELLM_SHA = "ac98cabae5ffd386ee11d41487f915fc9fd27848" RAW = "https://raw.githubusercontent.com" INPUTS = { "alpha.jsonl": f"{RAW}/ccusage/ccusage/{CCUSAGE_SHA}/apps/ccusage/test/fixtures/claude/projects/project-alpha/session-alpha/chat.jsonl", "beta.jsonl": f"{RAW}/ccusage/ccusage/{CCUSAGE_SHA}/apps/ccusage/test/fixtures/claude/projects/project-beta/session-beta/chat.jsonl", "models-dev-pricing.json": f"{RAW}/ccusage/ccusage/{CCUSAGE_SHA}/rust/crates/ccusage-core/src/models-dev-pricing.json", "litellm-prices.json": f"{RAW}/BerriAI/litellm/{LITELLM_SHA}/model_prices_and_context_window.json", } CACHE_1H_MULT = 2.0 # cost.rs: CACHE_CREATE_1H_INPUT_MULTIPLIER LITELLM_THRESHOLD = 200_000 # pricing.rs: DEFAULT_LONG_CONTEXT_THRESHOLD_TOKENS (marginal, per bucket) BLOCK_HOURS = 5 def fetch_inputs(cache: str) -> dict[str, str]: os.makedirs(cache, exist_ok=True) paths = {} for name, url in INPUTS.items(): dst = os.path.join(cache, name) if not os.path.exists(dst): req = urllib.request.Request(url, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=60) as r: body = r.read() with open(dst, "wb") as f: f.write(body) time.sleep(1) paths[name] = dst return paths # ---------- prices ---------- def strip_date_suffix(model: str) -> str: head, _, tail = model.rpartition("-") return head if len(tail) == 8 and tail.isdigit() else model def snapshot_rates(snapshot: dict, model: str) -> dict: """Per-token rates from the models.dev snapshot (prices there are $ per million tokens).""" entry = snapshot.get(model) or snapshot[strip_date_suffix(model)] cost = entry["cost"] inp = cost["input"] / 1e6 rates = { "input": inp, "output": cost["output"] / 1e6, "cache_create": cost.get("cache_write", cost["input"] * 1.25) / 1e6, "cache_read": cost.get("cache_read", cost["input"] * 0.1) / 1e6, "tier": None, } for t in cost.get("tiers") or []: if t.get("tier", {}).get("type") == "context": rates["tier"] = { "size": t["tier"]["size"], "input": t["input"] / 1e6, "output": t["output"] / 1e6, "cache_create": t.get("cache_write", t["input"] * 1.25) / 1e6, "cache_read": t.get("cache_read", t["input"] * 0.1) / 1e6, } return rates def price_whole_request(u: dict, r: dict) -> float: """models.dev tier semantics: the request's whole context picks the tier for every bucket.""" c5, c1 = u.get("cache_5m", u.get("cache_creation_input_tokens", 0)), u.get("cache_1h", 0) ctx = u.get("input_tokens", 0) + u.get("cache_read_input_tokens", 0) + c5 + c1 p = r["tier"] if (r["tier"] and ctx > r["tier"]["size"]) else r return (u.get("input_tokens", 0) * p["input"] + u.get("output_tokens", 0) * p["output"] + c5 * p["cache_create"] + c1 * p["input"] * CACHE_1H_MULT + u.get("cache_read_input_tokens", 0) * p["cache_read"]) def tiered(tokens: int, base: float, above: float | None) -> float: if above is not None and tokens > LITELLM_THRESHOLD: return LITELLM_THRESHOLD * base + (tokens - LITELLM_THRESHOLD) * above return tokens * base def price_marginal(u: dict, r: dict) -> float: """LiteLLM *_above_200k semantics: each bucket pays the higher rate only past 200,000 of its own tokens.""" t = r["tier"] or {} c5, c1 = u.get("cache_5m", u.get("cache_creation_input_tokens", 0)), u.get("cache_1h", 0) return (tiered(u.get("input_tokens", 0), r["input"], t.get("input")) + tiered(u.get("output_tokens", 0), r["output"], t.get("output")) + tiered(c5, r["cache_create"], t.get("cache_create")) + tiered(c1, r["input"] * CACHE_1H_MULT, t.get("input") and t["input"] * CACHE_1H_MULT) + tiered(u.get("cache_read_input_tokens", 0), r["cache_read"], t.get("cache_read"))) def entry_cost(e: dict, mode: str, snapshot: dict) -> float: stored = e.get("costUSD") if mode == "display": return stored or 0.0 if mode == "auto" and stored is not None: return stored return price_whole_request(e["message"]["usage"], snapshot_rates(snapshot, e["message"]["model"])) # ---------- de-duplication ---------- def dedupe_key(e: dict) -> tuple: mid, rid = e["message"]["id"], e.get("requestId") if rid is not None: return (mid, rid) if mid.startswith("msg_"): return (mid, None, e["sessionId"]) return (mid, None, e["sessionId"], e["timestamp"]) def tokens(e: dict) -> int: u = e["message"]["usage"] return sum(u.get(k, 0) for k in ("input_tokens", "output_tokens", "cache_creation_input_tokens", "cache_read_input_tokens")) def dedupe(entries: list[dict]) -> list[dict]: kept: dict[tuple, dict] = {} for e in entries: k = dedupe_key(e) if k not in kept or tokens(e) > tokens(kept[k]): kept[k] = e return list(kept.values()) # ---------- blocks ---------- def ts(s: str) -> dt.datetime: return dt.datetime.fromisoformat(s.replace("Z", "+00:00")) def blocks(stamps: list[str], floor: bool = True) -> list[tuple[str, list[str]]]: span = dt.timedelta(hours=BLOCK_HOURS) out, start, cur = [], None, [] for s in sorted(stamps, key=ts): t = ts(s) if start is not None and (t - start > span or t - ts(cur[-1]) > span): out.append((start.strftime("%H:%M"), cur)) start, cur = None, [] if start is None: start = t.replace(minute=0, second=0, microsecond=0) if floor else t cur.append(s) out.append((start.strftime("%H:%M"), cur)) return out # ---------- constructed inputs (labelled; not taken from any real log) ---------- def line(ts_, session, mid, rid, out_tokens, inp=3, cache_read=20_000): e = {"timestamp": ts_, "sessionId": session, "message": {"id": mid, "model": "claude-sonnet-5-5", "usage": {"input_tokens": inp, "output_tokens": out_tokens, "cache_creation_input_tokens": 0, "cache_read_input_tokens": cache_read}}} if rid: e["requestId"] = rid return e def constructed_duplicates() -> list[dict]: return [ # one response written as three lines (one per content block), the last carrying the final usage line("2026-10-12T09:00:01.000Z", "s1", "msg_A", "req_A", 12), line("2026-10-12T09:00:02.000Z", "s1", "msg_A", "req_A", 12), line("2026-10-12T09:00:03.000Z", "s1", "msg_A", "req_A", 480), # the same response copied into a second session's transcript line("2026-10-12T09:00:03.000Z", "s2", "msg_A", "req_A", 480), # a distinct response line("2026-10-12T09:01:00.000Z", "s1", "msg_B", "req_B", 300), # a gateway that reuses one non-Anthropic id and sends no request id: distinct timestamps stay separate line("2026-10-12T09:02:00.000Z", "s1", "gw-1", None, 200), line("2026-10-12T09:03:00.000Z", "s1", "gw-1", None, 250), ] def main() -> int: sys.stdout.reconfigure(encoding="utf-8") ap = argparse.ArgumentParser() ap.add_argument("--cache", default="ccusage_inputs") ap.add_argument("--csv", default="ccusage-cost-dashboard-explained.csv") a = ap.parse_args() p = fetch_inputs(a.cache) snapshot = json.load(open(p["models-dev-pricing.json"], encoding="utf-8")) litellm = json.load(open(p["litellm-prices.json"], encoding="utf-8")) rows = [] def add(section, item, value, unit, note=""): rows.append({"section": section, "item": item, "value": value, "unit": unit, "note": note}) print(f"{section:<10} {item:<46} {value!s:>14} {unit} {note}") # 0. the price sources at the pinned commits add("sources", "models.dev snapshot entries", len(snapshot), "models") add("sources", "models.dev snapshot claude-* entries", sum(k.startswith("claude") for k in snapshot), "models") add("sources", "LiteLLM file entries", len(litellm), "models") for m in ("claude-opus-5-5", "claude-opus-5", "claude-opus-4-8", "claude-sonnet-5-5"): fast = (litellm.get(m, {}).get("provider_specific_entry") or {}).get("fast") add("sources", f"LiteLLM fast multiplier {m}", fast if fast is not None else "none", "x") # 1. the repo's own fixtures under the three modes fx = [json.loads(x) for n in ("alpha.jsonl", "beta.jsonl") for x in open(p[n], encoding="utf-8") if x.strip()] add("fixtures", "entries", len(fx), "lines") add("fixtures", "tokens, all buckets", sum(tokens(e) for e in fx), "tokens") for mode in ("display", "auto", "calculate"): add("fixtures", f"total cost, {mode} mode", round(sum(entry_cost(e, mode, snapshot) for e in fx), 6), "USD") # 2. duplicate lines (constructed) dup = constructed_duplicates() kept = dedupe(dup) r55 = snapshot_rates(snapshot, "claude-sonnet-5-5") naive = sum(price_whole_request(e["message"]["usage"], r55) for e in dup) clean = sum(price_whole_request(e["message"]["usage"], r55) for e in kept) add("dedupe", "lines in", len(dup), "lines", "constructed") add("dedupe", "entries kept", len(kept), "entries", "constructed") add("dedupe", "output tokens, naive sum", sum(e["message"]["usage"]["output_tokens"] for e in dup), "tokens") add("dedupe", "output tokens, after dedupe", sum(e["message"]["usage"]["output_tokens"] for e in kept), "tokens") add("dedupe", "cost, naive sum", round(naive, 6), "USD", "claude-sonnet-5-5 snapshot rates") add("dedupe", "cost, after dedupe", round(clean, 6), "USD") add("dedupe", "naive overcount", round(100 * (naive / clean - 1), 1), "percent") # 3. one long request under the two long-context semantics (constructed request) req = {"input_tokens": 60_000, "output_tokens": 2_000, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 150_000} r = snapshot_rates(snapshot, "claude-sonnet-4") add("tiers", "claude-sonnet-4 tier size", r["tier"]["size"], "tokens") add("tiers", "claude-sonnet-4 whole-request price", round(price_whole_request(req, r), 6), "USD", "60k input + 150k cache read + 2k output") add("tiers", "claude-sonnet-4 per-bucket marginal price", round(price_marginal(req, r), 6), "USD", "same rates; no bucket passes 200k") r = snapshot_rates(snapshot, "claude-haiku-5-5") add("tiers", "claude-haiku-5-5 tier size", r["tier"]["size"], "tokens") add("tiers", "claude-haiku-5-5 whole-request price", round(price_whole_request(req, r), 6), "USD", "same request") add("tiers", "claude-haiku-5-5 price at base rates", round(price_whole_request(req, dict(r, tier=None)), 6), "USD", "what ignoring the tier would give") # 4. 1-hour vs 5-minute cache writes for 100,000 tokens for m in ("claude-opus-5-5", "claude-sonnet-5-5"): r = snapshot_rates(snapshot, m) add("cache", f"{m} 5m write of 100k", round(100_000 * r["cache_create"], 6), "USD") add("cache", f"{m} 1h write of 100k (2x input)", round(100_000 * r["input"] * CACHE_1H_MULT, 6), "USD") lit = litellm[m].get("cache_creation_input_token_cost_above_1hr") add("cache", f"{m} LiteLLM 1h field x 100k", round(100_000 * lit, 6), "USD", "field ccusage does not read") # 5. blocks: floored start vs first-message start (constructed timeline) day = ["2026-10-12T10:47:00Z", "2026-10-12T13:30:00Z", "2026-10-12T15:20:00Z", "2026-10-12T21:00:00Z"] for floor in (True, False): bl = blocks(day, floor) label = "floored start" if floor else "first-message start" add("blocks", f"{label}: blocks", len(bl), "blocks") add("blocks", f"{label}: block holding 15:20", next(s for s, xs in bl if "2026-10-12T15:20:00Z" in xs), "UTC") with open(a.csv, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=["section", "item", "value", "unit", "note"]) w.writeheader() w.writerows(rows) print(f"wrote {len(rows)} rows to {a.csv}") return 0 if __name__ == "__main__": raise SystemExit(main())