#!/usr/bin/env python3 """Deterministic token fold + API-list USD + proxy bands.""" from __future__ import annotations import argparse import json from datetime import datetime, timezone from pathlib import Path # USD per 1M tokens — skill defaults (update when vendors change list prices) PRICE = { "opus": {"in": 15.0, "out": 75.0, "cache_read": 1.50, "cache_write": 18.75}, "sonnet": {"in": 3.0, "out": 15.0, "cache_read": 0.30, "cache_write": 3.75}, "haiku": {"in": 1.0, "out": 5.0, "cache_read": 0.10, "cache_write": 1.25}, } PRICE_AS_OF = "2026-07-19 (skill defaults; not live vendor API)" def tier(model: str) -> str: n = model.lower() if "haiku" in n: return "haiku" if "sonnet" in n: return "sonnet" return "opus" def fold_claude(sc: dict | None, since: str) -> dict: if not sc: return {"available": False} mu = sc.get("modelUsage") or {} rows = [] tot = {"in": 0, "out": 0, "cache_read": 0, "cache_write": 0, "api_usd": 0.0} for model, v in mu.items(): if not isinstance(v, dict): continue tin = int(v.get("inputTokens") or 0) tout = int(v.get("outputTokens") or 0) cr = int(v.get("cacheReadInputTokens") or 0) cw = int(v.get("cacheCreationInputTokens") or 0) p = PRICE[tier(model)] usd = (tin * p["in"] + tout * p["out"] + cr * p["cache_read"] + cw * p["cache_write"]) / 1e6 rows.append( { "model": model, "tier": tier(model), "input": tin, "output": tout, "cache_read": cr, "cache_write": cw, "api_list_usd": round(usd, 2), } ) tot["in"] += tin tot["out"] += tout tot["cache_read"] += cr tot["cache_write"] += cw tot["api_usd"] += usd da = sc.get("dailyActivity") or [] msgs = sessions = tools = days = 0 for v in da: if not isinstance(v, dict): continue if str(v.get("date", "")) < since: continue days += 1 msgs += int(v.get("messageCount") or 0) sessions += int(v.get("sessionCount") or 0) tools += int(v.get("toolCallCount") or 0) dmt_sum = 0 for day in sc.get("dailyModelTokens") or []: if not isinstance(day, dict) or str(day.get("date", "")) < since: continue for _m, t in (day.get("tokensByModel") or {}).items(): dmt_sum += int(t) return { "available": True, "firstSessionDate": sc.get("firstSessionDate"), "lastComputedDate": sc.get("lastComputedDate"), "totalSessions_lifetime": sc.get("totalSessions"), "totalMessages_lifetime": sc.get("totalMessages"), "daily_activity_since": { "days": days, "messages": msgs, "sessions": sessions, "toolCalls": tools, }, "dailyModelTokens_sum_non_cache_proxy": dmt_sum, "per_model": sorted(rows, key=lambda r: -r["api_list_usd"]), "totals": { "input": tot["in"], "output": tot["out"], "cache_read": tot["cache_read"], "cache_write": tot["cache_write"], "non_cache_in_plus_out": tot["in"] + tot["out"], "all_token_classes": tot["in"] + tot["out"] + tot["cache_read"] + tot["cache_write"], "api_list_usd": round(tot["api_usd"], 2), }, "pricing_as_of": PRICE_AS_OF, "note": ( "Claude stats-cache modelUsage is lifetime of this machine install; " f"firstSessionDate may be after horizon {since}. Cache tokens dominate API-list $." ), } def proxy_harness(sess: dict, util: float, usd_per_mtok_lo: float, usd_per_mtok_hi: float) -> dict: raw = int(sess.get("raw_token_proxy") or 0) util_tok = int(raw * util) return { "files": sess.get("files"), "bytes": sess.get("bytes"), "raw_token_proxy": raw, "utilization": util, "token_band": [int(util_tok * 0.7), int(util_tok * 1.3)], "point_tokens": util_tok, "usd_per_mtok_band": [usd_per_mtok_lo, usd_per_mtok_hi], "usd_band": [ round(util_tok * 0.7 / 1e6 * usd_per_mtok_lo, 2), round(util_tok * 1.3 / 1e6 * usd_per_mtok_hi, 2), ], "class": "semi-decidable-proxy", } def human_band(claude_daily: dict) -> dict: sessions = int((claude_daily or {}).get("sessions") or 0) tools = int((claude_daily or {}).get("toolCalls") or 0) # values-question rates return { "class": "values-question", "sessions_x_2h_x_150usd": sessions * 2 * 150, "toolcalls_x_2min_x_150usd_hr": round(tools * 2 / 60 * 150, 2), "note": "Not cash paid; not market value of artifacts. Opportunity-cost band only.", } def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--evidence", required=True) ap.add_argument("--out", default="estimate.json") ap.add_argument("--since", default=None) args = ap.parse_args() ev = json.loads(Path(args.evidence).read_text(encoding="utf-8")) since = args.since or ev.get("since") or "2025-11-01" claude = fold_claude(ev.get("stats_cache"), since) sessions = ev.get("sessions") or {} grok = proxy_harness(sessions.get("grok_sessions") or {}, 0.35, 4.0, 8.0) codex = proxy_harness(sessions.get("codex_sessions") or {}, 0.35, 3.0, 10.0) c_tot = (claude.get("totals") or {}) if claude.get("available") else {} c_tokens_lo = int(c_tot.get("non_cache_in_plus_out") or 0) c_tokens_hi = int(c_tot.get("all_token_classes") or 0) c_usd = float(c_tot.get("api_list_usd") or 0) g_lo, g_hi = grok["token_band"] x_lo, x_hi = codex["token_band"] gu_lo, gu_hi = grok["usd_band"] xu_lo, xu_hi = codex["usd_band"] estimate = { "estimated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "since": since, "claude_measured": claude, "grok_proxy": grok, "codex_proxy": codex, "combined": { "tokens_band": { "non_cache_plus_proxies": [c_tokens_lo + g_lo + x_lo, c_tokens_lo + g_hi + x_hi], "all_classes_plus_proxies": [c_tokens_hi + g_lo + x_lo, c_tokens_hi + g_hi + x_hi], }, "api_list_usd_band": { "claude_only": c_usd, "with_proxies": [round(c_usd + gu_lo + xu_lo, 2), round(c_usd + gu_hi + xu_hi, 2)], }, }, "human_replacement_band": human_band( (claude.get("daily_activity_since") if claude.get("available") else None) or {} ), "value_layers": { "L1_cash_paid": { "status": "unknown", "class": "tarski-residual", "note": "Requires billing export / seat invoices — not on disk.", }, "L2_api_list_equivalent": { "claude_usd": c_usd, "combined_usd_band": [round(c_usd + gu_lo + xu_lo, 2), round(c_usd + gu_hi + xu_hi, 2)], "class": "measured+proxy", }, "L3_human_opportunity": "see human_replacement_band", "artifact_market_value": { "status": "undecidable", "class": "oracle-required", "note": "Cannot derive product/IP market value from tokens alone.", }, }, "backends": ev.get("backends"), "residuals": [ "Cash paid (subscription seats) unknown without billing API/CSV", "Claude cache tokens dominate API-list $; non-cache in+out is a lower bound", "Grok/Codex exact tokens not available — session-byte proxies only", "stats-cache lastComputedDate may lag wall clock", "Work on other machines / phone / other harnesses not included", "Artifact market value is oracle-required (not token-derived)", ], } Path(args.out).write_text(json.dumps(estimate, indent=2), encoding="utf-8") print(json.dumps({"ok": True, "out": args.out, "combined": estimate["combined"]}, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())