Claude Code local token usage summarizer (privacy-safe: emits only token counts, no code/prompt content)

cc_local_usage.py

#!/usr/bin/env python3
"""Summarize local Claude Code token usage from on-disk session transcripts.

This reads ~/.claude/projects/**/*.jsonl (the same data behind `/usage`) and
emits ONLY aggregate token counts + estimated cost by day and model. It does
NOT read or emit any prompt, code, or message content -- safe to share.

Each of your Claude Code users runs this ONE self-contained file and sends you
the JSON:  python3 cc_local_usage.py --label alice --start 2026-06-01 --end 2026-07-01 --json > alice.json

Retention note: Claude Code prunes old transcripts (default ~30 days), so only
recent history is recoverable. Collect regularly for full coverage.
"""

import argparse
import glob
import json
import os
import sys
from collections import defaultdict
from decimal import Decimal

# Self-contained list prices (USD per 1M tokens): input, output. Confirm at
# https://platform.claude.com/docs/en/about-claude/pricing -- prices drift.
PRICES = {
    "claude-fable-5": (Decimal("10"), Decimal("50")),
    "claude-mythos-5": (Decimal("10"), Decimal("50")),
    "claude-opus-4-8": (Decimal("5"), Decimal("25")),
    "claude-opus-4-7": (Decimal("5"), Decimal("25")),
    "claude-opus-4-6": (Decimal("5"), Decimal("25")),
    "claude-opus-4-5": (Decimal("5"), Decimal("25")),
    # Sonnet 5 introductory pricing ($2/$10) runs through 2026-08-31; $3/$15 after (see SONNET5_* below).
    "claude-sonnet-5": (Decimal("2"), Decimal("10")),
    "claude-sonnet-4-6": (Decimal("3"), Decimal("15")),
    "claude-sonnet-4-5": (Decimal("3"), Decimal("15")),
    "claude-sonnet-4": (Decimal("3"), Decimal("15")),
    "claude-haiku-4-5": (Decimal("1"), Decimal("5")),
}
CACHE_READ_MULT = Decimal("0.1")
CACHE_WRITE_5M_MULT = Decimal("1.25")  # 5-minute cache write
CACHE_WRITE_1H_MULT = Decimal("2")     # 1-hour cache write
MTOK = Decimal("1000000")

# Sonnet 5 standard pricing takes effect 2026-09-01; introductory rates apply on/before 2026-08-31.
SONNET5_STANDARD = (Decimal("3"), Decimal("15"))
SONNET5_INTRO_END = "2026-08-31"


def price_key(model: str) -> str:
    # Map dated ids like claude-sonnet-4-5-20250929 -> claude-sonnet-4-5
    if not model:
        return ""
    for k in sorted(PRICES, key=len, reverse=True):
        if model.startswith(k):
            return k
    return ""


def rates_for(model, day=None):
    """(input, output) $/MTok for a model, or None if unknown. Handles Sonnet 5's
    dated introductory pricing (day is a 'YYYY-MM-DD' string)."""
    k = price_key(model)
    if not k:
        return None
    if k == "claude-sonnet-5" and day and day > SONNET5_INTRO_END:
        return SONNET5_STANDARD
    return PRICES[k]


def cost_usd(model, inp, out, cr, cc5, cc1h, day=None) -> Decimal:
    rates = rates_for(model, day)
    if rates is None:
        return Decimal("0")
    ip, op = rates
    return (
        Decimal(inp) * ip
        + Decimal(out) * op
        + Decimal(cr) * ip * CACHE_READ_MULT
        + Decimal(cc5) * ip * CACHE_WRITE_5M_MULT
        + Decimal(cc1h) * ip * CACHE_WRITE_1H_MULT
    ) / MTOK


