Created
April 6, 2026 04:57
-
-
Save dslounge/410c8774727e0508185fb630b49f58b3 to your computer and use it in GitHub Desktop.
Analyze Claude Code token usage and cache efficiency from local JSONL conversation logs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """Analyze Claude Code token usage and cache efficiency from local JSONL logs. | |
| Reads conversation logs from ~/.claude/projects/, extracts token usage data, | |
| classifies cache patterns, and generates reports to identify what's burning | |
| through weekly usage. | |
| """ | |
| import argparse | |
| import json | |
| import glob | |
| import os | |
| import sys | |
| from collections import defaultdict | |
| from datetime import datetime, timedelta, timezone | |
| # ── Configuration ────────────────────────────────────────────────────────── | |
| CLAUDE_DIR = os.path.expanduser("~/.claude/projects") | |
| WEEKS_BACK = 4 | |
| CUTOFF = datetime.now(timezone.utc) - timedelta(weeks=WEEKS_BACK) | |
| # Cache miss detection thresholds (Pattern C) | |
| MIN_MSG_INDEX = 4 # skip initialization messages 0-3 | |
| MIN_GAP_SECONDS = 300 # 5 minutes | |
| MIN_CREATION_TOKENS = 5000 # recommended threshold for 95% accuracy | |
| # ── Pricing (per token, not per MTok) ────────────────────────────────────── | |
| PRICING = { | |
| "claude-opus-4-6": { | |
| "input": 5.0 / 1_000_000, | |
| "cache_write": 10.0 / 1_000_000, # 1h cache, 2x | |
| "cache_read": 0.5 / 1_000_000, # 0.1x | |
| "output": 25.0 / 1_000_000, | |
| }, | |
| "claude-sonnet-4-6": { | |
| "input": 3.0 / 1_000_000, | |
| "cache_write": 3.75 / 1_000_000, # 1.25x (sonnet uses 5m cache) | |
| "cache_read": 0.3 / 1_000_000, # 0.1x | |
| "output": 15.0 / 1_000_000, | |
| }, | |
| "claude-haiku-4-5-20251001": { | |
| "input": 1.0 / 1_000_000, | |
| "cache_write": 1.25 / 1_000_000, # 5m cache, 1.25x | |
| "cache_read": 0.1 / 1_000_000, # 0.1x | |
| "output": 5.0 / 1_000_000, | |
| }, | |
| } | |
| # Fallback to Opus pricing for unknown models | |
| DEFAULT_PRICING = PRICING["claude-opus-4-6"] | |
| def get_pricing(model): | |
| return PRICING.get(model, DEFAULT_PRICING) | |
| def compute_cost(msg): | |
| p = get_pricing(msg["model"]) | |
| return ( | |
| msg["input_tokens"] * p["input"] | |
| + msg["cache_creation_input_tokens"] * p["cache_write"] | |
| + msg["cache_read_input_tokens"] * p["cache_read"] | |
| + msg["output_tokens"] * p["output"] | |
| ) | |
| # ── Pattern Classification ───────────────────────────────────────────────── | |
| def classify_pattern(msg, prev_msg, msg_index): | |
| """Classify a message into one of five cache patterns.""" | |
| if msg_index <= 3: | |
| return "initialization" | |
| creation = msg["cache_creation_input_tokens"] | |
| gap = msg["gap_seconds"] | |
| # Pattern C: Time gap recovery (true cache miss) | |
| if gap >= MIN_GAP_SECONDS and creation >= MIN_CREATION_TOKENS: | |
| return "time_gap_recovery" | |
| # Pattern E: Agentic loop (identical cache values, tiny gap) | |
| if prev_msg and gap < 5: | |
| if (creation == prev_msg["cache_creation_input_tokens"] | |
| and msg["cache_read_input_tokens"] == prev_msg["cache_read_input_tokens"] | |
| and creation > 0): | |
| return "agentic_loop" | |
| # Pattern D: Context injection (high creation, short gap) | |
| if creation >= 20000 and gap < MIN_GAP_SECONDS: | |
| return "context_injection" | |
| # Pattern B: Normal streaming | |
| return "normal" | |
| # ── Data Extraction ──────────────────────────────────────────────────────── | |
| def decode_project_name(slug): | |
| """Convert project slug back to a readable path.""" | |
| # -Users-rmendiola-code-me-raf-dev-astro -> ~/code/me/raf-dev-astro | |
| parts = slug.split("-") | |
| # Find "Users" and skip the home prefix | |
| try: | |
| idx = parts.index("Users") | |
| # Skip Users, username | |
| remainder = parts[idx + 2:] | |
| return "/".join(remainder) | |
| except (ValueError, IndexError): | |
| return slug | |
| def find_jsonl_files(): | |
| """Find all JSONL files including subagent logs.""" | |
| patterns = [ | |
| os.path.join(CLAUDE_DIR, "*", "*.jsonl"), | |
| os.path.join(CLAUDE_DIR, "*", "*", "subagents", "*.jsonl"), | |
| ] | |
| files = [] | |
| for pattern in patterns: | |
| files.extend(glob.glob(pattern)) | |
| return files | |
| def extract_messages(filepath): | |
| """Extract assistant messages with usage data from a JSONL file.""" | |
| session_id = os.path.splitext(os.path.basename(filepath))[0] | |
| project_dir = os.path.basename(os.path.dirname(filepath)) | |
| # Handle subagent paths: .../project/session/subagents/file.jsonl | |
| if os.path.basename(os.path.dirname(filepath)) == "subagents": | |
| project_dir = os.path.basename( | |
| os.path.dirname(os.path.dirname(os.path.dirname(filepath))) | |
| ) | |
| session_id = f"subagent-{session_id}" | |
| project_name = decode_project_name(project_dir) | |
| messages = [] | |
| msg_index = 0 | |
| prev_ts = None | |
| synthetic_count = 0 | |
| try: | |
| with open(filepath) as f: | |
| for line in f: | |
| line = line.strip() | |
| if not line: | |
| continue | |
| try: | |
| obj = json.loads(line) | |
| except json.JSONDecodeError: | |
| continue | |
| if obj.get("type") != "assistant": | |
| continue | |
| inner = obj.get("message", {}) | |
| if not isinstance(inner, dict): | |
| continue | |
| model = inner.get("model", "") | |
| usage = inner.get("usage", {}) | |
| # Count synthetic errors separately | |
| if model == "<synthetic>": | |
| synthetic_count += 1 | |
| continue | |
| if not usage or not usage.get("output_tokens", 0) and not usage.get("input_tokens", 0): | |
| continue | |
| ts_str = obj.get("timestamp", "") | |
| try: | |
| ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) | |
| except (ValueError, AttributeError): | |
| continue | |
| if ts < CUTOFF: | |
| continue | |
| cache_creation = usage.get("cache_creation", {}) | |
| gap = (ts - prev_ts).total_seconds() if prev_ts else 0 | |
| msg = { | |
| "timestamp": ts, | |
| "session_id": session_id, | |
| "project": project_name, | |
| "model": model, | |
| "version": obj.get("version", ""), | |
| "input_tokens": usage.get("input_tokens", 0), | |
| "cache_creation_input_tokens": usage.get("cache_creation_input_tokens", 0), | |
| "cache_read_input_tokens": usage.get("cache_read_input_tokens", 0), | |
| "output_tokens": usage.get("output_tokens", 0), | |
| "cache_5m_tokens": cache_creation.get("ephemeral_5m_input_tokens", 0), | |
| "cache_1h_tokens": cache_creation.get("ephemeral_1h_input_tokens", 0), | |
| "stop_reason": inner.get("stop_reason", ""), | |
| "msg_index": msg_index, | |
| "gap_seconds": gap, | |
| } | |
| msg["cost"] = compute_cost(msg) | |
| prev_msg = messages[-1] if messages else None | |
| msg["pattern"] = classify_pattern(msg, prev_msg, msg_index) | |
| messages.append(msg) | |
| prev_ts = ts | |
| msg_index += 1 | |
| except (OSError, IOError) as e: | |
| print(f" Warning: could not read {filepath}: {e}", file=sys.stderr) | |
| return messages, synthetic_count | |
| # ── Report Helpers ───────────────────────────────────────────────────────── | |
| def fmt_cost(c): | |
| return f"${c:,.2f}" | |
| def fmt_tokens(t): | |
| if t >= 1_000_000_000: | |
| return f"{t / 1_000_000_000:.1f}B" | |
| if t >= 1_000_000: | |
| return f"{t / 1_000_000:.1f}M" | |
| if t >= 1_000: | |
| return f"{t / 1_000:.1f}K" | |
| return str(t) | |
| def fmt_pct(ratio): | |
| return f"{ratio * 100:.1f}%" | |
| def cache_hit_rate(msgs): | |
| total_read = sum(m["cache_read_input_tokens"] for m in msgs) | |
| total_create = sum(m["cache_creation_input_tokens"] for m in msgs) | |
| total_input = sum(m["input_tokens"] for m in msgs) | |
| denom = total_read + total_create + total_input | |
| return total_read / denom if denom > 0 else 0 | |
| def print_divider(title): | |
| print(f"\n{'='*72}") | |
| print(f" {title}") | |
| print(f"{'='*72}\n") | |
| # ── Reports ──────────────────────────────────────────────────────────────── | |
| def report_weekly_summary(all_msgs): | |
| print_divider("WEEKLY SUMMARY") | |
| # Group by ISO week | |
| weeks = defaultdict(list) | |
| for m in all_msgs: | |
| week_key = m["timestamp"].strftime("%Y-W%W") | |
| weeks[week_key].append(m) | |
| header = f"{'Week':<12} {'Messages':>8} {'Input':>8} {'CacheWr':>8} {'CacheRd':>8} {'Output':>8} {'Cost':>10} {'Hit Rate':>9}" | |
| print(header) | |
| print("-" * len(header)) | |
| for week in sorted(weeks.keys()): | |
| msgs = weeks[week] | |
| inp = sum(m["input_tokens"] for m in msgs) | |
| cw = sum(m["cache_creation_input_tokens"] for m in msgs) | |
| cr = sum(m["cache_read_input_tokens"] for m in msgs) | |
| out = sum(m["output_tokens"] for m in msgs) | |
| cost = sum(m["cost"] for m in msgs) | |
| hr = cache_hit_rate(msgs) | |
| print(f"{week:<12} {len(msgs):>8} {fmt_tokens(inp):>8} {fmt_tokens(cw):>8} {fmt_tokens(cr):>8} {fmt_tokens(out):>8} {fmt_cost(cost):>10} {fmt_pct(hr):>9}") | |
| # Model breakdown | |
| print(f"\n Model Breakdown:") | |
| by_model = defaultdict(list) | |
| for m in all_msgs: | |
| by_model[m["model"]].append(m) | |
| for model in sorted(by_model.keys()): | |
| msgs = by_model[model] | |
| cost = sum(m["cost"] for m in msgs) | |
| hr = cache_hit_rate(msgs) | |
| print(f" {model}: {len(msgs)} messages, {fmt_cost(cost)}, cache hit rate {fmt_pct(hr)}") | |
| def report_top_conversations(all_msgs): | |
| print_divider("TOP 10 MOST EXPENSIVE CONVERSATIONS") | |
| by_session = defaultdict(list) | |
| for m in all_msgs: | |
| by_session[m["session_id"]].append(m) | |
| sessions = [] | |
| for sid, msgs in by_session.items(): | |
| cost = sum(m["cost"] for m in msgs) | |
| misses = sum(1 for m in msgs if m["pattern"] == "time_gap_recovery") | |
| versions = set(m["version"] for m in msgs if m["version"]) | |
| sessions.append({ | |
| "session_id": sid[:12], | |
| "project": msgs[0]["project"], | |
| "cost": cost, | |
| "messages": len(msgs), | |
| "hit_rate": cache_hit_rate(msgs), | |
| "misses": misses, | |
| "versions": ", ".join(sorted(versions)), | |
| }) | |
| sessions.sort(key=lambda s: s["cost"], reverse=True) | |
| header = f"{'Session':<14} {'Project':<30} {'Cost':>10} {'Msgs':>6} {'HitRate':>8} {'Misses':>7} {'Version':<10}" | |
| print(header) | |
| print("-" * len(header)) | |
| for s in sessions[:10]: | |
| proj = s["project"][:29] | |
| print(f"{s['session_id']:<14} {proj:<30} {fmt_cost(s['cost']):>10} {s['messages']:>6} {fmt_pct(s['hit_rate']):>8} {s['misses']:>7} {s['versions']:<10}") | |
| def report_top_projects(all_msgs): | |
| print_divider("TOP 5 MOST EXPENSIVE PROJECTS") | |
| by_project = defaultdict(list) | |
| for m in all_msgs: | |
| by_project[m["project"]].append(m) | |
| projects = [] | |
| for proj, msgs in by_project.items(): | |
| sessions = set(m["session_id"] for m in msgs) | |
| projects.append({ | |
| "project": proj, | |
| "cost": sum(m["cost"] for m in msgs), | |
| "sessions": len(sessions), | |
| "messages": len(msgs), | |
| "hit_rate": cache_hit_rate(msgs), | |
| }) | |
| projects.sort(key=lambda p: p["cost"], reverse=True) | |
| header = f"{'Project':<40} {'Cost':>10} {'Sessions':>9} {'Msgs':>7} {'HitRate':>8}" | |
| print(header) | |
| print("-" * len(header)) | |
| for p in projects[:5]: | |
| proj = p["project"][:39] | |
| print(f"{proj:<40} {fmt_cost(p['cost']):>10} {p['sessions']:>9} {p['messages']:>7} {fmt_pct(p['hit_rate']):>8}") | |
| def report_cache_anomalies(all_msgs): | |
| print_divider("CACHE ANOMALY REPORT") | |
| patterns = defaultdict(int) | |
| for m in all_msgs: | |
| patterns[m["pattern"]] += 1 | |
| print(" Pattern Distribution:") | |
| for pat in ["initialization", "normal", "time_gap_recovery", "context_injection", "agentic_loop"]: | |
| count = patterns.get(pat, 0) | |
| pct = count / len(all_msgs) * 100 if all_msgs else 0 | |
| label = { | |
| "initialization": "A: Initialization (msgs 0-3)", | |
| "normal": "B: Normal streaming", | |
| "time_gap_recovery": "C: Time gap recovery (TRUE MISS)", | |
| "context_injection": "D: Context injection", | |
| "agentic_loop": "E: Agentic loop", | |
| }.get(pat, pat) | |
| flag = " <<<" if pat == "time_gap_recovery" else "" | |
| print(f" {label}: {count} ({pct:.1f}%){flag}") | |
| # True cache misses detail | |
| misses = [m for m in all_msgs if m["pattern"] == "time_gap_recovery"] | |
| if misses: | |
| total_miss_cost = sum(m["cost"] for m in misses) | |
| # Estimate what these would have cost with cache hits | |
| # Replace cache_creation with equivalent cache_read pricing | |
| counterfactual_cost = 0 | |
| for m in misses: | |
| p = get_pricing(m["model"]) | |
| counterfactual_cost += ( | |
| m["input_tokens"] * p["input"] | |
| + m["cache_creation_input_tokens"] * p["cache_read"] # if it had been a hit | |
| + m["cache_read_input_tokens"] * p["cache_read"] | |
| + m["output_tokens"] * p["output"] | |
| ) | |
| waste = total_miss_cost - counterfactual_cost | |
| print(f"\n True Cache Misses (Pattern C):") | |
| print(f" Count: {len(misses)}") | |
| print(f" Total cost: {fmt_cost(total_miss_cost)}") | |
| print(f" Cost if cache had hit: {fmt_cost(counterfactual_cost)}") | |
| print(f" Wasted spend: {fmt_cost(waste)}") | |
| # Show worst misses | |
| misses.sort(key=lambda m: m["cost"], reverse=True) | |
| print(f"\n Top 5 Most Expensive Cache Misses:") | |
| header = f" {'Project':<30} {'Gap':>8} {'CacheWr':>10} {'Cost':>10} {'Version':<10}" | |
| print(header) | |
| for m in misses[:5]: | |
| gap_hrs = m["gap_seconds"] / 3600 | |
| gap_str = f"{gap_hrs:.1f}h" if gap_hrs >= 1 else f"{m['gap_seconds']/60:.0f}m" | |
| print(f" {m['project'][:29]:<30} {gap_str:>8} {fmt_tokens(m['cache_creation_input_tokens']):>10} {fmt_cost(m['cost']):>10} {m['version']:<10}") | |
| # Sessions with low cache hit rate (excluding init messages) | |
| by_session = defaultdict(list) | |
| for m in all_msgs: | |
| if m["msg_index"] > 3: | |
| by_session[m["session_id"]].append(m) | |
| low_sessions = [] | |
| for sid, msgs in by_session.items(): | |
| hr = cache_hit_rate(msgs) | |
| if hr < 0.60 and len(msgs) >= 5: | |
| low_sessions.append((sid[:12], msgs[0]["project"], hr, len(msgs))) | |
| if low_sessions: | |
| low_sessions.sort(key=lambda x: x[2]) | |
| print(f"\n Sessions with Cache Hit Rate Below 60% (min 5 msgs, excluding init):") | |
| for sid, proj, hr, count in low_sessions[:10]: | |
| print(f" {sid} ({proj[:25]}): {fmt_pct(hr)} over {count} messages") | |
| def report_version_comparison(all_msgs): | |
| print_divider("VERSION COMPARISON") | |
| by_version = defaultdict(list) | |
| for m in all_msgs: | |
| v = m["version"] | |
| if v: | |
| by_version[v].append(m) | |
| header = f"{'Version':<12} {'Messages':>8} {'Cost/Msg':>10} {'HitRate':>8} {'Misses':>7} {'Miss/Sess':>10}" | |
| print(header) | |
| print("-" * len(header)) | |
| for version in sorted(by_version.keys()): | |
| msgs = by_version[version] | |
| avg_cost = sum(m["cost"] for m in msgs) / len(msgs) | |
| hr = cache_hit_rate(msgs) | |
| miss_count = sum(1 for m in msgs if m["pattern"] == "time_gap_recovery") | |
| sessions = len(set(m["session_id"] for m in msgs)) | |
| miss_per_session = miss_count / sessions if sessions > 0 else 0 | |
| flag = "" | |
| from packaging.version import Version | |
| try: | |
| if Version(version) < Version("2.1.91"): | |
| flag = " (pre-fix)" | |
| except Exception: | |
| pass | |
| print(f"{version:<12} {len(msgs):>8} {fmt_cost(avg_cost):>10} {fmt_pct(hr):>8} {miss_count:>7} {miss_per_session:>9.1f}{flag}") | |
| def report_version_comparison_simple(all_msgs): | |
| """Version comparison without packaging dependency.""" | |
| print_divider("VERSION COMPARISON") | |
| by_version = defaultdict(list) | |
| for m in all_msgs: | |
| v = m["version"] | |
| if v: | |
| by_version[v].append(m) | |
| header = f"{'Version':<12} {'Messages':>8} {'Cost/Msg':>10} {'HitRate':>8} {'Misses':>7} {'Miss/Sess':>10}" | |
| print(header) | |
| print("-" * len(header)) | |
| for version in sorted(by_version.keys()): | |
| msgs = by_version[version] | |
| avg_cost = sum(m["cost"] for m in msgs) / len(msgs) | |
| hr = cache_hit_rate(msgs) | |
| miss_count = sum(1 for m in msgs if m["pattern"] == "time_gap_recovery") | |
| sessions = len(set(m["session_id"] for m in msgs)) | |
| miss_per_session = miss_count / sessions if sessions > 0 else 0 | |
| # Simple version comparison: split on dots, compare as tuples | |
| flag = "" | |
| try: | |
| parts = [int(x) for x in version.split(".")] | |
| if parts < [2, 1, 91]: | |
| flag = " (pre-fix)" | |
| except (ValueError, IndexError): | |
| pass | |
| print(f"{version:<12} {len(msgs):>8} {fmt_cost(avg_cost):>10} {fmt_pct(hr):>8} {miss_count:>7} {miss_per_session:>9.1f}{flag}") | |
| def report_context_overflow(synthetic_count): | |
| print_divider("CONTEXT OVERFLOW") | |
| print(f" 'Prompt is too long' errors (synthetic messages): {synthetic_count}") | |
| def report_time_of_day(all_msgs): | |
| print_divider("TIME-OF-DAY ANALYSIS (Cache Misses)") | |
| misses = [m for m in all_msgs if m["pattern"] == "time_gap_recovery"] | |
| if not misses: | |
| print(" No cache misses detected.") | |
| return | |
| by_hour = defaultdict(int) | |
| for m in misses: | |
| by_hour[m["timestamp"].hour] += 1 | |
| max_count = max(by_hour.values()) if by_hour else 1 | |
| print(f" Hour Misses Bar") | |
| print(f" ---- ------ ---") | |
| for hour in range(24): | |
| count = by_hour.get(hour, 0) | |
| bar_len = int(count / max_count * 40) if max_count > 0 else 0 | |
| bar = "█" * bar_len | |
| if count > 0: | |
| print(f" {hour:02d}:00 {count:>6} {bar}") | |
| def report_weekly_deep_dive(all_msgs): | |
| """Detailed week-by-week breakdown with per-project drill-down.""" | |
| print_divider("WEEKLY DEEP DIVE") | |
| weeks = defaultdict(list) | |
| for m in all_msgs: | |
| week_key = m["timestamp"].strftime("%Y-W%W") | |
| weeks[week_key].append(m) | |
| for week in sorted(weeks.keys()): | |
| msgs = weeks[week] | |
| dates = [m["timestamp"] for m in msgs] | |
| start = min(dates).strftime("%b %d") | |
| end = max(dates).strftime("%b %d") | |
| sessions = set(m["session_id"] for m in msgs) | |
| projects = set(m["project"] for m in msgs) | |
| total_cost = sum(m["cost"] for m in msgs) | |
| hr = cache_hit_rate(msgs) | |
| misses = [m for m in msgs if m["pattern"] == "time_gap_recovery"] | |
| # Waste calculation | |
| miss_cost = sum(m["cost"] for m in misses) | |
| counterfactual = sum( | |
| m["input_tokens"] * get_pricing(m["model"])["input"] | |
| + m["cache_creation_input_tokens"] * get_pricing(m["model"])["cache_read"] | |
| + m["cache_read_input_tokens"] * get_pricing(m["model"])["cache_read"] | |
| + m["output_tokens"] * get_pricing(m["model"])["output"] | |
| for m in misses | |
| ) | |
| waste = miss_cost - counterfactual | |
| # Token totals | |
| inp = sum(m["input_tokens"] for m in msgs) | |
| cw = sum(m["cache_creation_input_tokens"] for m in msgs) | |
| cr = sum(m["cache_read_input_tokens"] for m in msgs) | |
| out = sum(m["output_tokens"] for m in msgs) | |
| print(f" ┌─ {week} ({start} - {end}) ─────────────────────────────────────") | |
| print(f" │ {len(msgs):,} messages | {len(sessions):,} sessions | {len(projects):,} projects") | |
| print(f" ��� Cost: {fmt_cost(total_cost)} | Cache hit rate: {fmt_pct(hr)}") | |
| print(f" │ Tokens: input={fmt_tokens(inp)} write={fmt_tokens(cw)} read={fmt_tokens(cr)} output={fmt_tokens(out)}") | |
| print(f" │ Cache misses: {len(misses)} | Wasted: {fmt_cost(waste)}") | |
| # Model breakdown | |
| by_model = defaultdict(list) | |
| for m in msgs: | |
| short = m["model"].replace("claude-", "").replace("-20251001", "") | |
| by_model[short].append(m) | |
| model_parts = [] | |
| for model in sorted(by_model.keys()): | |
| mm = by_model[model] | |
| model_parts.append(f"{model}: {len(mm):,} msgs {fmt_cost(sum(m['cost'] for m in mm))}") | |
| print(f" │ Models: {' | '.join(model_parts)}") | |
| # Per-project breakdown | |
| by_project = defaultdict(list) | |
| for m in msgs: | |
| by_project[m["project"]].append(m) | |
| project_rows = [] | |
| for proj, pmsg in by_project.items(): | |
| pcost = sum(m["cost"] for m in pmsg) | |
| phr = cache_hit_rate(pmsg) | |
| psessions = len(set(m["session_id"] for m in pmsg)) | |
| pmisses = sum(1 for m in pmsg if m["pattern"] == "time_gap_recovery") | |
| project_rows.append((proj, pcost, len(pmsg), psessions, phr, pmisses)) | |
| project_rows.sort(key=lambda r: r[1], reverse=True) | |
| print(f" │") | |
| print(f" │ {'Project':<38} {'Cost':>10} {'Msgs':>7} {'Sess':>5} {'HitRate':>8} {'Miss':>5}") | |
| print(f" │ {'─'*38} {'─'*10} {'─'*7} {'─'*5} {'─'*8} {'─'*5}") | |
| # Show all projects, not just top N | |
| for proj, pcost, pmsg_count, psessions, phr, pmisses in project_rows: | |
| proj_display = proj[:37] | |
| miss_str = str(pmisses) if pmisses > 0 else "" | |
| print(f" │ {proj_display:<38} {fmt_cost(pcost):>10} {pmsg_count:>7} {psessions:>5} {fmt_pct(phr):>8} {miss_str:>5}") | |
| print(f" └{'─'*71}") | |
| print() | |
| def report_daily_cache_trend(all_msgs): | |
| print_divider("DAILY CACHE EFFICIENCY TREND") | |
| by_day = defaultdict(list) | |
| for m in all_msgs: | |
| day = m["timestamp"].strftime("%Y-%m-%d") | |
| by_day[day].append(m) | |
| header = f"{'Date':<12} {'Msgs':>6} {'Cost':>10} {'HitRate':>8} {'Misses':>7}" | |
| print(header) | |
| print("-" * len(header)) | |
| for day in sorted(by_day.keys()): | |
| msgs = by_day[day] | |
| cost = sum(m["cost"] for m in msgs) | |
| hr = cache_hit_rate(msgs) | |
| misses = sum(1 for m in msgs if m["pattern"] == "time_gap_recovery") | |
| print(f"{day:<12} {len(msgs):>6} {fmt_cost(cost):>10} {fmt_pct(hr):>8} {misses:>7}") | |
| # ── JSON Export ──────────────────────────────────────────────────────────── | |
| def export_json(all_msgs, synthetic_count, output_path): | |
| """Export a structured JSON summary for blog post reference.""" | |
| by_week = defaultdict(list) | |
| for m in all_msgs: | |
| by_week[m["timestamp"].strftime("%Y-W%W")].append(m) | |
| by_project = defaultdict(list) | |
| for m in all_msgs: | |
| by_project[m["project"]].append(m) | |
| by_session = defaultdict(list) | |
| for m in all_msgs: | |
| by_session[m["session_id"]].append(m) | |
| by_version = defaultdict(list) | |
| for m in all_msgs: | |
| if m["version"]: | |
| by_version[m["version"]].append(m) | |
| patterns = defaultdict(int) | |
| for m in all_msgs: | |
| patterns[m["pattern"]] += 1 | |
| misses = [m for m in all_msgs if m["pattern"] == "time_gap_recovery"] | |
| total_miss_cost = sum(m["cost"] for m in misses) | |
| counterfactual = sum( | |
| m["input_tokens"] * get_pricing(m["model"])["input"] | |
| + m["cache_creation_input_tokens"] * get_pricing(m["model"])["cache_read"] | |
| + m["cache_read_input_tokens"] * get_pricing(m["model"])["cache_read"] | |
| + m["output_tokens"] * get_pricing(m["model"])["output"] | |
| for m in misses | |
| ) | |
| report = { | |
| "generated_at": datetime.now(timezone.utc).isoformat(), | |
| "analysis_window": { | |
| "from": CUTOFF.isoformat(), | |
| "to": datetime.now(timezone.utc).isoformat(), | |
| "weeks": WEEKS_BACK, | |
| }, | |
| "totals": { | |
| "messages": len(all_msgs), | |
| "sessions": len(by_session), | |
| "projects": len(by_project), | |
| "total_cost": round(sum(m["cost"] for m in all_msgs), 2), | |
| "input_tokens": sum(m["input_tokens"] for m in all_msgs), | |
| "cache_creation_tokens": sum(m["cache_creation_input_tokens"] for m in all_msgs), | |
| "cache_read_tokens": sum(m["cache_read_input_tokens"] for m in all_msgs), | |
| "output_tokens": sum(m["output_tokens"] for m in all_msgs), | |
| "cache_hit_rate": round(cache_hit_rate(all_msgs), 4), | |
| "synthetic_errors": synthetic_count, | |
| }, | |
| "patterns": dict(patterns), | |
| "cache_misses": { | |
| "count": len(misses), | |
| "total_cost": round(total_miss_cost, 2), | |
| "counterfactual_cost": round(counterfactual, 2), | |
| "wasted_spend": round(total_miss_cost - counterfactual, 2), | |
| }, | |
| "weekly": [ | |
| { | |
| "week": w, | |
| "messages": len(msgs), | |
| "cost": round(sum(m["cost"] for m in msgs), 2), | |
| "cache_hit_rate": round(cache_hit_rate(msgs), 4), | |
| "misses": sum(1 for m in msgs if m["pattern"] == "time_gap_recovery"), | |
| } | |
| for w, msgs in sorted(by_week.items()) | |
| ], | |
| "top_projects": sorted( | |
| [ | |
| { | |
| "project": proj, | |
| "cost": round(sum(m["cost"] for m in msgs), 2), | |
| "sessions": len(set(m["session_id"] for m in msgs)), | |
| "messages": len(msgs), | |
| "cache_hit_rate": round(cache_hit_rate(msgs), 4), | |
| } | |
| for proj, msgs in by_project.items() | |
| ], | |
| key=lambda x: x["cost"], | |
| reverse=True, | |
| )[:10], | |
| "versions": sorted( | |
| [ | |
| { | |
| "version": v, | |
| "messages": len(msgs), | |
| "avg_cost": round(sum(m["cost"] for m in msgs) / len(msgs), 4), | |
| "cache_hit_rate": round(cache_hit_rate(msgs), 4), | |
| "misses": sum(1 for m in msgs if m["pattern"] == "time_gap_recovery"), | |
| } | |
| for v, msgs in by_version.items() | |
| ], | |
| key=lambda x: x["version"], | |
| ), | |
| } | |
| with open(output_path, "w") as f: | |
| json.dump(report, f, indent=2) | |
| print(f"\n JSON report written to {output_path}") | |
| # ── Main ─────────────────────────────────────────────────────────────────── | |
| def load_all_messages(): | |
| """Load and return all messages plus synthetic count.""" | |
| print(f"Claude Code Usage Analysis") | |
| print(f"Analysis window: {CUTOFF.strftime('%Y-%m-%d')} to {datetime.now(timezone.utc).strftime('%Y-%m-%d')}") | |
| print(f"Source: {CLAUDE_DIR}") | |
| files = find_jsonl_files() | |
| print(f"Found {len(files)} JSONL files") | |
| all_msgs = [] | |
| total_synthetic = 0 | |
| files_with_data = 0 | |
| for filepath in files: | |
| msgs, syn = extract_messages(filepath) | |
| if msgs: | |
| all_msgs.extend(msgs) | |
| files_with_data += 1 | |
| total_synthetic += syn | |
| all_msgs.sort(key=lambda m: m["timestamp"]) | |
| print(f"Extracted {len(all_msgs)} assistant messages from {files_with_data} files") | |
| print(f"Total API-equivalent cost: {fmt_cost(sum(m['cost'] for m in all_msgs))}") | |
| return all_msgs, total_synthetic | |
| def main(): | |
| parser = argparse.ArgumentParser( | |
| description="Analyze Claude Code token usage and cache efficiency." | |
| ) | |
| parser.add_argument( | |
| "--weekly", action="store_true", | |
| help="Show detailed week-by-week breakdown with per-project drill-down" | |
| ) | |
| parser.add_argument( | |
| "--weeks", type=int, default=None, | |
| help=f"Number of weeks to analyze (default: {WEEKS_BACK})" | |
| ) | |
| args = parser.parse_args() | |
| if args.weeks is not None: | |
| global CUTOFF | |
| CUTOFF = datetime.now(timezone.utc) - timedelta(weeks=args.weeks) | |
| all_msgs, total_synthetic = load_all_messages() | |
| if not all_msgs: | |
| print("\nNo messages found in the analysis window.") | |
| return | |
| if args.weekly: | |
| report_weekly_deep_dive(all_msgs) | |
| return | |
| # Default: full report | |
| report_weekly_summary(all_msgs) | |
| report_daily_cache_trend(all_msgs) | |
| report_top_conversations(all_msgs) | |
| report_top_projects(all_msgs) | |
| report_cache_anomalies(all_msgs) | |
| report_version_comparison_simple(all_msgs) | |
| report_context_overflow(total_synthetic) | |
| report_time_of_day(all_msgs) | |
| # Export JSON | |
| script_dir = os.path.dirname(os.path.abspath(__file__)) | |
| json_path = os.path.join(script_dir, "claude-usage-report.json") | |
| export_json(all_msgs, total_synthetic, json_path) | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment