da23d22519
Lines the plan windows' official percentages (HA sensors from the homelab's claude-usage-notifier) up against the account's archived token spend, priced at list price, and fits dollars-per-1% for the 5h and week windows plus a Fable multiplier. docs/usage-calibration.md records the method and the 2026-10 calibration (1% week ≈ $22, 1% 5h ≈ $5.6, Fable ×1.75, no separate Fable bucket) and notes that pricing.py overstates the newest models. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
263 lines
12 KiB
Python
Executable File
263 lines
12 KiB
Python
Executable File
#!/usr/bin/env -S uv run --script
|
||
# /// script
|
||
# requires-python = ">=3.11"
|
||
# dependencies = ["pandas>=2.2", "numpy>=1.26", "scipy>=1.13"]
|
||
# ///
|
||
"""What is 1 % of a Claude subscription window worth, in tokens and API dollars?
|
||
|
||
Calibrates the official `/usage` percentages (rolling 5-hour window, 7-day
|
||
window, and the model-specific weekly bucket when it exists) against the tokens
|
||
the same account actually spent, as recorded in the ai-agent transcript archive.
|
||
|
||
Two inputs, both already on the box:
|
||
|
||
* **Usage %** — Home Assistant's long-term statistics (hourly max/min) for the
|
||
sensors `claude-usage-notifier` pushes: `<prefix>_5h`, `<prefix>_week`,
|
||
`<prefix>_fable`. Read straight from the recorder SQLite db (read-only).
|
||
* **Tokens** — every assistant message of the account's conversations, listed
|
||
through the ai-agent API (which resolves each conversation's account) and
|
||
parsed from the archive, subagent transcripts included, de-duplicated on
|
||
(message id, request id) like the backend does.
|
||
|
||
Method: bin both series (default 1 h), take the increase of each window's %
|
||
per bin (bins containing a reset are dropped, and so are 5h bins above 100 %,
|
||
i.e. overage), and fit `Δ% = k · cost + c` by non-negative least squares, where
|
||
`cost` is the bin's API-equivalent spend at today's list prices. A Fable
|
||
multiplier `f` (Fable dollars count `f`× an Opus dollar) is picked by grid
|
||
search on the combined fit of the 5h and week windows. `c` absorbs usage the
|
||
archive cannot see (claude.ai chat, desktop, an unpaired machine).
|
||
|
||
uv run scripts/usage-calibration.py # personal, last 9 days
|
||
uv run scripts/usage-calibration.py --since 2026-09-13 --json out.json
|
||
uv run scripts/usage-calibration.py --account work --prefix sensor.claude_work_usage
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import sqlite3
|
||
import sys
|
||
import urllib.request
|
||
from concurrent.futures import ProcessPoolExecutor
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
from scipy.optimize import nnls
|
||
|
||
HOME = pathlib.Path.home()
|
||
|
||
# USD / MTok: (input, output, cache read). Cache writes are 1.25x (5 min TTL) /
|
||
# 2x (1 h TTL) the input price. List prices as of 2026-10 — note the cache-read
|
||
# cuts on the newest models (Opus 5.5: 0.05x input, Fable 5.1: 0.025x), which
|
||
# backend/conversations/pricing.py does not model yet.
|
||
PRICES = {
|
||
"claude-fable-5-1": (10.0, 50.0, 0.25),
|
||
"claude-fable-5": (10.0, 50.0, 1.00),
|
||
"claude-opus-5-5": (4.0, 20.0, 0.20),
|
||
"claude-opus-5": (5.0, 25.0, 0.50),
|
||
"claude-opus-4-8": (5.0, 25.0, 0.50),
|
||
"claude-sonnet-5-5": (2.0, 10.0, 0.20),
|
||
"claude-sonnet-5": (2.0, 10.0, 0.20),
|
||
"claude-sonnet-4-6": (3.0, 15.0, 0.30),
|
||
"claude-haiku-5-5": (0.10, 0.50, 0.01),
|
||
"claude-haiku-4-5-20251001": (1.0, 5.0, 0.10),
|
||
}
|
||
WEEKS_PER_MONTH = 365.25 / 12 / 7
|
||
|
||
|
||
def family(model: str) -> str:
|
||
return next((f for f in ("fable", "opus", "sonnet", "haiku") if f in model), "other")
|
||
|
||
|
||
# ── tokens ────────────────────────────────────────────────────────────────────
|
||
|
||
def list_sessions(api: str, account: str) -> list[str]:
|
||
url = f"{api}/api/conversations?limit=0&archived=all&account={account}"
|
||
with urllib.request.urlopen(url, timeout=120) as r:
|
||
return [c["id"] for c in json.load(r)["conversations"]]
|
||
|
||
|
||
def _parse(path: str) -> list[dict]:
|
||
out: dict[tuple, dict] = {}
|
||
with open(path, errors="replace") as fh:
|
||
for line in fh:
|
||
if '"usage"' not in line:
|
||
continue
|
||
try:
|
||
d = json.loads(line)
|
||
except ValueError:
|
||
continue
|
||
m = d.get("message") or {}
|
||
u = m.get("usage")
|
||
if d.get("type") != "assistant" or not isinstance(u, dict):
|
||
continue
|
||
key = (m.get("id"), d.get("requestId"))
|
||
if key in out:
|
||
continue
|
||
cc = u.get("cache_creation") or {}
|
||
c5 = cc.get("ephemeral_5m_input_tokens") or 0
|
||
c1 = cc.get("ephemeral_1h_input_tokens") or 0
|
||
if not c5 + c1:
|
||
c5 = u.get("cache_creation_input_tokens") or 0
|
||
out[key] = {"key": f"{key[0]}|{key[1]}", "ts": d.get("timestamp"),
|
||
"model": m.get("model") or "", "inp": u.get("input_tokens") or 0,
|
||
"cw5": c5, "cw1": c1, "cr": u.get("cache_read_input_tokens") or 0,
|
||
"out": u.get("output_tokens") or 0}
|
||
return list(out.values())
|
||
|
||
|
||
def load_messages(api: str, archive: pathlib.Path, account: str) -> pd.DataFrame:
|
||
files = []
|
||
for cid in list_sessions(api, account):
|
||
p = archive / cid
|
||
if p.exists():
|
||
files.append(str(p))
|
||
files += [str(f) for f in p.with_suffix("").glob("subagents/*.jsonl")]
|
||
rows: dict[str, dict] = {}
|
||
with ProcessPoolExecutor(min(12, os.cpu_count() or 4)) as ex:
|
||
for res in ex.map(_parse, files, chunksize=8):
|
||
for r in res: # a fork/resume repeats earlier messages: keep the first
|
||
if r["key"] not in rows or r["ts"] < rows[r["key"]]["ts"]:
|
||
rows[r["key"]] = r
|
||
df = pd.DataFrame(rows.values())
|
||
df["ts"] = pd.to_datetime(df.ts, utc=True, format="ISO8601")
|
||
df = df[df.model.isin(PRICES)].set_index("ts").sort_index()
|
||
df["fam"] = df.model.map(family)
|
||
p = df.model.map(PRICES)
|
||
pi, po, pc = (p.map(lambda x, i=i: x[i]) / 1e6 for i in range(3))
|
||
df["c_in"] = df.inp * pi
|
||
df["c_cw"] = (df.cw5 * 1.25 + df.cw1 * 2.0) * pi
|
||
df["c_cr"] = df.cr * pc
|
||
df["c_out"] = df.out * po
|
||
df["cost"] = df.c_in + df.c_cw + df.c_cr + df.c_out
|
||
df["tok"] = df.inp + df.cw5 + df.cw1 + df.cr + df.out
|
||
return df
|
||
|
||
|
||
# ── usage % ───────────────────────────────────────────────────────────────────
|
||
|
||
def load_ha(db: pathlib.Path, prefix: str) -> dict[str, pd.DataFrame]:
|
||
con = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
|
||
out = {}
|
||
for win in ("5h", "week", "fable"):
|
||
rows = con.execute(
|
||
"SELECT s.start_ts, s.min, s.max FROM statistics s "
|
||
"JOIN statistics_meta m ON s.metadata_id = m.id "
|
||
"WHERE m.statistic_id = ? ORDER BY s.start_ts", (f"{prefix}_{win}",)).fetchall()
|
||
df = pd.DataFrame(rows, columns=["start", "min", "max"])
|
||
df["start"] = pd.to_datetime(df.start, unit="s", utc=True)
|
||
out[win] = df.set_index("start")
|
||
return out
|
||
|
||
|
||
def increments(stats: pd.DataFrame, freq: str) -> pd.DataFrame:
|
||
b = stats.resample(freq).agg({"max": "last", "min": "min"})
|
||
b["prev"] = b["max"].shift()
|
||
b["d"] = b["max"] - b["prev"]
|
||
reset = (b["min"] < b["prev"] - 0.5) | (b["d"] < 0)
|
||
return b[~reset & b.prev.notna() & (b["max"] <= 100)]
|
||
|
||
|
||
# ── fit ───────────────────────────────────────────────────────────────────────
|
||
|
||
def fit(inc: pd.DataFrame, cost: pd.Series) -> tuple[float, float, float, int]:
|
||
"""Δ% = k·cost + c → (dollars per 1 %, intercept %/bin, R², n)."""
|
||
x = inc.join(cost.rename("c"), how="left").fillna({"c": 0})
|
||
a = np.column_stack([x.c.values, np.ones(len(x))])
|
||
y = x.d.values
|
||
(k, c), _ = nnls(a, y)
|
||
pred = a @ np.array([k, c])
|
||
r2 = 1 - ((y - pred) ** 2).sum() / max(((y - y.mean()) ** 2).sum(), 1e-12)
|
||
return (1 / k if k > 0 else float("inf")), c, r2, len(x)
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
|
||
ap.add_argument("--api", default="http://127.0.0.1:8096")
|
||
ap.add_argument("--archive", type=pathlib.Path,
|
||
default=HOME / "homelab/services/ai-agent/data/transcripts")
|
||
ap.add_argument("--ha-db", type=pathlib.Path,
|
||
default=HOME / "homelab/services/home-assistant/config/home-assistant_v2.db")
|
||
ap.add_argument("--prefix", default="sensor.claude_usage")
|
||
ap.add_argument("--account", default="personal")
|
||
ap.add_argument("--since", default=None, help="calibration start (default: 9 days ago)")
|
||
ap.add_argument("--until", default=None)
|
||
ap.add_argument("--freq", default="1h")
|
||
ap.add_argument("--eur", type=float, default=0.92, help="USD→EUR (ai-agent default)")
|
||
ap.add_argument("--plan-eur", type=float, default=200.0, help="monthly plan price, EUR")
|
||
ap.add_argument("--json", type=pathlib.Path)
|
||
args = ap.parse_args()
|
||
|
||
now = pd.Timestamp.now(tz="UTC")
|
||
since = pd.Timestamp(args.since, tz="UTC") if args.since else now - pd.Timedelta(days=9)
|
||
until = pd.Timestamp(args.until, tz="UTC") if args.until else now
|
||
|
||
msgs = load_messages(args.api, args.archive, args.account)
|
||
ha = load_ha(args.ha_db, args.prefix)
|
||
win = msgs.loc[since:until]
|
||
is_fable = (msgs.fam == "fable").to_numpy()
|
||
inc = {w: increments(ha[w], args.freq).loc[since:until] for w in ("5h", "week")}
|
||
|
||
# Fable multiplier: best combined R² over the 5h and week windows.
|
||
grid = np.round(np.arange(0.5, 3.01, 0.05), 2)
|
||
scores = []
|
||
for f in grid:
|
||
cost = (msgs.cost * np.where(is_fable, f, 1.0)).resample(args.freq).sum()
|
||
scores.append(sum(fit(inc[w], cost)[2] for w in inc))
|
||
fmult = float(grid[int(np.argmax(scores))])
|
||
weighted = (msgs.cost * np.where(is_fable, fmult, 1.0)).resample(args.freq).sum()
|
||
plain = msgs.cost.resample(args.freq).sum()
|
||
|
||
res: dict = {"account": args.account, "since": str(since), "until": str(until),
|
||
"freq": args.freq, "fableMultiplier": fmult, "windows": {}}
|
||
for w in inc:
|
||
usd, icpt, r2, n = fit(inc[w], weighted)
|
||
usd0, _, r20, _ = fit(inc[w], plain)
|
||
res["windows"][w] = {"usdPerPct": usd, "usdPerPctFable": usd / fmult,
|
||
"r2": r2, "bins": n, "interceptPctPerBin": icpt,
|
||
"usdPerPctUnweighted": usd0, "r2Unweighted": r20}
|
||
fable = ha["fable"].loc[since:until]
|
||
res["fableBucketSeen"] = bool(len(fable) and fable["max"].max() > 0)
|
||
|
||
# Token equivalents at the window's own mix, per family.
|
||
mix = {}
|
||
for fam, g in win.groupby("fam"):
|
||
if g.cost.sum() < 1:
|
||
continue
|
||
t = g[["inp", "cw5", "cw1", "cr", "out"]].sum()
|
||
mix[fam] = {"usd": g.cost.sum(), "tokens": int(g.tok.sum()),
|
||
"tokPerUsd": g.tok.sum() / g.cost.sum(),
|
||
"share": {k: t[k] / t.sum() for k in t.index},
|
||
"costShare": {k: g[k].sum() / g.cost.sum()
|
||
for k in ("c_in", "c_cw", "c_cr", "c_out")}}
|
||
res["mix"] = mix
|
||
|
||
wk = res["windows"]["week"]["usdPerPct"]
|
||
month_usd = 100 * wk * WEEKS_PER_MONTH
|
||
plan_usd = args.plan_eur / args.eur
|
||
res["month"] = {"opusUsd": month_usd, "fableUsd": month_usd / fmult,
|
||
"planUsd": plan_usd, "leverage": month_usd / plan_usd}
|
||
|
||
print(f"account={args.account} {since:%Y-%m-%d %H:%M} → {until:%Y-%m-%d %H:%M} "
|
||
f"bins={args.freq} messages={len(win)} spend=${win.cost.sum():,.0f}")
|
||
print(f"Fable multiplier (Fable $ vs Opus $): ×{fmult}")
|
||
for w, r in res["windows"].items():
|
||
print(f" 1 % of {w:>4}: ${r['usdPerPct']:7.2f} Opus-$ (${r['usdPerPctFable']:.2f} of Fable)"
|
||
f" R²={r['r2']:.3f} [unweighted ${r['usdPerPctUnweighted']:.2f}, R²={r['r2Unweighted']:.3f}]"
|
||
f" n={r['bins']} untracked≈{r['interceptPctPerBin']:.3f}%/bin")
|
||
print(f" model-specific weekly bucket seen: {res['fableBucketSeen']}")
|
||
for fam, m in mix.items():
|
||
per = {w: r["usdPerPct"] / (fmult if fam == "fable" else 1) for w, r in res["windows"].items()}
|
||
print(f" {fam:6s}: {m['tokPerUsd'] / 1e6:5.2f} Mtok/$ → 1 % 5h ≈ {per['5h'] * m['tokPerUsd'] / 1e6:6.1f} Mtok,"
|
||
f" 1 % week ≈ {per['week'] * m['tokPerUsd'] / 1e6:6.1f} Mtok")
|
||
print(f"100 % of every week for a month: ${month_usd:,.0f} of Opus API (${month_usd / fmult:,.0f} of Fable)"
|
||
f" vs plan ${plan_usd:,.0f} → ×{res['month']['leverage']:.0f}")
|
||
if args.json:
|
||
args.json.write_text(json.dumps(res, indent=2, default=float))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|