from __future__ import annotations
import argparse
import os
import duckdb
ROOT = os.path.expanduser("~/.claude/projects")
PATTERN = os.path.join(ROOT, "-Users-cody-netsky*/*.jsonl")
RATES = {
"claude-opus-4-6": {
"input": 5.00,
"output": 25.00,
"cache_read": 0.50,
"cw_5m": 6.25,
"cw_1h": 10.00,
},
"claude-haiku-4-5-20251001": {
"input": 1.00,
"output": 5.00,
"cache_read": 0.10,
"cw_5m": 1.25,
"cw_1h": 2.00,
},
}
PRICED_MODELS_SQL = "', '".join(RATES.keys())
PRICE_CASE = """
CASE model
WHEN 'claude-opus-4-6' THEN
(input_tokens / 1e6) * 5.00
+ (output_tokens / 1e6) * 25.00
+ (cache_read / 1e6) * 0.50
+ (cw_5m / 1e6) * 6.25
+ (cw_1h / 1e6) * 10.00
WHEN 'claude-haiku-4-5-20251001' THEN
(input_tokens / 1e6) * 1.00
+ (output_tokens / 1e6) * 5.00
+ (cache_read / 1e6) * 0.10
+ (cw_5m / 1e6) * 1.25
+ (cw_1h / 1e6) * 2.00
ELSE 0.00
END
"""
def open_db():
con = duckdb.connect()
con.execute(f"""
CREATE OR REPLACE TABLE raw AS
SELECT *, filename FROM read_json('{PATTERN}',
union_by_name=true, ignore_errors=true,
format='newline_delimited', filename=true);
""")
con.execute(f"""
CREATE OR REPLACE TABLE asst AS
SELECT
sessionId,
timestamp::TIMESTAMPTZ AS ts,
message.id AS msg_id,
message.model AS model,
COALESCE(CAST(message.usage.input_tokens AS BIGINT), 0) AS input_tokens,
COALESCE(CAST(message.usage.output_tokens AS BIGINT), 0) AS output_tokens,
COALESCE(CAST(message.usage.cache_read_input_tokens AS BIGINT), 0) AS cache_read,
COALESCE(CAST(message.usage.cache_creation.ephemeral_5m_input_tokens AS BIGINT), 0) AS cw_5m_raw,
COALESCE(CAST(message.usage.cache_creation.ephemeral_1h_input_tokens AS BIGINT), 0) AS cw_1h_raw,
COALESCE(CAST(message.usage.cache_creation_input_tokens AS BIGINT), 0) AS cw_total,
message.usage.service_tier AS service_tier,
filename
FROM raw
WHERE type = 'assistant'
AND message IS NOT NULL
AND message.usage IS NOT NULL
AND message.model IN ('{PRICED_MODELS_SQL}')
AND (
COALESCE(CAST(message.usage.input_tokens AS BIGINT), 0)
+ COALESCE(CAST(message.usage.output_tokens AS BIGINT), 0)
+ COALESCE(CAST(message.usage.cache_read_input_tokens AS BIGINT), 0)
+ COALESCE(CAST(message.usage.cache_creation_input_tokens AS BIGINT), 0)
) > 0;
""")
con.execute("""
CREATE OR REPLACE TABLE deduped AS
WITH ranked AS (
SELECT *, ROW_NUMBER() OVER (PARTITION BY msg_id ORDER BY ts) AS rn FROM asst
)
SELECT
sessionId, ts, msg_id, model, input_tokens, output_tokens, cache_read,
CASE WHEN (cw_5m_raw + cw_1h_raw) = 0 AND cw_total > 0 THEN cw_total ELSE cw_5m_raw END AS cw_5m,
CASE WHEN (cw_5m_raw + cw_1h_raw) = 0 AND cw_total > 0 THEN 0 ELSE cw_1h_raw END AS cw_1h,
service_tier, filename
FROM ranked
WHERE rn = 1;
""")
con.execute(f"""
CREATE OR REPLACE TABLE priced AS
SELECT *, {PRICE_CASE} AS cost_usd FROM deduped;
""")
return con
def show(rows, headers):
if not rows:
print("(no rows)")
return
widths = [
max(len(str(h)), max(len(str(r[i])) for r in rows))
for i, h in enumerate(headers)
]
print(" ".join(str(h).ljust(widths[i]) for i, h in enumerate(headers)))
print(" ".join("-" * widths[i] for i in range(len(headers))))
for r in rows:
print(" ".join(str(c).ljust(widths[i]) for i, c in enumerate(r)))
def main():
ap = argparse.ArgumentParser()
ap.add_argument(
"--by",
choices=["day", "model", "project", "session", "peak", "all"],
default="all",
)
args = ap.parse_args()
con = open_db()
print("=== TOTAL (deduped, current Opus 4.6 rates, no fictional 1M surcharge) ===")
show(
con.execute("""
SELECT
printf('$%,.2f', SUM(cost_usd)) AS total_cost,
COUNT(*) AS distinct_messages,
COUNT(DISTINCT sessionId) AS sessions,
printf('%,.0f', SUM(input_tokens)) AS input,
printf('%,.0f', SUM(output_tokens)) AS output,
printf('%,.0f', SUM(cache_read)) AS cache_read,
printf('%,.0f', SUM(cw_5m)) AS cw_5m,
printf('%,.0f', SUM(cw_1h)) AS cw_1h
FROM priced;
""").fetchall(),
[
"total",
"messages",
"sessions",
"input",
"output",
"cache_read",
"cw_5m",
"cw_1h",
],
)
if args.by in ("day", "all"):
print("\n=== BY DAY (UTC) ===")
show(
con.execute("""
SELECT
strftime(ts AT TIME ZONE 'UTC', '%Y-%m-%d') AS day,
COUNT(*) AS calls,
printf('$%,.2f', SUM(cost_usd)) AS cost
FROM priced GROUP BY 1 ORDER BY 1;
""").fetchall(),
["day", "calls", "cost"],
)
if args.by in ("model", "all"):
print("\n=== BY MODEL ===")
show(
con.execute("""
SELECT model, COUNT(*) AS calls, printf('$%,.2f', SUM(cost_usd)) AS cost
FROM priced GROUP BY 1 ORDER BY SUM(cost_usd) DESC;
""").fetchall(),
["model", "calls", "cost"],
)
if args.by in ("project", "all"):
print("\n=== BY PROJECT (best-effort) ===")
show(
con.execute("""
SELECT
regexp_extract(filename, '/-Users-cody-netsky([^/]*)', 1) AS project_suffix,
COUNT(*) AS calls,
printf('$%,.2f', SUM(cost_usd)) AS cost
FROM priced GROUP BY 1 ORDER BY SUM(cost_usd) DESC LIMIT 20;
""").fetchall(),
["project_suffix (empty = netsky root)", "calls", "cost"],
)
if args.by in ("session", "all"):
print("\n=== TOP 10 SESSIONS BY COST ===")
show(
con.execute("""
WITH s AS (
SELECT sessionId, MIN(ts) AS started, MAX(ts) AS ended,
COUNT(*) AS calls, SUM(cost_usd) AS cost,
SUM(input_tokens + cache_read + cw_5m + cw_1h) AS prompt_t,
SUM(output_tokens) AS output_t,
any_value(filename) AS filename
FROM priced GROUP BY sessionId
)
SELECT
substr(CAST(sessionId AS VARCHAR), 1, 8) AS session,
strftime(started AT TIME ZONE 'UTC', '%Y-%m-%d %H:%M') AS started_utc,
CASE WHEN epoch(ended - started) >= 3600
THEN printf('%.1fh', epoch(ended - started) / 3600.0)
ELSE printf('%dm', CAST(epoch(ended - started) / 60.0 AS INT))
END AS dur,
calls,
printf('$%,.2f', cost) AS cost,
printf('%.1fM', prompt_t / 1e6) AS prompt,
printf('%.0fk', output_t / 1e3) AS output
FROM s ORDER BY cost DESC LIMIT 10;
""").fetchall(),
["session", "started", "dur", "calls", "cost", "prompt", "output"],
)
if args.by in ("peak", "all"):
print("\n=== PEAK 10-MIN BURN ===")
show(
con.execute("""
WITH t AS (SELECT ts, cost_usd FROM priced)
SELECT
strftime(p.ts AT TIME ZONE 'UTC', '%Y-%m-%d %H:%M:%SZ') AS window_start,
COUNT(*) AS calls_in_window,
printf('$%,.2f', SUM(c.cost_usd)) AS cost_in_window
FROM t p JOIN t c
ON c.ts >= p.ts AND c.ts < p.ts + INTERVAL 10 MINUTES
GROUP BY p.ts
ORDER BY SUM(c.cost_usd) DESC
LIMIT 1;
""").fetchall(),
["window_start", "calls", "cost"],
)
if __name__ == "__main__":
main()