Gabriel Vidal
4bedccc005
backend/usage/ polls each account's rate-limits.json (read-only mount,
USAGE_STATE_DIRS) and keeps every change (+ a 30 min heartbeat) in
usage_limits. A daily fit over 14 days of hourly rises vs the runs'
meterBins gives plan units per 1 % of the 5h / week window; only trusted
fits (n>=24, R2>=0.5, within x3 of the seed) replace the rate, and a >20 %
shift pushes a notification. Cards carry planPct, --final notifications
get ' · $X · ≈Y% wk', and /api/usage/{limits,history,calibration,
calibrate,import} serve the Plan usage page (owner only).
scripts/usage-backfill-ha.py imports the HA hourly statistics since July.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
306 lines
12 KiB
Python
306 lines
12 KiB
Python
"""Calibrate the plan meter: how many plan units (pricing.plan_units — what the
|
||
subscription's limits count) make 1 % of the 5h and week windows.
|
||
|
||
Inputs, both already hourly: the official % series (``usage_limits``: live
|
||
rate-limits.json points + the HA backfill) and every run's ``meterBins``
|
||
(epoch hour → [list $, plan units], from the parser). Per hour, the rise of a
|
||
window's % is the usage spent in it; hours that contain a reset, sit above
|
||
100 % (overage) or have no fresh reading on either side are dropped. The fit is
|
||
``Δ% = units / k + c`` by least squares — ``k`` = units per 1 %, ``c`` ≥ 0 the
|
||
usage the archive can't see (claude.ai, the desktop app, an unpaired machine).
|
||
|
||
A fit only replaces the current rate when it is trustworthy (enough hours, a
|
||
good R²), and a jump of more than ``SHIFT`` against the previous one pushes a
|
||
notification: that is what a model launch or a limits change looks like.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
import threading
|
||
import time
|
||
|
||
HOUR = 3600
|
||
# Seed rates (2026-10-12 fit, docs/usage-calibration.md): window → (plan units
|
||
# per 1 %, list $ per 1 %). Used until the first trustworthy fit lands.
|
||
DEFAULTS = {
|
||
"personal": {"week": (45.3, 22.2), "h5": (11.5, 5.6)},
|
||
"work": {"week": (22.3, 11.5), "h5": (3.5, 1.9)},
|
||
}
|
||
WINDOWS = {"week": "seven_day", "h5": "five_hour"}
|
||
FIT_DAYS = 14
|
||
MIN_N = 24 # hours with spend needed for a fit to count
|
||
MIN_R2 = 0.5 # a fit below this never replaces the current rate
|
||
SHIFT = 0.2 # |new/old - 1| above this (on the week) = a rate shift
|
||
FRESH_S = 2 * HOUR # a level reading older than this at an hour's end = unknown
|
||
|
||
_meter_cache: dict = {}
|
||
_meter_lock = threading.Lock()
|
||
|
||
|
||
def meters(deps, account: str) -> dict[int, list[float]]:
|
||
"""``{epoch hour: [list $, plan units]}`` summed over every run of
|
||
``account`` (subagents included; a session resumed into several files
|
||
counted once). Memoized on the store/meta versions."""
|
||
from conversations.cards import _summaries_with_account
|
||
key = (deps.store.summaries_version, deps.meta_store.version, account)
|
||
with _meter_lock:
|
||
hit = _meter_cache.get(account)
|
||
if hit and hit[0] == key:
|
||
return hit[1]
|
||
best: dict[str, tuple] = {}
|
||
rows = []
|
||
for path, s, a in _summaries_with_account(deps):
|
||
if a != account:
|
||
continue
|
||
sid = s.get("sessionId")
|
||
if s.get("isSidechain") or not sid:
|
||
rows.append(s)
|
||
continue
|
||
rank = (s.get("endedAt") or "", s.get("messages") or 0, path)
|
||
if sid not in best or rank > best[sid][0]:
|
||
best[sid] = (rank, s)
|
||
rows += [s for _, s in best.values()]
|
||
out: dict[int, list[float]] = {}
|
||
for s in rows:
|
||
for h, v in (s.get("meterBins") or {}).items():
|
||
b = out.setdefault(int(h), [0.0, 0.0])
|
||
b[0] += v[0]
|
||
b[1] += v[1]
|
||
with _meter_lock:
|
||
_meter_cache[account] = (key, out)
|
||
return out
|
||
|
||
|
||
def hourly_increments(points: list[tuple], h0: int, h1: int) -> dict[int, float]:
|
||
"""``{hour: Δ%}`` for every hour in [h0, h1) whose rise is measurable.
|
||
|
||
``points`` are (t, pct, resets_at) sorted by t. The level at an hour's end
|
||
is the last reading before it; the hour counts when both its end and its
|
||
start have a reading fresher than ``FRESH_S``, the window did not reset in
|
||
between (``resets_at`` moved or the level dropped) and the level is ≤ 100."""
|
||
out: dict[int, float] = {}
|
||
if not points:
|
||
return out
|
||
i = 0
|
||
prev = None # (t, pct, resets_at) at the previous hour's end
|
||
for h in range(h0, h1 + 1):
|
||
end = (h + 1) * HOUR
|
||
while i < len(points) and points[i][0] < end:
|
||
i += 1
|
||
cur = points[i - 1] if i else None
|
||
if cur is not None and end - cur[0] > FRESH_S:
|
||
cur = None
|
||
if cur is not None and prev is not None and h < h1:
|
||
reset = (cur[1] < prev[1] - 0.5
|
||
or (cur[2] and prev[2] and abs(cur[2] - prev[2]) > 120))
|
||
if not reset and cur[1] <= 100:
|
||
out[h] = cur[1] - prev[1]
|
||
prev = cur
|
||
return out
|
||
|
||
|
||
def _lsq(xs: list[float], ys: list[float]) -> tuple[float, float, float] | None:
|
||
"""y = a·x + c, c ≥ 0 → (a, c, R²); None when x carries no signal."""
|
||
n = len(xs)
|
||
if n < 3:
|
||
return None
|
||
mx, my = sum(xs) / n, sum(ys) / n
|
||
sxx = sum((x - mx) ** 2 for x in xs)
|
||
if sxx <= 0:
|
||
return None
|
||
a = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / sxx
|
||
c = my - a * mx
|
||
if c < 0: # no negative "untracked" usage: refit through the origin
|
||
sx2 = sum(x * x for x in xs)
|
||
a, c = sum(x * y for x, y in zip(xs, ys)) / sx2, 0.0
|
||
if a <= 0:
|
||
return None
|
||
ss = sum((y - my) ** 2 for y in ys)
|
||
res = sum((y - a * x - c) ** 2 for x, y in zip(xs, ys))
|
||
return a, c, (1 - res / ss) if ss > 0 else 0.0
|
||
|
||
|
||
def fit(deps, account: str, end: float | None = None, days: int = FIT_DAYS) -> dict | None:
|
||
"""One calibration over the ``days`` before ``end``, or None when a window
|
||
has too little signal. Not stored — see ``run``."""
|
||
end = end or time.time()
|
||
h1 = int(end // HOUR)
|
||
h0 = h1 - days * 24
|
||
m = meters(deps, account)
|
||
series = deps.usage_store.series(account, h0 * HOUR - FRESH_S, end)
|
||
cal = {"account": account, "fittedAt": end, "days": days, "source": "fit"}
|
||
for name, win in WINDOWS.items():
|
||
inc = hourly_increments(series.get(win, []), h0, h1)
|
||
xs, ys, cost = [], [], 0.0
|
||
for h, d in inc.items():
|
||
c, u = m.get(h, (0.0, 0.0))
|
||
xs.append(u)
|
||
ys.append(d)
|
||
cost += c
|
||
units = sum(xs)
|
||
r = _lsq(xs, ys)
|
||
busy = sum(1 for x in xs if x > 0)
|
||
if r is None or busy < MIN_N:
|
||
return None
|
||
a, c, r2 = r
|
||
k = 1 / a
|
||
cal[name] = {"unitsPerPct": k, "usdPerPct": k * cost / units if units else None,
|
||
"r2": r2, "n": busy, "interceptPct": c}
|
||
cal["trusted"] = _trusted(cal)
|
||
return cal
|
||
|
||
|
||
def default_calibration(account: str) -> dict:
|
||
d = DEFAULTS.get(account) or DEFAULTS["personal"]
|
||
return {"account": account, "fittedAt": 0.0, "days": 0, "source": "default",
|
||
"trusted": True,
|
||
**{w: {"unitsPerPct": d[w][0], "usdPerPct": d[w][1], "r2": None, "n": 0,
|
||
"interceptPct": None} for w in WINDOWS}}
|
||
|
||
|
||
# A fit more than this far off the seed is implausible however good its R²:
|
||
# usage the archive can't see that happens in the *same* hours as archived work
|
||
# (Sep 2026: Orus work on the personal seat from an unpaired Mac) fits cleanly
|
||
# at a rate 10× off. A real rate change (a model launch) is tens of percent.
|
||
PLAUSIBLE = 3.0
|
||
|
||
|
||
def _trusted(cal: dict) -> bool:
|
||
seed = DEFAULTS.get(cal["account"]) or DEFAULTS["personal"]
|
||
return all((cal[w].get("r2") or 0) >= MIN_R2 and cal[w].get("n", 0) >= MIN_N
|
||
and 1 / PLAUSIBLE <= cal[w]["unitsPerPct"] / seed[w][0] <= PLAUSIBLE
|
||
for w in WINDOWS)
|
||
|
||
|
||
_current_cache: dict = {}
|
||
|
||
|
||
def current(deps, account: str) -> dict:
|
||
"""The rate in force: the latest trustworthy fit, else the seed."""
|
||
key = deps.usage_store.version
|
||
hit = _current_cache.get(account)
|
||
if hit and hit[0] == key:
|
||
return hit[1]
|
||
cal = next((c for c in reversed(deps.usage_store.calibrations(account))
|
||
if c.get("source") == "fit" and _trusted(c)), None)
|
||
cal = cal or default_calibration(account)
|
||
_current_cache[account] = (key, cal)
|
||
return cal
|
||
|
||
|
||
def plan_pct(deps, account: str, units: float | None) -> dict | None:
|
||
if units is None:
|
||
return None
|
||
cal = current(deps, account)
|
||
return {"week": units / cal["week"]["unitsPerPct"],
|
||
"h5": units / cal["h5"]["unitsPerPct"],
|
||
"account": account, "calibratedAt": cal["fittedAt"] or None}
|
||
|
||
|
||
def run(deps, account: str, notify=None) -> tuple[dict | None, bool]:
|
||
"""Fit now, store it, and report whether the week rate shifted."""
|
||
before = current(deps, account)
|
||
cal = fit(deps, account)
|
||
if cal is None:
|
||
return None, False
|
||
deps.usage_store.add_calibration(cal)
|
||
shifted = False
|
||
if _trusted(cal) and before.get("source") == "fit":
|
||
old, new = before["week"]["unitsPerPct"], cal["week"]["unitsPerPct"]
|
||
shifted = abs(new / old - 1) > SHIFT
|
||
if shifted and notify:
|
||
day = dt.datetime.fromtimestamp(before["fittedAt"]).strftime("%b %d")
|
||
notify(account, f"1 % of the {account} week was {old:.1f} plan units "
|
||
f"(fit of {day}), now {new:.1f} "
|
||
f"(≈ ${cal['week']['usdPerPct'] or 0:.1f} list) — limits or "
|
||
f"metering changed (a model launch?).")
|
||
return cal, shifted
|
||
|
||
|
||
def backfill(deps, account: str, step_days: int = 7) -> int:
|
||
"""Weekly fits over the whole recorded history (``$/% over time``)."""
|
||
first = deps.usage_store.first_t(account)
|
||
if not first:
|
||
return 0
|
||
n = 0
|
||
end = (first // 86400 + FIT_DAYS) * 86400
|
||
now = time.time()
|
||
while end < now:
|
||
cal = fit(deps, account, end=end)
|
||
if cal:
|
||
deps.usage_store.add_calibration(cal)
|
||
n += 1
|
||
end += step_days * 86400
|
||
return n
|
||
|
||
|
||
def history(deps, account: str, days: int) -> dict:
|
||
"""What the Plan usage panel draws: the raw % series, hourly meters with
|
||
the measurable rises, and per-day observed / predicted / untracked %."""
|
||
now = time.time()
|
||
h1 = int(now // HOUR)
|
||
h0 = h1 - days * 24
|
||
cal = current(deps, account)
|
||
m = meters(deps, account)
|
||
series = deps.usage_store.series(account, h0 * HOUR - FRESH_S)
|
||
inc = {n: hourly_increments(series.get(w, []), h0, h1) for n, w in WINDOWS.items()}
|
||
hourly, daily = [], {}
|
||
for h in range(h0, h1 + 1):
|
||
c, u = m.get(h, (0.0, 0.0))
|
||
dw, d5 = inc["week"].get(h), inc["h5"].get(h)
|
||
if c or u or dw or d5:
|
||
hourly.append({"t": h * HOUR, "cost": c, "units": u, "dWeek": dw, "d5h": d5})
|
||
day = dt.datetime.fromtimestamp(h * HOUR).strftime("%Y-%m-%d")
|
||
e = daily.setdefault(day, {"day": day, "cost": 0.0, "units": 0.0,
|
||
"observedWeek": 0.0, "predictedWeek": 0.0,
|
||
"observed5h": 0.0, "predicted5h": 0.0})
|
||
e["cost"] += c
|
||
e["units"] += u
|
||
if dw is not None:
|
||
e["observedWeek"] += dw
|
||
e["predictedWeek"] += u / cal["week"]["unitsPerPct"]
|
||
if d5 is not None:
|
||
e["observed5h"] += d5
|
||
e["predicted5h"] += u / cal["h5"]["unitsPerPct"]
|
||
for e in daily.values():
|
||
e["untrackedWeek"] = max(0.0, e["observedWeek"] - e["predictedWeek"])
|
||
return {"account": account, "days": days,
|
||
"series": {w: [{"t": t, "pct": p, "resetsAt": r} for t, p, r in pts]
|
||
for w, pts in series.items()},
|
||
"hourly": hourly, "daily": list(daily.values()), "calibration": cal}
|
||
|
||
|
||
def session_totals(deps, sid: str) -> tuple[float, float, str] | None:
|
||
"""(list $, plan units, account) of one session — its main transcript
|
||
(the newest copy, if resumed into several files) plus every subagent."""
|
||
from conversations.cards import _summaries_with_account
|
||
best, cost, units, acct = None, 0.0, 0.0, None
|
||
for path, s, a in _summaries_with_account(deps):
|
||
if s.get("sessionId") != sid:
|
||
continue
|
||
acct = acct or a
|
||
if s.get("isSidechain"):
|
||
cost += s.get("cost") or 0.0
|
||
units += s.get("planUnits") or 0.0
|
||
continue
|
||
rank = (s.get("endedAt") or "", s.get("messages") or 0, path)
|
||
if best is None or rank > best[0]:
|
||
best = (rank, s)
|
||
if best is None:
|
||
return None
|
||
cost += best[1].get("cost") or 0.0
|
||
units += best[1].get("planUnits") or 0.0
|
||
return cost, units, acct or "personal"
|
||
|
||
|
||
def finish_suffix(deps, sid: str) -> str:
|
||
"""`` · $3.20 · ≈0.4% wk`` for a finished session's closing push."""
|
||
tot = session_totals(deps, sid) if sid else None
|
||
if not tot or tot[0] <= 0:
|
||
return ""
|
||
cost, units, acct = tot
|
||
pct = plan_pct(deps, acct, units) or {}
|
||
wk = pct.get("week") or 0.0
|
||
return f" · ${cost:.2f} · ≈{wk:.1f}% wk" if wk >= 0.05 else f" · ${cost:.2f}"
|