Claude Code status line with spend tracking

statusline.py

#!/usr/bin/env python3
"""Claude Code status line: token usage, accurate spend, and rate limits.

Shows cwd/git branch, model, token counts, current session cost, context
window usage, rate limits, and today's/month's actual spend - computed by
summing the real per-API-call `usage` block recorded in every assistant
message across all local project transcripts, priced per-model.

Maintains a per-file cache (keyed on mtime+size) under
~/.claude/daily-cost/usage-cache.json so repeated invocations don't re-parse
unchanged transcripts.

Setup:
    1. Save this file somewhere stable, e.g. ~/.config/claude/statusline.py.
    2. Add to ~/.claude/settings.json:

        {
          "statusLine": {
            "type": "command",
            "command": "python3 ~/.config/claude/statusline.py"
          }
        }

    3. Restart Claude Code (or start a new session) to pick up the change.

Adjust DAILY_BUDGET / MONTHLY_BUDGET below to your own plan's spend limits -
they're only used for the %-of-budget coloring.
"""
import json
import os
import subprocess
import sys
from datetime import datetime, date
from glob import glob

HOME = os.path.expanduser("~")
PROJECTS_DIR = os.path.join(HOME, ".claude", "projects")
CACHE_FILE = os.path.join(HOME, ".claude", "daily-cost", "usage-cache.json")

# Daily/monthly spend budgets (used for % display in status line)
DAILY_BUDGET = 115.00
MONTHLY_BUDGET = 2000.00

RED = "\033[31m"
YELLOW = "\033[33m"
RESET = "\033[0m"

# $ per MTok: (input, output, cache_write_5m, cache_write_1h, cache_read)
PRICING = {
    "claude-opus-4-8": (5.00, 25.00, 6.25, 10.00, 0.50),
    "claude-opus-4-7": (5.00, 25.00, 6.25, 10.00, 0.50),
    "claude-opus-4-6": (5.00, 25.00, 6.25, 10.00, 0.50),
    "claude-opus-4-5": (5.00, 25.00, 6.25, 10.00, 0.50),
    "claude-opus-4-1": (5.00, 25.00, 6.25, 10.00, 0.50),
    "claude-opus-4-0": (5.00, 25.00, 6.25, 10.00, 0.50),
    "claude-sonnet-4-6": (3.00, 15.00, 3.75, 6.00, 0.30),
    "claude-sonnet-4-5": (3.00, 15.00, 3.75, 6.00, 0.30),
    "claude-sonnet-4-0": (3.00, 15.00, 3.75, 6.00, 0.30),
    "claude-haiku-4-5": (1.00, 5.00, 1.25, 2.00, 0.10),
    "claude-fable-5": (10.00, 50.00, 12.50, 20.00, 1.00),
    "claude-mythos-5": (10.00, 50.00, 12.50, 20.00, 1.00),
}
# Sonnet 5 launched with introductory pricing through this date.
SONNET_5_INTRO_END = date(2026, 8, 31)


def price_for(model, as_of):
    if model.startswith("claude-sonnet-5"):
        if as_of <= SONNET_5_INTRO_END:
            return (2.00, 10.00, 2.50, 4.00, 0.20)
        return (3.00, 15.00, 3.75, 6.00, 0.30)
    if model in PRICING:
        return PRICING[model]
    for k, v in PRICING.items():
        if model.startswith(k):
            return v
    return None


def usage_cost(usage, model, as_of):
    p = price_for(model, as_of)
    if not p:
        return 0.0
    p_in, p_out, p_cw5, p_cw1h, p_cr = p
    cc = usage.get("cache_creation") or {}
    e5 = cc.get("ephemeral_5m_input_tokens", 0) or 0
    e1 = cc.get("ephemeral_1h_input_tokens", 0) or 0
    tot_cc = usage.get("cache_creation_input_tokens", 0) or 0
    if not e5 and not e1 and tot_cc:
        e5 = tot_cc  # older transcripts lack the ephemeral_5m/1h breakdown
    return (
        (usage.get("input_tokens", 0) or 0) * p_in
        + (usage.get("output_tokens", 0) or 0) * p_out
        + e5 * p_cw5
        + e1 * p_cw1h
        + (usage.get("cache_read_input_tokens", 0) or 0) * p_cr
    ) / 1_000_000


def parse_transcript(fp):
    """Yield (timestamp, model, usage) for real, deduped assistant messages."""
    seen = set()
    try:
        with open(fp, errors="ignore") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    d = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if d.get("type") != "assistant":
                    continue
                msg = d.get("message", {})
                model = msg.get("model")
                if not model or model == "<synthetic>":
                    continue
                mid = msg.get("id")
                ts = d.get("timestamp")
                if not mid or not ts:
                    continue
                # Same message.id can appear multiple times with identical
                # usage (streamed/continuation entries) - count it once.
                if mid in seen:
                    continue
                seen.add(mid)
                yield ts, model, msg.get("usage", {}) or {}
    except (IOError, OSError):
        return


def load_cache():
    try:
        with open(CACHE_FILE) as fh:
            return json.load(fh)
    except (IOError, OSError, json.JSONDecodeError):
        return {"files": {}}


def save_cache(cache):
    os.makedirs(os.path.dirname(CACHE_FILE), exist_ok=True)
    tmp = CACHE_FILE + ".tmp"
    with open(tmp, "w") as fh:
        json.dump(cache, fh)
    os.replace(tmp, CACHE_FILE)


def file_key(fp):
    st = os.stat(fp)
    return f"{st.st_mtime_ns}:{st.st_size}"


def refresh_cache():
    cache = load_cache()
    files_cache = cache.setdefault("files", {})
    # recursive: subagent (Task/Agent tool) transcripts live one level deeper,
    # at <session-dir>/subagents/agent-*.jsonl - a plain */*.jsonl glob misses
    # them entirely and silently undercounts every subagent call's usage.
    files = glob(os.path.join(PROJECTS_DIR, "**", "*.jsonl"), recursive=True)
    seen_files = set()
    for fp in files:
        seen_files.add(fp)
        try:
            key = file_key(fp)
        except OSError:
            continue
        entry = files_cache.get(fp)
        if entry and entry.get("key") == key:
            continue  # unchanged since last run - reuse cached per-day sums
        by_day = {}
        for ts, model, usage in parse_transcript(fp):
            try:
                dt = datetime.fromisoformat(ts.replace("Z", "+00:00")).astimezone()
            except ValueError:
                continue
            day = dt.date().isoformat()
            cost = usage_cost(usage, model, dt.date())
            by_day[day] = by_day.get(day, 0.0) + cost
        files_cache[fp] = {"key": key, "by_day": by_day}
    for fp in list(files_cache.keys()):
        if fp not in seen_files:
            del files_cache[fp]
    save_cache(cache)
    return cache


def aggregate(cache, day=None, month=None):
    total = 0.0
    for entry in cache["files"].values():
        for d, cost in entry["by_day"].items():
            if day and d != day:
                continue
            if month and d[:7] != month:
                continue
            total += cost
    return total


def session_cost(transcript_path):
    if not transcript_path or not os.path.exists(transcript_path):
        return 0.0
    total = 0.0
    for ts, model, usage in parse_transcript(transcript_path):
        try:
            dt = datetime.fromisoformat(ts.replace("Z", "+00:00")).astimezone()
        except ValueError:
            continue
        total += usage_cost(usage, model, dt.date())
    return total


# --- status line formatting -------------------------------------------------

def fmt_k(n):
    if n is None:
        return "0"
    if n >= 1000:
        return f"{n / 1000:.1f}K"
    return str(n)


def git_branch(cwd):
    try:
        out = subprocess.run(
            ["git", "-C", cwd, "rev-parse", "--abbrev-ref", "HEAD"],
            capture_output=True, text=True, timeout=2,
        )
        branch = out.stdout.strip()
        return branch if out.returncode == 0 and branch else None
    except (OSError, subprocess.SubprocessError):
        return None


def budget_part(label, amount, budget):
    pct = amount / budget * 100
    text = f"{label}:${amount:.2f} ({pct:.1f}%)"
    if pct >= 90:
        return f"{RED}{text}{RESET}"
    if pct >= 75:
        return f"{YELLOW}{text}{RESET}"
    return text


def build_statusline(hook_input):
    parts = []

    cwd_full = hook_input.get("cwd") or os.environ.get("PWD", "")
    cwd_name = os.path.basename(cwd_full.rstrip("/")) if cwd_full else ""
    branch = git_branch(cwd_full) if cwd_full else None
    if cwd_name:
        parts.append(f"{cwd_name}({branch})" if branch else cwd_name)

    model = (hook_input.get("model") or {}).get("display_name")
    if model:
        parts.append(model)

    ctx = hook_input.get("context_window") or {}
    total_input = ctx.get("total_input_tokens")
    cur_output = (ctx.get("current_usage") or {}).get("output_tokens")
    transcript_path = hook_input.get("transcript_path")
    sess_cost = session_cost(transcript_path)

    if total_input:
        cost_str = f" sess:${sess_cost:.2f}" if sess_cost else ""
        parts.append(f"in:{fmt_k(total_input)} out:{fmt_k(cur_output)}{cost_str}")

    used_pct = ctx.get("used_percentage")
    if used_pct is not None:
        used_int = round(used_pct)
        text = f"ctx:{used_int}%"
        if used_int >= 80:
            parts.append(f"{RED}{text}{RESET}")
        elif used_int >= 50:
            parts.append(f"{YELLOW}{text}{RESET}")
        else:
            parts.append(text)

    rate_limits = hook_input.get("rate_limits") or {}
    five_hour = (rate_limits.get("five_hour") or {}).get("used_percentage")
    seven_day = (rate_limits.get("seven_day") or {}).get("used_percentage")
    rate_parts = []
    if five_hour is not None:
        rate_parts.append(f"5h:{round(five_hour)}%")
    if seven_day is not None:
        rate_parts.append(f"7d:{round(seven_day)}%")
    if rate_parts:
        parts.append(f"limits: {' '.join(rate_parts)}")

    cache = refresh_cache()
    today = datetime.now().astimezone().date()
    today_cost = aggregate(cache, day=today.isoformat())
    month_cost = aggregate(cache, month=today.isoformat()[:7])
    parts.append(budget_part("today", today_cost, DAILY_BUDGET))
    parts.append(budget_part("mtd", month_cost, MONTHLY_BUDGET))

    return " · ".join(parts)


def main():
    try:
        hook_input = json.load(sys.stdin)
    except (json.JSONDecodeError, ValueError):
        hook_input = {}
    print(build_statusline(hook_input), end="")


if __name__ == "__main__":
    main()