Usage and Cost Tracking skill for Claude Code

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

Reads ~/.claude/projects/*/*.jsonl (or a path you pass), de-duplicates API responses
(one response is logged as several lines that share message.id), and reports tokens by
day, project, model, and session, plus cache hit ratio, subagent share, and skill
attribution where the transcript records it.

Usage:
  python3 usage_report.py [--root ~/.claude/projects] [--days 7] [--project substr]
                          [--top 10] [--prices prices.json] [--json]

prices.json (optional, USD per million tokens; you supply current rates):
  {"claude-sonnet-5": {"input": 3, "output": 15, "cache_write": 3.75, "cache_read": 0.3}}
Model keys match by prefix, so "claude-sonnet" covers every Sonnet id.
Without --prices the report shows tokens only (no dollar figures are guessed).
Exit 0 on success, 1 if no usage records were found.
"""
import argparse
import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path

FIELDS = ("input", "output", "cache_write", "cache_read")


def iter_records(root: Path, since: datetime, project_filter: str):
    files = [root] if root.is_file() else sorted(root.glob("*/*.jsonl")) + sorted(root.glob("*/*/subagents/*.jsonl"))
    seen = set()
    for f in files:
        project = f.parent.name if root.is_dir() else f.parent.name
        if f.parent.name == "subagents":
            project = f.parent.parent.parent.name
        if project_filter and project_filter not in project:
            continue
        try:
            fh = open(f, encoding="utf-8", errors="replace")
        except OSError:
            continue
        with fh:
            for line in fh:
                if '"usage"' not in line:
                    continue
                try:
                    d = json.loads(line)
                except json.JSONDecodeError:
                    continue
                msg = d.get("message") or {}
                usage = msg.get("usage")
                if d.get("type") != "assistant" or not isinstance(usage, dict):
                    continue
                key = msg.get("id") or d.get("requestId") or d.get("uuid")
                if key in seen:
                    continue
                seen.add(key)
                ts = d.get("timestamp", "")
                try:
                    when = datetime.fromisoformat(ts.replace("Z", "+00:00"))
                except ValueError:
                    continue
                if when < since:
                    continue
                yield {
                    "day": when.date().isoformat(),
                    "project": project,
                    "session": d.get("sessionId", f.stem),
                    "model": msg.get("model", "unknown"),
                    "sidechain": bool(d.get("isSidechain")) or f.parent.name == "subagents",
                    "skill": d.get("attributionSkill"),
                    "input": usage.get("input_tokens", 0) or 0,
                    "output": usage.get("output_tokens", 0) or 0,
                    "cache_write": usage.get("cache_creation_input_tokens", 0) or 0,
                    "cache_read": usage.get("cache_read_input_tokens", 0) or 0,
                }


def cost(rec, prices):
    if not prices:
        return None
    for prefix, rate in sorted(prices.items(), key=lambda kv: -len(kv[0])):
        if rec["model"].startswith(prefix):
            return sum(rec[k] * rate.get(k, 0) for k in FIELDS) / 1_000_000
    return None


def fmt(n):
    for unit, size in (("B", 1e9), ("M", 1e6), ("k", 1e3)):
        if n >= size:
            return f"{n / size:.1f}{unit}"
    return str(int(n))


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--root", type=Path, default=Path.home() / ".claude" / "projects")
    ap.add_argument("--days", type=int, default=7)
    ap.add_argument("--project", default="")
    ap.add_argument("--top", type=int, default=10)
    ap.add_argument("--prices", type=Path)
    ap.add_argument("--json", action="store_true")
    a = ap.parse_args()
    prices = json.loads(a.prices.read_text()) if a.prices else None
    since = datetime.now(timezone.utc) - timedelta(days=a.days)

    groups = {k: defaultdict(lambda: defaultdict(float)) for k in ("day", "project", "model", "session", "skill")}
    total = defaultdict(float)
    count = 0
    for r in iter_records(a.root.expanduser(), since, a.project):
        count += 1
        c = cost(r, prices)
        r["tokens"] = sum(r[k] for k in FIELDS)
        for g in groups:
            key = r[g] if g != "skill" else (r["skill"] or "(none)")
            if g == "session":
                key = f"{r['project'][-30:]} {r['session'][:8]}"
            for k in FIELDS + ("tokens",):
                groups[g][key][k] += r[k]
            groups[g][key]["requests"] += 1
            if c is not None:
                groups[g][key]["usd"] += c
        for k in FIELDS + ("tokens",):
            total[k] += r[k]
            if r["sidechain"]:
                total["sidechain_tokens"] += r[k] if k == "tokens" else 0
        total["requests"] += 1
        if c is not None:
            total["usd"] += c

    if not count:
        print(f"no usage records in the last {a.days} day(s) under {a.root}", file=sys.stderr)
        return 1
    all_input = total["input"] + total["cache_write"] + total["cache_read"]
    summary = {
        "days": a.days,
        "requests": int(total["requests"]),
        "tokens": {k: int(total[k]) for k in FIELDS},
        "cache_hit_ratio": round(total["cache_read"] / all_input, 3) if all_input else None,
        "subagent_share": round(total["sidechain_tokens"] / total["tokens"], 3) if total["tokens"] else 0,
        "usd_estimate": round(total["usd"], 2) if prices else None,
        "groups": {g: {k: dict(v) for k, v in d.items()} for g, d in groups.items()},
    }
    if a.json:
        print(json.dumps(summary, indent=2))
        return 0

    print(f"Last {a.days} day(s): {summary['requests']} API responses")
    print("tokens  " + "  ".join(f"{k}={fmt(total[k])}" for k in FIELDS))
    print(f"cache hit ratio {summary['cache_hit_ratio']:.0%} | subagent share {summary['subagent_share']:.0%}"
          + (f" | est. ${summary['usd_estimate']:.2f} (your price file, list-price estimate)" if prices else ""))
    for g, title in (("day", "By day"), ("model", "By model"), ("project", "By project"),
                     ("session", f"Top {a.top} sessions"), ("skill", "By attributed skill")):
        rows = sorted(groups[g].items(), key=(lambda kv: kv[0]) if g == "day" else (lambda kv: -kv[1]["tokens"]))
        if g in ("session", "project", "skill"):
            rows = rows[: a.top]
        if g == "skill" and len(rows) == 1 and rows[0][0] == "(none)":
            continue
        print(f"\n{title}")
        for key, v in rows:
            usd = f"  ${v['usd']:.2f}" if prices else ""
            print(f"  {key[:44]:44} {fmt(v['tokens']):>8} tok  out {fmt(v['output']):>7}  "
                  f"req {int(v['requests']):>5}{usd}")
    return 0


if __name__ == "__main__":
    sys.exit(main())