#!/usr/bin/env python3 # MIT License # # Copyright (c) 2026 Tony Perez # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. """token_audit: see where your Claude Code tokens go. Reads the session transcripts Claude Code keeps on your machine and prints a summary of what drove your token use. Standard library only. What it does NOT do: * It makes no network calls. * It writes no files. (--json prints to stdout.) * It never prints transcript content, file names, project names or commands. Only totals and percentages come out, so the report is safe to read and reasonable to share. Your transcripts themselves can hold secrets; share the report, not them. Costs are at public list price. On a subscription they are a way to compare categories, not what you paid. Check PRICES_AS_OF before relying on them. """ import argparse import collections import datetime as dt import hashlib import json import os import re import statistics import sys VERSION = "0.1.0" SCHEMA = 1 # USD per million tokens: input, 5m cache write, 1h cache write, cache read, output. # From https://platform.claude.com/docs/en/about-claude/pricing PRICES_AS_OF = "2026-11-03" PRICES = { "claude-fable-5-1": (10, 12.5, 20, 0.25, 50), "claude-mythos-5-1": (10, 12.5, 20, 0.25, 50), "claude-fable-5": (10, 12.5, 20, 1, 50), "claude-mythos-5": (10, 12.5, 20, 1, 50), "claude-opus-5-5": (4, 5, 8, 0.20, 20), "claude-opus-5": (5, 6.25, 10, 0.50, 25), "claude-opus-4-8": (5, 6.25, 10, 0.50, 25), "claude-opus-4-7": (5, 6.25, 10, 0.50, 25), "claude-opus-4-6": (5, 6.25, 10, 0.50, 25), "claude-opus-4-5": (5, 6.25, 10, 0.50, 25), "claude-opus-4-1": (15, 18.75, 30, 1.50, 75), "claude-opus-4": (15, 18.75, 30, 1.50, 75), "claude-sonnet-5-5": (2, 2.5, 4, 0.10, 10), "claude-sonnet-5": (2, 2.5, 4, 0.20, 10), "claude-sonnet-4-6": (3, 3.75, 6, 0.30, 15), "claude-sonnet-4-5": (3, 3.75, 6, 0.30, 15), "claude-sonnet-4": (3, 3.75, 6, 0.30, 15), "claude-haiku-4-5": (1, 1.25, 2, 0.10, 5), "claude-haiku-3-5": (0.80, 1, 1.60, 0.08, 4), } # Haiku 5.5 is priced by prompt length: every category changes above 100,000 tokens. HAIKU_5_5 = "claude-haiku-5-5" HAIKU_5_5_SMALL = (0.10, 0.125, 0.20, 0.01, 0.50) HAIKU_5_5_LARGE = (0.50, 0.625, 1, 0.05, 2.50) HAIKU_5_5_LIMIT = 100_000 FAST_MULTIPLIER = 2.0 # fast mode is double the base rates US_GEO_MULTIPLIER = 1.1 # inference_geo "us" MAX_LINE_CHARS = 30_000_000 MAX_MODEL_CHARS = 80 MAX_ID_CHARS = 200 IMAGE_TOKEN_ESTIMATE = 1500 CHARS_PER_TOKEN = 4 REWRITE_SHARE = 0.3 # a call rewrote the cache if over 30% of its prompt was written... REWRITE_MIN_CONTEXT = 20_000 # ...and the prompt was at least this big LARGE_CONTEXT = 250_000 COMPACT_CAPS = (200_000, 300_000, 400_000) TOP_TRANSCRIPTS = 10 CONTEXT_BUCKETS = ((0, 100_000, "under 100k"), (100_000, 250_000, "100k to 250k"), (250_000, 500_000, "250k to 500k"), (500_000, float("inf"), "over 500k")) IMAGE_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".pdf"} DOC_EXT = {".md", ".txt", ".rst", ".json", ".yml", ".yaml", ".csv", ".log", ".xml", ".html"} REMOTE_RE = re.compile(r"\bssh\b|Invoke-Remote|\bplink\b|\bscp\b") LOCAL_READ_RE = re.compile( r"\b(grep|rg|find|cat|head|tail|sed -n|awk|ls|wc|git (show|log|diff|status)|svn (cat|log|diff|status|info)" r"|Get-Content|Select-String|Get-ChildItem)\b") ITEM_LABELS = ( ("searches", "Searches (Grep, Glob, shell)"), ("read_code", "Read: code"), ("read_docs", "Read: docs and data"), ("read_config", "Read: memory and config"), ("read_media", "Read: images and PDFs"), ("remote", "Remote commands (ssh)"), ("other_tools", "Other tool results"), ("assistant", "Assistant text and tool inputs"), ("user", "Your messages"), ("attachments", "Hooks and attachments"), ("unitemized", "Not itemized (system, tools, thinking)"), ) def num(value): """A usable token count, or 0. Transcripts are data; do not trust their types.""" if isinstance(value, bool) or not isinstance(value, (int, float)): return 0 return value if 0 <= value < 1e15 else 0 def clean_model(model): """Model ids come from the transcript, so only print ones shaped like a model id.""" text = str(model) if len(text) <= MAX_MODEL_CHARS and re.fullmatch(r"claude-(fable|mythos|opus|sonnet|haiku)-[0-9-]{1,20}(\[[a-z0-9]{1,10}\])?", text): return text return "unrecognized" def price_for(model, prompt_tokens): if len(str(model)) > MAX_MODEL_CHARS: return None base = re.sub(r"\[[^\]]*\]$", "", str(model)) base = re.sub(r"-\d{8}$", "", base) if base == HAIKU_5_5: return HAIKU_5_5_LARGE if prompt_tokens > HAIKU_5_5_LIMIT else HAIKU_5_5_SMALL return PRICES.get(base) def classify_tool(name, tool_input): if not isinstance(tool_input, dict): tool_input = {} if name == "Read": path = str(tool_input.get("file_path") or "").replace("\\", "/").lower() ext = os.path.splitext(path)[1] if ext in IMAGE_EXT: return "read_media" if "/memory/" in path or "/.claude/" in path: return "read_config" return "read_docs" if ext in DOC_EXT else "read_code" if name in ("Grep", "Glob"): return "searches" if name in ("Bash", "PowerShell"): command = str(tool_input.get("command", "")) if REMOTE_RE.search(command): return "remote" return "searches" if LOCAL_READ_RE.search(command) else "other_tools" return "other_tools" def parse_time(raw): text = re.sub(r"\.(\d+)", lambda m: "." + m.group(1)[:6].ljust(6, "0"), str(raw).replace("Z", "+00:00"), count=1) try: value = dt.datetime.fromisoformat(text) except ValueError: return None return value if value.tzinfo else value.replace(tzinfo=dt.timezone.utc) def tool_result_tokens(content): if isinstance(content, str): return len(content) / CHARS_PER_TOKEN total = 0.0 if isinstance(content, list): for part in content: if isinstance(part, dict): if part.get("type") == "text": total += len(str(part.get("text", ""))) / CHARS_PER_TOKEN else: total += IMAGE_TOKEN_ESTIMATE return total def json_tokens(value): """Estimated tokens in a parsed JSON value. Deep nesting counts as nothing rather than crashing.""" try: return len(json.dumps(value)) / CHARS_PER_TOKEN except (ValueError, RecursionError): return 0 def read_transcript(path, stats): """Returns (events, calls). Events are in file order; calls holds one entry per API call.""" events, calls, tools = [], {}, {} with open(path, encoding="utf-8", errors="replace") as handle: line_number = 0 while True: line = handle.readline(MAX_LINE_CHARS + 1) if not line: break line_number += 1 if len(line) > MAX_LINE_CHARS: stats["lines_too_long"] += 1 while line and not line.endswith("\n"): # drop the rest without holding it in memory line = handle.readline(MAX_LINE_CHARS + 1) continue try: row = json.loads(line) except (ValueError, RecursionError): stats["lines_unparsed"] += 1 continue if not isinstance(row, dict): continue kind = row.get("type") if kind == "attachment": attachment = row.get("attachment") if isinstance(attachment, dict) and attachment.get("type") != "prompt_snapshot": events.append(("add", "attachments", json_tokens(attachment))) elif kind == "assistant" and not row.get("isApiErrorMessage"): read_assistant(row, events, calls, tools, f"{path}:{line_number}") elif kind == "user": read_user(row, events, tools) return events, calls def read_assistant(row, events, calls, tools, fallback_id): message = row.get("message") if not isinstance(message, dict): return call_id = str(message.get("id") or row.get("uuid") or fallback_id) if len(call_id) > MAX_ID_CHARS: # ids are kept in memory for the whole run call_id = hashlib.sha256(call_id.encode("utf-8", "replace")).hexdigest() if call_id not in calls: calls[call_id] = {"input": 0, "read": 0, "output": 0, "w5": 0, "w1h": 0, "write": 0, "thinking": 0, "has_details": False, "time": row.get("timestamp"), "model": message.get("model"), "fast": False, "us": False} events.append(("call", call_id)) call = calls[call_id] usage = message.get("usage") if isinstance(usage, dict): cache = usage.get("cache_creation") cache = cache if isinstance(cache, dict) else {} details = usage.get("output_tokens_details") for key, value in (("input", usage.get("input_tokens")), ("read", usage.get("cache_read_input_tokens")), ("output", usage.get("output_tokens")), ("w5", cache.get("ephemeral_5m_input_tokens")), ("w1h", cache.get("ephemeral_1h_input_tokens")), ("write", usage.get("cache_creation_input_tokens"))): call[key] = max(call[key], num(value)) if isinstance(details, dict): call["has_details"] = True call["thinking"] = max(call["thinking"], num(details.get("thinking_tokens"))) call["fast"] = call["fast"] or usage.get("speed") == "fast" call["us"] = call["us"] or usage.get("inference_geo") == "us" content = message.get("content") for block in content if isinstance(content, list) else []: if not isinstance(block, dict): continue if block.get("type") == "text": events.append(("add", "assistant", len(str(block.get("text", ""))) / CHARS_PER_TOKEN)) elif block.get("type") == "tool_use": tool_input = block.get("input") tools[block.get("id") if isinstance(block.get("id"), str) else ""] = classify_tool(block.get("name"), tool_input) events.append(("add", "assistant", json_tokens(tool_input))) def read_user(row, events, tools): if row.get("isCompactSummary"): events.append(("compact",)) message = row.get("message") content = message.get("content") if isinstance(message, dict) else None if isinstance(content, str): events.append(("add", "user", len(content) / CHARS_PER_TOKEN)) elif isinstance(content, list): for block in content: if not isinstance(block, dict): continue if block.get("type") == "tool_result": tool_id = block.get("tool_use_id") item = tools.get(tool_id if isinstance(tool_id, str) else "", "other_tools") events.append(("add", item, tool_result_tokens(block.get("content")))) elif block.get("type") == "text": events.append(("add", "user", len(str(block.get("text", ""))) / CHARS_PER_TOKEN)) def new_totals(): return {"cost": 0.0, "tokens": 0, "calls": 0, "transcripts": 0, "input": 0.0, "write": 0.0, "read": 0.0, "output": 0.0, "read_tokens": 0, "main": 0.0, "sub": 0.0, "startup": 0.0, "rewrites": 0.0, "rewrite_calls": 0, "gap_over_60m": 0.0, "gap_5_to_60m": 0.0, "gap_under_5m": 0.0, "thinking": 0.0, "details_calls": 0, "large_calls": 0, "large_cost": 0.0, "fast_calls": 0, "unknown_calls": 0, "unknown_tokens": 0, "compactions": 0} def is_link(path): is_junction = getattr(os.path, "isjunction", None) # Python 3.12 and later return os.path.islink(path) or bool(is_junction and is_junction(path)) def audit(root, since=None, until=None): """Walks every transcript under root. since and until are aware datetimes or None.""" totals = new_totals() stats = collections.Counter() items = collections.Counter() by_model = collections.Counter() unknown_models = collections.Counter() buckets = collections.defaultdict(lambda: {"calls": 0, "cost": 0.0}) caps = {cap: 0.0 for cap in COMPACT_CAPS} contexts, starts_main, starts_sub, peaks, per_transcript = [], [], [], [], [] first_time, last_time = None, None seen_calls = set() paths = [] for folder, subfolders, names in os.walk(root): subfolders[:] = sorted(name for name in subfolders if not is_link(os.path.join(folder, name))) for name in sorted(names): path = os.path.join(folder, name) if name.endswith(".jsonl") and not is_link(path) and os.path.isfile(path): paths.append(path) for path in paths: stats["files"] += 1 is_sub = "subagents" in os.path.relpath(path, root).replace("\\", "/").split("/") try: events, calls = read_transcript(path, stats) except OSError: stats["files_unreadable"] += 1 continue parts = collections.Counter() transcript_cost, transcript_calls, peak = 0.0, 0, 0 start_context, previous_time, is_first, compacted = 0, None, True, False for event in events: if event[0] == "add": parts[event[1]] += event[2] continue if event[0] == "compact": parts = collections.Counter() compacted = True continue call = calls[event[1]] context = call["input"] + call["read"] + call["write"] when = parse_time(call["time"]) if context == 0 and call["output"] == 0: continue # synthetic or errored turn; nothing was billed if is_first: start_context = context duplicate = event[1] in seen_calls # forked or resumed sessions can copy calls seen_calls.add(event[1]) bounded = since is not None or until is not None in_range = not duplicate and (not bounded or (when is not None and not ( (since and when < since) or (until and when > until)))) if bounded and when is None and not duplicate: stats["calls_without_time_skipped"] += 1 if in_range and is_first: (starts_sub if is_sub else starts_main).append(start_context) # forks copy their first call, so they skip this if in_range: if compacted: totals["compactions"] += 1 w5, w1h = call["w5"], call["w1h"] w5 += max(0, call["write"] - w5 - w1h) # no breakdown, or a partial one: price the rest at the 5m rate price = price_for(call["model"], context) totals["calls"] += 1 transcript_calls += 1 totals["tokens"] += context + call["output"] totals["read_tokens"] += call["read"] contexts.append(context) peak = max(peak, context) if call["has_details"]: totals["details_calls"] += 1 if call["fast"]: totals["fast_calls"] += 1 if when: first_time = when if first_time is None or when < first_time else first_time last_time = when if last_time is None or when > last_time else last_time bucket = next(b[2] for b in CONTEXT_BUCKETS if b[0] <= context < b[1]) buckets[bucket]["calls"] += 1 if context > LARGE_CONTEXT: totals["large_calls"] += 1 if price is None: totals["unknown_calls"] += 1 totals["unknown_tokens"] += context + call["output"] unknown_models[clean_model(call["model"])] += 1 else: scale = (FAST_MULTIPLIER if call["fast"] else 1.0) * (US_GEO_MULTIPLIER if call["us"] else 1.0) c_input = call["input"] * price[0] * scale / 1e6 c_write = (w5 * price[1] + w1h * price[2]) * scale / 1e6 c_read = call["read"] * price[3] * scale / 1e6 c_output = call["output"] * price[4] * scale / 1e6 cost = c_input + c_write + c_read + c_output prompt_cost = c_input + c_write + c_read totals["cost"] += cost totals["input"] += c_input totals["write"] += c_write totals["read"] += c_read totals["output"] += c_output totals["sub" if is_sub else "main"] += cost totals["thinking"] += min(call["thinking"], call["output"]) * price[4] * scale / 1e6 by_model[clean_model(call["model"])] += cost transcript_cost += cost buckets[bucket]["cost"] += cost if context > LARGE_CONTEXT: totals["large_cost"] += cost if not is_first and call["write"] > REWRITE_SHARE * context and context > REWRITE_MIN_CONTEXT: totals["rewrites"] += c_write totals["rewrite_calls"] += 1 if previous_time and when: gap = (when - previous_time).total_seconds() key = "gap_under_5m" if gap < 300 else ("gap_5_to_60m" if gap < 3600 else "gap_over_60m") totals[key] += c_write if context: totals["startup"] += prompt_cost * min(start_context, context) / context known = sum(parts.values()) scale_to = max(known, context) # estimates can overshoot; never exceed the real prompt cost for item, tokens in parts.items(): items[item] += prompt_cost * tokens / scale_to items["unitemized"] += prompt_cost * max(0, context - known) / scale_to for cap in COMPACT_CAPS: caps[cap] += c_read * max(0, context - cap) / context if when: previous_time = when is_first, compacted = False, False if transcript_calls: totals["transcripts"] += 1 per_transcript.append(transcript_cost) if not is_sub: peaks.append(peak) return build_result(totals, stats, items, by_model, unknown_models, buckets, caps, contexts, starts_main, starts_sub, peaks, per_transcript, first_time, last_time) def build_result(totals, stats, items, by_model, unknown_models, buckets, caps, contexts, starts_main, starts_sub, peaks, per_transcript, first_time, last_time): cost = totals["cost"] def share(value): return round(100 * value / cost, 1) if cost else 0.0 span = 0 if first_time and last_time: span = max(1, (last_time.date() - first_time.date()).days + 1) calls = totals["calls"] ranked = sorted(per_transcript, reverse=True) metrics = { "list_price_usd": round(cost, 2), "days": span, "transcripts": totals["transcripts"], "model_calls": calls, "total_tokens": totals["tokens"], "cache_read_share_of_tokens": round(100 * totals["read_tokens"] / totals["tokens"], 1) if totals["tokens"] else 0.0, "median_context": int(statistics.median(contexts)) if contexts else 0, "peak_context": max(contexts) if contexts else 0, "median_start_main": int(statistics.median(starts_main)) if starts_main else 0, "median_start_subagent": int(statistics.median(starts_sub)) if starts_sub else 0, "cost_cache_read": share(totals["read"]), "cost_cache_write": share(totals["write"]), "cost_output": share(totals["output"]), "cost_input": share(totals["input"]), "cost_main": share(totals["main"]), "cost_subagents": share(totals["sub"]), "driver_startup": share(totals["startup"]), "driver_rewrites": share(totals["rewrites"]), "rewrites_gap_over_60m": share(totals["gap_over_60m"]), "rewrites_gap_5_to_60m": share(totals["gap_5_to_60m"]), "rewrites_gap_under_5m": share(totals["gap_under_5m"]), "driver_thinking": share(totals["thinking"]), "driver_file_reads": share(items["searches"] + items["read_code"] + items["read_docs"]), "large_context_call_share": round(100 * totals["large_calls"] / calls, 1) if calls else 0.0, "large_context_cost_share": share(totals["large_cost"]), "top_transcripts_cost_share": share(sum(ranked[:TOP_TRANSCRIPTS])), } for cap in COMPACT_CAPS: metrics[f"compact_ceiling_{cap // 1000}k"] = share(caps[cap]) return { "tool": "token_audit", "schema": SCHEMA, "version": VERSION, "prices_as_of": PRICES_AS_OF, "first_date": first_time.date().isoformat() if first_time else None, "last_date": last_time.date().isoformat() if last_time else None, "metrics": metrics, "prompt_items": {key: share(items[key]) for key, _ in ITEM_LABELS}, "models": {name: share(value) for name, value in by_model.most_common(10)}, "context_buckets": {b[2]: {"call_share": round(100 * buckets[b[2]]["calls"] / calls, 1) if calls else 0.0, "cost_share": share(buckets[b[2]]["cost"])} for b in CONTEXT_BUCKETS}, "data_quality": { "files": stats["files"], "files_unreadable": stats["files_unreadable"], "lines_unparsed": stats["lines_unparsed"], "lines_too_long": stats["lines_too_long"], "calls_without_time_skipped": stats["calls_without_time_skipped"], "calls_with_thinking_breakdown": totals["details_calls"], "fast_mode_calls": totals["fast_calls"], "calls_without_a_price": totals["unknown_calls"], "tokens_without_a_price": totals["unknown_tokens"], "models_without_a_price": dict(unknown_models.most_common(10)), "compactions": totals["compactions"], }, } def advice(m): """Each item is (share of cost it relates to, what to try). These overlap and are not savings.""" rows = [] if m["driver_startup"] >= 10: rows.append((m["driver_startup"], "Trim what loads every time: CLAUDE.md, MCP servers, custom agent and skill lists.")) if m["rewrites_gap_over_60m"] >= 5: rows.append((m["rewrites_gap_over_60m"], "Stop returning to cold sessions: start fresh with a short handoff after a long gap.")) if m["compact_ceiling_200k"] >= 5: rows.append((m["compact_ceiling_200k"], "Run shorter sessions: /clear between tasks, /compact at natural breaks, /autocompact to lower the threshold.")) if m["driver_thinking"] >= 10: rows.append((m["driver_thinking"], "Match effort and model to the task: /effort and /model, carefully on hard work.")) if m["driver_file_reads"] >= 15: rows.append((m["driver_file_reads"], "Test a code navigation tool: file reads and searches are a real share of your cost.")) return sorted(rows, key=lambda row: -row[0]) def render(result): m, q = result["metrics"], result["data_quality"] out = [f"token_audit {VERSION} (prices as of {result['prices_as_of']})", ""] if not m["model_calls"]: out.append("No model calls found in that range.") return "\n".join(out) out += [ f"{m['days']} days ({result['first_date']} to {result['last_date']}): {m['transcripts']:,} transcripts, " f"{m['model_calls']:,} model calls, {m['total_tokens']:,} tokens", f"List-price equivalent: ${m['list_price_usd']:,.2f}. On a subscription this is not what you paid.", f"Cache reads: {m['cache_read_share_of_tokens']}% of tokens.", f"Context: median call {m['median_context']:,}, peak {m['peak_context']:,}. " f"Median start: main sessions {m['median_start_main']:,}, subagents {m['median_start_subagent']:,}.", "", "COST BY BILLING CATEGORY (adds up to 100%)", ] def row(label, value): out.append(f" {label:<42}{value:>6.1f}%") row("Cache reads", m["cost_cache_read"]) row("Cache writes", m["cost_cache_write"]) row("Output", m["cost_output"]) row("Uncached input", m["cost_input"]) out += ["", "WHAT DROVE IT (overlaps, does not add up to 100%)"] row("Context loaded at startup", m["driver_startup"]) row("Cache rewrites", m["driver_rewrites"]) row(" more than an hour since the last call", m["rewrites_gap_over_60m"]) row(" 5 to 60 minutes", m["rewrites_gap_5_to_60m"]) row(" under 5 minutes", m["rewrites_gap_under_5m"]) row("Thinking (recorded in the transcript)", m["driver_thinking"]) row("File reads and searches (estimate)", m["driver_file_reads"]) out += ["", "WHAT THE PROMPT WAS MADE OF (estimate: 4 characters per token, share of total cost)"] for key, label in ITEM_LABELS: row(label, result["prompt_items"][key]) out += ["", "CONCENTRATION"] out.append(f" Calls over {LARGE_CONTEXT:,} tokens of context: {m['large_context_call_share']}% of calls, " f"{m['large_context_cost_share']}% of cost") out.append(f" Top {min(TOP_TRANSCRIPTS, m['transcripts'])} of {m['transcripts']:,} transcripts: " f"{m['top_transcripts_cost_share']}% of cost") for name, bucket in result["context_buckets"].items(): out.append(f" {name:<14}{bucket['call_share']:>6.1f}% of calls {bucket['cost_share']:>6.1f}% of cost") out += ["", "COMPACTING EARLIER, UPPER BOUND ON COST AVOIDED"] for cap in COMPACT_CAPS: row(f"At {cap:,} tokens", m[f"compact_ceiling_{cap // 1000}k"]) out.append(" Ignores what compaction costs and what it can lose.") out += ["", "BY MODEL AND WHO RAN IT"] for name, value in result["models"].items(): row(name, value) row("Main sessions", m["cost_main"]) row("Subagents", m["cost_subagents"]) tips = advice(m) out += ["", "WHERE TO LOOK FIRST (share of cost each relates to, not a promised saving)"] if tips: out += [f" {share:>5.1f}% {tip}" for share, tip in tips] else: out.append(" Nothing stands out above the thresholds.") notes = [] if q["calls_without_a_price"]: notes.append(f"{q['calls_without_a_price']:,} calls used a model with no price here and are left out of every cost: " + ", ".join(q["models_without_a_price"])) if q["calls_with_thinking_breakdown"] < m["model_calls"]: notes.append(f"Thinking is only recorded on {q['calls_with_thinking_breakdown']:,} of {m['model_calls']:,} calls, " "so that figure may be low.") if q["files_unreadable"] or q["lines_unparsed"] or q["lines_too_long"]: notes.append(f"Skipped {q['files_unreadable']} unreadable files, {q['lines_unparsed']} unparsed lines " f"and {q['lines_too_long']} oversized lines.") if q["calls_without_time_skipped"]: notes.append(f"Left out {q['calls_without_time_skipped']:,} calls with no usable timestamp because a date range was set.") if q["fast_mode_calls"]: notes.append(f"{q['fast_mode_calls']:,} calls ran in fast mode and are priced at double the base rates.") notes.append("Claude Code deletes transcripts after 30 days by default (cleanupPeriodDays).") out += ["", "NOTES"] + [f" {note}" for note in notes] return "\n".join(out) MAX_BASELINE_BYTES = 10_000_000 BASELINE_METRICS = {"list_price_usd", "model_calls", "median_context", "driver_startup", "driver_rewrites", "rewrites_gap_over_60m", "driver_thinking", "driver_file_reads", "large_context_cost_share", "compact_ceiling_200k"} def render_compare(before, after): b, a = before["metrics"], after["metrics"] out = [f"Baseline {before.get('first_date')} to {before.get('last_date')} vs " f"now {after['first_date']} to {after['last_date']}", ""] if not a["model_calls"]: return "\n".join(out + ["No model calls found in that range, so there is nothing to compare."]) if before.get("prices_as_of") != after["prices_as_of"]: out += ["Prices changed between these runs, so cost per call also reflects the price change.", ""] if b.get("model_calls") and a["model_calls"]: b_each, a_each = b["list_price_usd"] / b["model_calls"], a["list_price_usd"] / a["model_calls"] out.append(f" {'List-price cost per model call':<42}${b_each:>8.4f} ${a_each:>8.4f}") out.append(f" {'Median context':<42}{b.get('median_context', 0):>9,} {a['median_context']:>9,}") for key, label in (("driver_startup", "Context loaded at startup"), ("driver_rewrites", "Cache rewrites"), ("rewrites_gap_over_60m", "Rewrites after an hour away"), ("driver_thinking", "Thinking"), ("driver_file_reads", "File reads and searches"), ("large_context_cost_share", "Cost in calls over 250k"), ("compact_ceiling_200k", "Compaction ceiling at 200k")): out.append(f" {label:<42}{b.get(key, 0):>8.1f}% {a[key]:>8.1f}% ({a[key] - b.get(key, 0):+.1f} points)") out += ["", "Compare the cost per call and the shares, not total dollars: how much you worked changes those."] return "\n".join(out) def load_baseline(path): with open(path, "rb") as handle: raw = handle.read(MAX_BASELINE_BYTES + 1) if len(raw) > MAX_BASELINE_BYTES: raise ValueError("baseline is too large") # PowerShell 5.1 redirects as UTF-16, so accept that as well as UTF-8. text = raw.decode("utf-16" if raw[:2] in (b"\xff\xfe", b"\xfe\xff") else "utf-8-sig") try: data = json.loads(text) except RecursionError: raise ValueError("baseline is nested too deeply") from None if not (isinstance(data, dict) and data.get("tool") == "token_audit" and data.get("schema") == SCHEMA and isinstance(data.get("metrics"), dict)): raise ValueError("not a token_audit --json file from this version") metrics = data["metrics"] if not all(num(value) == value for value in metrics.values()): raise ValueError("baseline metrics must be numbers") if not BASELINE_METRICS.issubset(metrics): raise ValueError("baseline is missing metrics") for key in ("first_date", "last_date"): if not (data.get(key) is None or (isinstance(data[key], str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", data[key], re.ASCII))): raise ValueError("baseline dates must look like 2026-11-03") return data def parse_date(text, end_of_day=False): value = dt.datetime.strptime(text, "%Y-%m-%d").replace(tzinfo=dt.timezone.utc) return value + dt.timedelta(days=1, microseconds=-1) if end_of_day else value def main(argv=None): """Never let a traceback print, because tracebacks contain local paths.""" try: return run(argv) except KeyboardInterrupt: return 130 except Exception as err: print(f"token_audit stopped: {type(err).__name__}", file=sys.stderr) return 1 def run(argv): parser = argparse.ArgumentParser(description="See where your Claude Code tokens go. Local and read-only.") parser.add_argument("--root", default=os.path.join(os.path.expanduser("~"), ".claude", "projects"), help="transcript folder (default: ~/.claude/projects)") parser.add_argument("--days", type=int, help="only the last N days") parser.add_argument("--since", help="only from this date, YYYY-MM-DD") parser.add_argument("--until", help="only through this date, YYYY-MM-DD") output = parser.add_mutually_exclusive_group() output.add_argument("--json", action="store_true", help="print JSON, for saving a baseline") output.add_argument("--compare", metavar="BASELINE.json", help="compare against a saved --json baseline") parser.add_argument("--version", action="version", version=VERSION) args = parser.parse_args(argv) if args.days is not None and (args.since or not 1 <= args.days <= 36500): parser.error("--days must be from 1 to 36500 and cannot be combined with --since") try: since = parse_date(args.since) if args.since else None until = parse_date(args.until, end_of_day=True) if args.until else None except ValueError: parser.error("dates must look like 2026-11-03") if args.days: since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=args.days) if not os.path.isdir(args.root): print("No transcript folder found. Use --root if yours is somewhere else.", file=sys.stderr) return 1 baseline = None if args.compare: try: baseline = load_baseline(args.compare) except (OSError, ValueError) as err: print(f"Could not use that baseline: {type(err).__name__}", file=sys.stderr) return 1 result = audit(args.root, since, until) if args.json: print(json.dumps(result, indent=2)) elif baseline: print(render_compare(baseline, result)) else: print(render(result)) return 0 if __name__ == "__main__": sys.exit(main())