Claude Code Usage

usage.py

#!/usr/bin/env python3
"""Analyze Claude Code token usage and costs from local JSONL logs.

Reads ~/.claude/projects/*/*.jsonl, extracts usage from assistant messages,
and reports token counts and costs by time period. Rates are defined in
COST_LINES (Opus 4.6 pricing).
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from bisect import bisect_left
from collections import defaultdict, deque
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import NamedTuple

COST_LINES = [
    ("Uncached input", "uncached", 5.00),
    ("Cache write 5m", "cw5m", 6.25),
    ("Cache write 1h", "cw1h", 10.00),
    ("Cache read", "cr", 0.50),
    ("Output", "output", 25.00),
]

RATES = {key: rate for _, key, rate in COST_LINES}


class Record(NamedTuple):
    date: date
    session_id: str
    uncached: int
    cw5m: int
    cw1h: int
    cr: int
    output: int


def parse_records(base: Path) -> tuple[list[Record], int]:
    """Parse JSONL files and return (sorted records, file count)."""
    records: list[Record] = []
    file_count = 0
    for fpath in base.glob("*/*.jsonl"):
        file_count += 1
        with open(fpath, "r") as f:
            for line in f:
                try:
                    d = json.loads(line)
                except (json.JSONDecodeError, ValueError):
                    continue
                if d.get("type") != "assistant":
                    continue
                usage = d.get("message", {}).get("usage")
                ts = d.get("timestamp")
                if not usage or not ts:
                    continue

                dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
                session_id = d.get(
                    "sessionId", fpath.stem
                )

                uncached = usage.get("input_tokens", 0)
                output = usage.get("output_tokens", 0)
                cr = usage.get(
                    "cache_read_input_tokens", 0
                )

                cache_total = usage.get(
                    "cache_creation_input_tokens", 0
                )
                cc = usage.get("cache_creation", {})
                cw5m = cc.get("ephemeral_5m_input_tokens", 0)
                cw1h = cc.get("ephemeral_1h_input_tokens", 0)
                if cw5m + cw1h < cache_total:
                    cw1h = cache_total - cw5m

                records.append(Record(
                    dt.date(), session_id, uncached,
                    cw5m, cw1h, cr, output,
                ))
    records.sort()
    return records, file_count


def cost(tokens: int, rate: float) -> float:
    return tokens * rate / 1_000_000


def analyze(recs: list[Record], label: str) -> dict | None:
    if not recs:
        return None

    dates: set[date] = set()
    sessions: set[str] = set()
    totals = defaultdict(int)

    for r in recs:
        dates.add(r.date)
        sessions.add(r.session_id)
        totals["uncached"] += r.uncached
        totals["cw5m"] += r.cw5m
        totals["cw1h"] += r.cw1h
        totals["cr"] += r.cr
        totals["output"] += r.output

    calls = len(recs)
    tot_input = (
        totals["uncached"] + totals["cw5m"]
        + totals["cw1h"] + totals["cr"]
    )
    hit_rate = (
        (totals["cr"] / tot_input * 100) if tot_input else 0.0
    )

    costs = {
        key: cost(totals[key], rate)
        for _, key, rate in COST_LINES
    }
    cost_total = sum(costs.values())

    cost_no_cache = (
        cost(tot_input, RATES["uncached"])
        + cost(totals["output"], RATES["output"])
    )
    savings = cost_no_cache - cost_total

    return {
        "label": label,
        "days": len(dates),
        "sessions": len(sessions),
        "calls": calls,
        **{key: totals[key] for _, key, _ in COST_LINES},
        "total_input": tot_input,
        "cache_hit": hit_rate,
        "io_ratio": (
            (tot_input / totals["output"])
            if totals["output"] else 0.0
        ),
        "avg_in": tot_input / calls,
        "avg_out": totals["output"] / calls,
        **{f"cost_{key}": v for key, v in costs.items()},
        "cost_total": cost_total,
        "cost_no_cache": cost_no_cache,
        "savings": savings,
        "savings_pct": (
            (savings / cost_no_cache * 100)
            if cost_no_cache else 0.0
        ),
    }


def _add_day(totals: dict[str, int], recs: list[Record]) -> None:
    for r in recs:
        totals["uncached"] += r.uncached
        totals["cw5m"] += r.cw5m
        totals["cw1h"] += r.cw1h
        totals["cr"] += r.cr
        totals["output"] += r.output
        totals["calls"] += 1


def _sub_day(totals: dict[str, int], recs: list[Record]) -> None:
    for r in recs:
        totals["uncached"] -= r.uncached
        totals["cw5m"] -= r.cw5m
        totals["cw1h"] -= r.cw1h
        totals["cr"] -= r.cr
        totals["output"] -= r.output
        totals["calls"] -= 1


def _window_cost(totals: dict[str, int]) -> float:
    return sum(
        cost(totals[key], rate) for _, key, rate in COST_LINES
    )


def peak_7d_window(records: list[Record]) -> dict | None:
    if not records:
        return None

    by_day: dict[date, list[Record]] = defaultdict(list)
    for r in records:
        by_day[r.date].append(r)

    days = sorted(by_day)
    if not days:
        return None

    totals: dict[str, int] = defaultdict(int)
    best_cost = 0.0
    best_range: tuple[date, date] | None = None
    day_queue: deque[date] = deque()

    for day in days:
        _add_day(totals, by_day[day])
        day_queue.append(day)
        while day_queue[0] < day - timedelta(days=6):
            _sub_day(totals, by_day[day_queue.popleft()])
        c = _window_cost(totals)
        if c > best_cost:
            best_cost = c
            best_range = (day_queue[0], day)

    if not best_range:
        return None

    start, end = best_range
    peak_recs = [
        r for r in records if start <= r.date <= end
    ]
    return analyze(
        peak_recs, f"Peak 7d ({start} to {end})"
    )


def fmt_tok(n: int | float) -> str:
    n = int(n)
    if n >= 1_000_000_000:
        return f"{n / 1_000_000_000:.2f}B"
    if n >= 1_000_000:
        return f"{n / 1_000_000:.2f}M"
    if n >= 1_000:
        return f"{n / 1_000:.1f}K"
    return str(n)


def fmt_cost(c: float) -> str:
    if c >= 1000:
        return f"${c:,.0f}"
    if c >= 100:
        return f"${c:.0f}"
    if c >= 1:
        return f"${c:.2f}"
    return f"${c:.3f}"


def print_section(s: dict) -> None:
    print(f"\n{'=' * 60}")
    print(f"  {s['label']}")
    print(f"{'=' * 60}")
    print(
        f"  Active days:  {s['days']}    "
        f"Sessions: {s['sessions']}    "
        f"API calls: {s['calls']:,}"
    )
    print()
    print("  INPUT TOKENS")
    print(f"    Total input:       {fmt_tok(s['total_input']):>10}")
    print(f"    \u251c\u2500 Uncached:       {fmt_tok(s['uncached']):>10}")
    print(f"    \u251c\u2500 Cache write 5m: {fmt_tok(s['cw5m']):>10}")
    print(f"    \u251c\u2500 Cache write 1h: {fmt_tok(s['cw1h']):>10}")
    print(
        f"    \u2514\u2500 Cache read:     {fmt_tok(s['cr']):>10}"
        f"  ({s['cache_hit']:.1f}% hit rate)"
    )
    print()
    print("  OUTPUT TOKENS")
    print(f"    Total output:      {fmt_tok(s['output']):>10}")
    print()
    print("  RATIOS")
    print(f"    I:O ratio:         {s['io_ratio']:.1f}:1")
    print(f"    Avg input/call:    {fmt_tok(int(s['avg_in'])):>10}")
    print(f"    Avg output/call:   {fmt_tok(int(s['avg_out'])):>10}")
    print()
    print("  COST BREAKDOWN")
    for label, key, rate in COST_LINES:
        print(
            f"    {label + ':':19}"
            f"{fmt_cost(s[f'cost_{key}']):>10}"
            f"  (${rate:.2f}/MTok)"
        )
    print("    " + "\u2500" * 31)
    print(f"    Total (cached):    {fmt_cost(s['cost_total']):>10}")
    print(
        f"    Total (no cache):  "
        f"{fmt_cost(s['cost_no_cache']):>10}"
    )
    print(
        f"    Savings:           {fmt_cost(s['savings']):>10}"
        f"  ({s['savings_pct']:.1f}%)"
    )


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Analyze Claude Code token usage and costs.",
    )
    parser.add_argument(
        "--path",
        default=os.path.expanduser("~/.claude/projects"),
        help=(
            "Path to projects directory "
            "(default: ~/.claude/projects)"
        ),
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output as JSON instead of formatted table",
    )
    args = parser.parse_args()

    base = Path(args.path)
    if not base.exists():
        print(f"Error: {base} does not exist", file=sys.stderr)
        sys.exit(1)

    records, file_count = parse_records(base)
    if not records:
        print("No usage records found.", file=sys.stderr)
        sys.exit(1)

    today = datetime.now(timezone.utc).date()

    cutoff_30 = bisect_left(
        records, today - timedelta(days=30), key=lambda r: r.date
    )
    cutoff_7 = bisect_left(
        records, today - timedelta(days=7), key=lambda r: r.date
    )

    all_time = analyze(records, "All-Time")
    last_30d = analyze(records[cutoff_30:], "Last 30d")
    last_7d = analyze(records[cutoff_7:], "Last 7d")
    peak = peak_7d_window(records)

    periods = [
        p for p in [all_time, last_30d, last_7d, peak] if p
    ]

    if args.json:
        print(json.dumps(periods, indent=2, default=str))
        return

    print()
    print("  Claude Code Token Usage Analysis")
    print("  ================================")
    print(f"  Data range: {records[0].date} to {records[-1].date}")
    print(f"  Files scanned: {file_count}")
    print(f"  Total records: {len(records):,}")

    for p in periods:
        print_section(p)
    print()


if __name__ == "__main__":
    main()