def main():
    ap = argparse.ArgumentParser(description="Local Claude Code token usage summary")
    ap.add_argument("--dir", default=os.path.expanduser("~/.claude/projects"))
    ap.add_argument("--label", default=os.environ.get("USER", "unknown"),
                    help="identity for this machine/user (e.g. your email)")
    ap.add_argument("--start", default=None, help="YYYY-MM-DD inclusive")
    ap.add_argument("--end", default=None, help="YYYY-MM-DD exclusive")
    ap.add_argument("--json", action="store_true", help="emit JSON instead of a table")
    args = ap.parse_args()

    files = glob.glob(os.path.join(args.dir, "**", "*.jsonl"), recursive=True)
    by_model = defaultdict(lambda: {"input": 0, "output": 0, "cache_read": 0, "cache_creation": 0, "cost": Decimal("0")})
    by_day = defaultdict(lambda: Decimal("0"))
    dates_seen = []
    unknown_models = set()
    # Claude Code writes one JSONL line per content block of an assistant turn
    # (thinking, text, each tool_use), and every one of those lines repeats the
    # same message/request ID and a copy of the whole response's usage. So we
    # count each response once, keyed by (message.id, requestId). Among the
    # repeated lines we keep the one with the largest usage: while a response is
    # still streaming, its early lines can report a partial output_tokens, and
    # only the final line carries the true total.
    best = {}
    total = {"input": 0, "output": 0, "cache_read": 0, "cache_creation": 0, "cost": Decimal("0"), "messages": 0}

    for path in files:
        try:
            with open(path, "r", errors="replace") as fh:
                for line in fh:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        o = json.loads(line)
                    except Exception:
                        continue
                    msg = o.get("message", {}) or {}
                    usage = msg.get("usage") or o.get("usage")
                    if not usage:
                        continue
                    ts = o.get("timestamp") or msg.get("timestamp") or ""
                    day = ts[:10]
                    if args.start and day and day < args.start:
                        continue
                    if args.end and day and day >= args.end:
                        continue
                    model = msg.get("model") or o.get("model") or "unknown"
                    if model in ("<synthetic>", "unknown"):
                        continue
                    inp = int(usage.get("input_tokens", 0) or 0)
                    out = int(usage.get("output_tokens", 0) or 0)
                    cr = int(usage.get("cache_read_input_tokens", 0) or 0)
                    cc = int(usage.get("cache_creation_input_tokens", 0) or 0)
                    ccd = usage.get("cache_creation")
                    if isinstance(ccd, dict):
                        cc5 = int(ccd.get("ephemeral_5m_input_tokens", 0) or 0)
                        cc1h = int(ccd.get("ephemeral_1h_input_tokens", 0) or 0)
                    else:
                        cc5, cc1h = cc, 0   # fallback: price all cache-creation as 5-minute
                    if not (inp or out or cr or cc5 or cc1h):
                        continue
                    key = (msg.get("id"), o.get("requestId"))
                    usage_total = inp + out + cr + cc5 + cc1h
                    prev = best.get(key)
                    if prev is None or usage_total > prev[0]:
                        best[key] = (usage_total, model, day, inp, out, cr, cc5, cc1h)
        except (OSError, IOError):
            continue

    # Aggregate the deduped records (one per response) into per-model/day/total tallies.
    for _usage_total, model, day, inp, out, cr, cc5, cc1h in best.values():
        if rates_for(model, day) is None:
            unknown_models.add(model)
        c = cost_usd(model, inp, out, cr, cc5, cc1h, day)
        m = by_model[model]
        m["input"] += inp; m["output"] += out; m["cache_read"] += cr
        m["cache_creation"] += cc5 + cc1h; m["cost"] += c
        by_day[day] += c
        total["input"] += inp; total["output"] += out
        total["cache_read"] += cr; total["cache_creation"] += cc5 + cc1h
        total["cost"] += c; total["messages"] += 1
        if day:
            dates_seen.append(day)

    date_min = min(dates_seen) if dates_seen else None
    date_max = max(dates_seen) if dates_seen else None

    if args.json:
        out = {
            "label": args.label,
            "window": {"start": args.start, "end": args.end},
            "coverage": {"earliest": date_min, "latest": date_max},
            "unknown_models": sorted(unknown_models),
            "total": {**{k: (str(v) if isinstance(v, Decimal) else v) for k, v in total.items()}},
            "by_model": {k: {kk: (str(vv) if isinstance(vv, Decimal) else vv) for kk, vv in v.items()} for k, v in by_model.items()},
            "by_day_cost_usd": {k: str(v) for k, v in sorted(by_day.items())},
        }
        print(json.dumps(out, indent=2))
        return

    print(f"label: {args.label}   transcripts scanned: {len(files)}")
    print(f"coverage on disk: {date_min} -> {date_max}"
          + (f"   (window {args.start}..{args.end})" if args.start or args.end else ""))
    print(f"messages with usage: {total['messages']:,}")
    print("-" * 66)
    print(f"{'model':30} {'in':>12} {'out':>10} {'est $':>10}")
    for model, v in sorted(by_model.items(), key=lambda kv: -kv[1]["cost"]):
        print(f"{model:30} {v['input']:>12,} {v['output']:>10,} ${v['cost']:>9,.2f}")
    print("-" * 66)
    print(f"{'TOTAL est. cost':30} {'':>12} {'':>10} ${total['cost']:>9,.2f}")
    print(f"(cache read {total['cache_read']:,} + cache write {total['cache_creation']:,} tokens)")
    if unknown_models:
        print("WARNING: no price on file for these models, counted as $0 -> update PRICES: "
              + ", ".join(sorted(unknown_models)))


if __name__ == "__main__":
    main()