Pascal MichelandClaude Opus 5
13213ea0ac
- .claude/agents/planer.md - .claude/settings.json - .claude/workflows/executer-welle.js - .claude/workflows/plankritik.js - .claude/workflows/planung.js - .claude/workflows/quorum.js - .claude/workflows/verbrauch.js - .claude/workflows/welle-mit-pruefung.js - .vault-tools/shell-wache/ENTWURF.md - .vault-tools/shell-wache/docs/code/shell_wache.md - .vault-tools/shell-wache/docs/plans/2026-10-30-shell-wache/00-uebersicht.md - .vault-tools/shell-wache/docs/plans/2026-10-30-shell-wache/task-01-shell-wache.md - … und 34 weitere Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py. Co-Authored-By: Claude Opus 5 <[email protected]>
602 lines
32 KiB
Python
602 lines
32 KiB
Python
"""Zählt Tokenverbrauch und Listenpreis aus den Claude-Code-Transkripten eines projects-Ordners.
|
|
|
|
Ersteller: Pascal Michel
|
|
Erstellt: 2026-10-30
|
|
Doku: LIESMICH.md in diesem Ordner · Fassung und Änderungen: dort unter „Fassungen"
|
|
|
|
Was es misst: je Nachricht (message.id) die usage-Felder input_tokens, cache_creation_input_tokens
|
|
(getrennt nach 5 min und 1 h, wo das Transkript das trägt), cache_read_input_tokens und
|
|
output_tokens, dazu Modell, Effort, Werkzeugaufrufe, Größe der Werkzeugergebnisse und Sitzungstitel.
|
|
Es gibt keine Inhalte aus, nur Zahlen, Namen von Modellen, Werkzeugen und Sitzungstitel.
|
|
|
|
Drei Zählregeln, ohne die die Zahlen falsch werden (Harness-Erinnerung
|
|
transkript-usage-groesster-wert-je-nachricht):
|
|
- Je message.id der größte Wert jedes usage-Felds. Bis Claude Code 2.1.304 trägt in
|
|
Subagenten-Transkripten die erste Zeile einer Nachricht nur eine Teilzahl der output_tokens.
|
|
- message.id über alle Dateien dedupliziert. Fortgesetzte Sitzungen tragen die Nachrichten ihrer
|
|
Vorgängerin; gezählt wird die Nachricht in der ersten Datei in Sortierreihenfolge.
|
|
- Haiku 5.5 über 100k Prompt mit der höheren Preisstufe.
|
|
|
|
Was es nicht sieht:
|
|
- Wie der Max-Plan zählt. Die Gewichte der Planprozente sind nicht veröffentlicht; die USD-Zahlen
|
|
sind Listenpreis-Äquivalente nach PREISE und taugen nur zum Vergleich, nicht als Planpunkte.
|
|
- Ob output_tokens der abgerechnete Wert ist. Angenommen, nicht belegt; Thinking steckt nach
|
|
derselben Annahme darin.
|
|
- Gelöschte Transkripte, Verbrauch anderer Konten und Rechner, Codex.
|
|
- PREISE ist abgeschrieben aus der Preistabelle des Workflow-Reviews vom 2026-10-30, nicht aus einer
|
|
Quelle, die das Skript prüft. Ein unbekanntes Modell rechnet es mit Opus-5.5-Preisen und weist es
|
|
als „unbekannt:<name>" aus.
|
|
|
|
Ein Messwerkzeug, kein Nachweis: Die Fixture in faelle.py belegt nur die drei Zählregeln.
|
|
Aufruf: python -I -X utf8 analyse_transkripte.py <projects-Ordner> --ausgabe <Ordner außerhalb des Vaults>
|
|
[--seit JJJJ-MM-TT] [--auswahl REGEX]
|
|
Exit 0 mit Auswertung, 2 bei fehlendem projects-Ordner, Ausgabe im Vault oder ungültigem Argument.
|
|
"""
|
|
import argparse
|
|
import collections
|
|
import datetime
|
|
import json
|
|
import os
|
|
import re
|
|
import statistics
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# $/MTok: Eingabe, Ausgabe, Cache-Lesen, Cache-Schreiben 5 min, Cache-Schreiben 1 h
|
|
PREISE = {
|
|
"claude-fable-5-1": (10, 50, 0.25, 12.5, 20),
|
|
"claude-fable-5": (10, 50, 1.0, 12.5, 20),
|
|
"claude-opus-5-5": (4, 20, 0.20, 5, 8),
|
|
"claude-opus-5": (5, 25, 0.5, 6.25, 10),
|
|
"claude-opus-4-8": (5, 25, 0.5, 6.25, 10),
|
|
"claude-opus-4-7": (5, 25, 0.5, 6.25, 10),
|
|
"claude-opus-4-6": (5, 25, 0.5, 6.25, 10),
|
|
"claude-sonnet-5-5": (2, 10, 0.10, 2.5, 4),
|
|
"claude-sonnet-5": (2, 10, 0.20, 2.5, 4),
|
|
"claude-sonnet-4-6": (3, 15, 0.3, 3.75, 6),
|
|
"claude-haiku-5-5": (0.10, 0.50, 0.01, 0.125, 0.2),
|
|
"claude-haiku-4-5": (1, 5, 0.1, 1.25, 2),
|
|
}
|
|
HAIKU55_UEBER_100K = (0.50, 2.50, 0.05, 0.625, 1.0)
|
|
# Längster Präfix zuerst, sonst fängt claude-opus-5 auch claude-opus-5-5 ab.
|
|
ORDNUNG = sorted(PREISE, key=len, reverse=True)
|
|
USAGE_FELDER = ("input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens")
|
|
STUFEN = [(0, 100000, "0-100k"), (100000, 200000, "100-200k"), (200000, 400000, "200-400k"), (400000, 700000, "400-700k"), (700000, 10**9, "700k+")]
|
|
VAULT = Path(__file__).resolve().parents[2]
|
|
|
|
ROOT = OUT = SEIT = None
|
|
AUSWAHL = None
|
|
|
|
|
|
def norm(m):
|
|
m = m or "?"
|
|
for k in ORDNUNG:
|
|
if m.startswith(k):
|
|
return k
|
|
return "unbekannt:" + m
|
|
|
|
|
|
def kosten(mk, u, cc, ctx):
|
|
p = PREISE.get(mk, PREISE["claude-opus-5-5"])
|
|
if mk == "claude-haiku-5-5" and ctx > 100000:
|
|
p = HAIKU55_UEBER_100K
|
|
c5 = cc.get("ephemeral_5m_input_tokens")
|
|
c1 = cc.get("ephemeral_1h_input_tokens")
|
|
# Ältere Transkripte ohne Aufteilung: alles als 5-min-Schreiben, das ist die billigere Annahme.
|
|
if c5 is None and c1 is None:
|
|
c5 = u.get("cache_creation_input_tokens") or 0
|
|
c1 = 0
|
|
return ((u.get("input_tokens") or 0) * p[0] + (u.get("output_tokens") or 0) * p[1]
|
|
+ (u.get("cache_read_input_tokens") or 0) * p[2] + (c5 or 0) * p[3] + (c1 or 0) * p[4]) / 1e6
|
|
|
|
|
|
def Z():
|
|
return collections.Counter()
|
|
|
|
|
|
agg = {
|
|
"modell": collections.defaultdict(Z), # (scope, art, modell)
|
|
"effort": collections.defaultdict(Z), # (scope, modell, effort) nur Hauptsitzungen
|
|
"tag": collections.defaultdict(Z), # (datum, art)
|
|
"tool": collections.defaultdict(Z), # name
|
|
"stop": collections.Counter(),
|
|
"versionen": collections.Counter(),
|
|
"teilausgabe_version": collections.Counter(),
|
|
"agent_tool": collections.Counter(), # (subagent_type, model)
|
|
"workflow_calls": 0,
|
|
"dateien": 0,
|
|
"zeilen": 0,
|
|
"doppelt_dateiuebergreifend": 0,
|
|
"nachrichten_mit_teilausgabe": 0,
|
|
"ctx_stufe": collections.defaultdict(Z), # (art, stufe) seit SEIT: Kosten je Runde nach Kontextgroesse
|
|
"read_pfad": collections.defaultdict(Z), # Pfad: Aufrufe, Zeichen (seit SEIT)
|
|
}
|
|
GESEHEN = set() # Nachrichten-IDs ueber alle Dateien
|
|
BEFUNDE = []
|
|
sessions = {}
|
|
subagents = []
|
|
meta_keys = collections.Counter()
|
|
|
|
|
|
def stufe(ctx):
|
|
for a, b, n in STUFEN:
|
|
if a <= ctx < b:
|
|
return n
|
|
return "?"
|
|
|
|
|
|
def kurzpfad(p):
|
|
p = str(p).replace("\\", "/")
|
|
teile = p.split("/")
|
|
return "/".join(teile[-3:]) if len(teile) > 3 else p
|
|
|
|
|
|
def dateien():
|
|
for proj in sorted(os.listdir(ROOT)):
|
|
pd = os.path.join(ROOT, proj)
|
|
if not os.path.isdir(pd):
|
|
continue
|
|
for root, dirs, files in os.walk(pd):
|
|
dirs.sort()
|
|
for f in sorted(files):
|
|
if not f.endswith(".jsonl"):
|
|
continue
|
|
p = os.path.join(root, f)
|
|
rel = os.path.relpath(p, pd).replace("\\", "/")
|
|
parts = rel.split("/")
|
|
if len(parts) == 1:
|
|
yield proj, parts[0][:-6], None, p
|
|
elif "subagents" in parts:
|
|
yield proj, parts[0], f[:-6], p
|
|
else:
|
|
yield proj, parts[0], "sonst:" + f[:-6], p
|
|
|
|
|
|
def lese_meta(path):
|
|
mp = path[:-6] + ".meta.json"
|
|
out = {}
|
|
if os.path.exists(mp):
|
|
try:
|
|
d = json.load(open(mp, encoding="utf-8"))
|
|
except Exception:
|
|
return out
|
|
if isinstance(d, dict):
|
|
for k, v in d.items():
|
|
meta_keys[k] += 1
|
|
if k.lower() in ("prompt", "message", "content", "messages", "systemprompt", "system_prompt"):
|
|
continue
|
|
if isinstance(v, str) and len(v) <= 80:
|
|
out[k] = v
|
|
elif isinstance(v, (int, float, bool)):
|
|
out[k] = v
|
|
return out
|
|
|
|
|
|
def verarbeite(proj, sid, agent_id, path):
|
|
art_datei = "haupt" if agent_id is None else "sub"
|
|
s = {"proj": proj, "sid": sid, "agent": agent_id, "turns": 0, "in": 0, "cc": 0, "cr": 0, "out": 0,
|
|
"usd": 0.0, "first": None, "last": None, "max_ctx": 0, "last_ctx": 0, "modelle": collections.Counter(),
|
|
"efforts": collections.Counter(), "tools": collections.Counter(), "title": None, "slug": None,
|
|
"cold": 0, "sidechain_usd": 0.0, "sidechain_turns": 0, "turns_seit": 0, "usd_seit": 0.0,
|
|
"out_seit": 0, "cr_seit": 0, "in_seit": 0, "cc_seit": 0, "doppelt": 0}
|
|
id2name = {}
|
|
msgs = {}
|
|
with open(path, encoding="utf-8", errors="replace") as fh:
|
|
for line in fh:
|
|
agg["zeilen"] += 1
|
|
try:
|
|
o = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
if not isinstance(o, dict):
|
|
continue
|
|
t = o.get("type")
|
|
if t == "custom-title":
|
|
s["title"] = o.get("customTitle")
|
|
continue
|
|
if t not in ("assistant", "user"):
|
|
continue
|
|
ts = o.get("timestamp") or ""
|
|
if t == "assistant":
|
|
m = o.get("message") or {}
|
|
if not isinstance(m, dict):
|
|
continue
|
|
c = m.get("content")
|
|
if isinstance(c, list):
|
|
for b in c:
|
|
if isinstance(b, dict) and b.get("type") == "tool_use":
|
|
n = b.get("name") or "?"
|
|
id2name[b.get("id")] = n
|
|
s["tools"][n] += 1
|
|
z = agg["tool"][n]
|
|
z["aufrufe"] += 1
|
|
if ts >= SEIT:
|
|
z["aufrufe_seit"] += 1
|
|
inp = b.get("input") or {}
|
|
if n == "Read" and isinstance(inp, dict) and inp.get("file_path"):
|
|
id2name[b.get("id")] = ("Read", kurzpfad(inp.get("file_path")))
|
|
if n == "Agent" and isinstance(inp, dict) and ts >= SEIT:
|
|
agg["agent_tool"][(str(inp.get("subagent_type") or "-"), str(inp.get("model") or "-"))] += 1
|
|
if n == "Workflow" and ts >= SEIT:
|
|
agg["workflow_calls"] += 1
|
|
u = m.get("usage") or {}
|
|
if not isinstance(u, dict) or not u:
|
|
continue
|
|
mid = m.get("id") or o.get("requestId")
|
|
rec = msgs.get(mid)
|
|
if rec is None:
|
|
msgs[mid] = {"ts": ts, "mk": norm(m.get("model") or o.get("requestedModel")), "eff": str(o.get("effort") or "?"),
|
|
"side": bool(o.get("isSidechain")), "stop": str(m.get("stop_reason")), "version": str(o.get("version")),
|
|
"slug": o.get("slug"), "u": {k: (u.get(k) or 0) for k in USAGE_FELDER},
|
|
"cc": dict(u.get("cache_creation") or {}), "erste_out": u.get("output_tokens") or 0}
|
|
else:
|
|
for k in USAGE_FELDER:
|
|
rec["u"][k] = max(rec["u"][k], u.get(k) or 0)
|
|
cc = u.get("cache_creation") or {}
|
|
for k in ("ephemeral_5m_input_tokens", "ephemeral_1h_input_tokens"):
|
|
if cc.get(k) is not None:
|
|
rec["cc"][k] = max(rec["cc"].get(k) or 0, cc.get(k) or 0)
|
|
else:
|
|
m = o.get("message") or {}
|
|
c = m.get("content") if isinstance(m, dict) else None
|
|
if isinstance(c, list):
|
|
for b in c:
|
|
if isinstance(b, dict) and b.get("type") == "tool_result":
|
|
n = id2name.get(b.get("tool_use_id"), "?")
|
|
pfad = None
|
|
if isinstance(n, tuple):
|
|
n, pfad = n
|
|
cont = b.get("content")
|
|
if isinstance(cont, str):
|
|
ch = len(cont); txt = cont[:300]
|
|
elif isinstance(cont, list):
|
|
ch = sum(len(x.get("text", "")) for x in cont if isinstance(x, dict))
|
|
txt = "".join(x.get("text", "") for x in cont if isinstance(x, dict))[:300]
|
|
else:
|
|
ch = 0; txt = ""
|
|
z = agg["tool"][n]
|
|
z["ergebnisse"] += 1; z["zeichen"] += ch
|
|
z["max"] = max(z["max"], ch)
|
|
if ch > 50000:
|
|
z["ueber50k"] += 1
|
|
if "persisted-output" in txt:
|
|
z["persistiert"] += 1
|
|
if ts >= SEIT:
|
|
z["zeichen_seit"] += ch; z["ergebnisse_seit"] += 1
|
|
if ch > 50000:
|
|
z["ueber50k_seit"] += 1
|
|
if pfad:
|
|
zp = agg["read_pfad"][pfad]
|
|
zp["aufrufe"] += 1; zp["zeichen"] += ch
|
|
# Nachrichten einmal zaehlen, mit dem groessten Wert je Feld; Kopien anderer Dateien ueberspringen
|
|
for mid, rec in msgs.items():
|
|
if mid in GESEHEN:
|
|
s["doppelt"] += 1; agg["doppelt_dateiuebergreifend"] += 1
|
|
continue
|
|
GESEHEN.add(mid)
|
|
u = rec["u"]; ts = rec["ts"]; mk = rec["mk"]; eff = rec["eff"]
|
|
if u["output_tokens"] > rec["erste_out"]:
|
|
agg["nachrichten_mit_teilausgabe"] += 1
|
|
agg["teilausgabe_version"][rec["version"]] += 1
|
|
ti, tc, tr, to = u["input_tokens"], u["cache_creation_input_tokens"], u["cache_read_input_tokens"], u["output_tokens"]
|
|
ctx = ti + tc + tr
|
|
usd = kosten(mk, u, rec["cc"], ctx)
|
|
art = "sub" if (art_datei == "sub" or rec["side"]) else "haupt"
|
|
if art_datei == "haupt" and rec["side"]:
|
|
s["sidechain_usd"] += usd; s["sidechain_turns"] += 1
|
|
scopes = ["gesamt"] + (["seit"] if ts >= SEIT else [])
|
|
for scope in scopes:
|
|
z = agg["modell"][(scope, art, mk)]
|
|
z["turns"] += 1; z["in"] += ti; z["cc"] += tc; z["cr"] += tr; z["out"] += to; z["usd_x1000"] += int(usd * 1000)
|
|
if tr == 0:
|
|
z["kalt"] += 1
|
|
if art == "haupt":
|
|
z2 = agg["effort"][(scope, mk, eff)]
|
|
z2["turns"] += 1; z2["out"] += to; z2["usd_x1000"] += int(usd * 1000); z2["ctx_sum"] += ctx
|
|
tag = ts[:10]
|
|
zt = agg["tag"][(tag, art)]
|
|
zt["turns"] += 1; zt["out"] += to; zt["usd_x1000"] += int(usd * 1000); zt["cr"] += tr; zt["cc"] += tc; zt["in"] += ti
|
|
if ts >= SEIT:
|
|
zs = agg["ctx_stufe"][(art, stufe(ctx))]
|
|
zs["turns"] += 1; zs["usd_x1000"] += int(usd * 1000); zs["cr"] += tr; zs["cc"] += tc; zs["out"] += to
|
|
if s["turns"] == 0:
|
|
s["first_cc"] = tc + ti
|
|
s["turns"] += 1; s["in"] += ti; s["cc"] += tc; s["cr"] += tr; s["out"] += to; s["usd"] += usd
|
|
if ts >= SEIT:
|
|
s["turns_seit"] += 1; s["usd_seit"] += usd; s["out_seit"] += to; s["cr_seit"] += tr; s["in_seit"] += ti; s["cc_seit"] += tc
|
|
s["first"] = ts if (s["first"] is None or ts < s["first"]) else s["first"]
|
|
s["last"] = ts if (s["last"] is None or ts > s["last"]) else s["last"]
|
|
s["max_ctx"] = max(s["max_ctx"], ctx)
|
|
s["last_ctx"] = ctx
|
|
s["modelle"][mk] += 1
|
|
s["efforts"][eff] += 1
|
|
if tr == 0:
|
|
s["cold"] += 1
|
|
if not s["slug"] and rec["slug"]:
|
|
s["slug"] = rec["slug"]
|
|
agg["stop"][rec["stop"]] += 1
|
|
agg["versionen"][rec["version"]] += 1
|
|
return s
|
|
|
|
|
|
def mio(n):
|
|
return f"{n/1e6:.2f}"
|
|
|
|
|
|
def usd(c):
|
|
return f"{c:.2f}"
|
|
|
|
|
|
def dauer_h(s):
|
|
try:
|
|
a = datetime.datetime.fromisoformat(s["first"].replace("Z", "+00:00"))
|
|
b = datetime.datetime.fromisoformat(s["last"].replace("Z", "+00:00"))
|
|
except (AttributeError, ValueError):
|
|
return "?"
|
|
return f"{(b - a).total_seconds() / 3600:.1f}"
|
|
|
|
|
|
def auswahl_bericht():
|
|
"""Je Hauptsitzung, deren Titel, Slug oder Kennung AUSWAHL trifft: Runden, Ausgabe, Effort, Subagenten."""
|
|
rx = re.compile(AUSWAHL, re.I)
|
|
L = [f"# Auswahl `{AUSWAHL}` (seit {SEIT}, Fassung 2)\n",
|
|
"Runden und Ausgabe ohne Subagenten; Dauer ist erste bis letzte Runde, Pausen eingeschlossen.\n",
|
|
"| Beginn | Titel | Kennung | Effort (Runden) | Runden | Dauer h | Ausgabe | Ausgabe je Runde | USD | max. Kontext |",
|
|
"|---|---|---|---|---|---|---|---|---|---|"]
|
|
treffer = [s for s in sessions.values()
|
|
if s["turns_seit"] > 0 and any(rx.search(str(x or "")) for x in (s["title"], s["slug"], s["sid"]))]
|
|
for s in sorted(treffer, key=lambda s: s["first"] or ""):
|
|
ef = ", ".join(f"{k} {v}" for k, v in s["efforts"].most_common())
|
|
L.append(f"| {(s['first'] or '')[:16]} | {(s['title'] or s['slug'] or '-')[:50]} | {s['sid'][:8]} | {ef} | {s['turns_seit']} "
|
|
f"| {dauer_h(s)} | {s['out_seit']} | {s['out_seit'] // max(s['turns_seit'], 1)} | {usd(s['usd_seit'])} | {s['max_ctx']} |")
|
|
L.append("\n## Subagenten der ausgewählten Sitzungen\n")
|
|
L.append("| Kennung | Typ (meta) | Modell | Effort | Anzahl | Runden | Ausgabe | USD |")
|
|
L.append("|---|---|---|---|---|---|---|---|")
|
|
schluessel = {(s["proj"], s["sid"]) for s in treffer}
|
|
grp = collections.defaultdict(lambda: [0, 0, 0, 0.0])
|
|
for sa in subagents:
|
|
if (sa["proj"], sa["sid"]) not in schluessel or sa["turns"] == 0:
|
|
continue
|
|
meta = sa.get("meta") or {}
|
|
mk = sa["modelle"].most_common(1)[0][0] if sa["modelle"] else "?"
|
|
ef = sa["efforts"].most_common(1)[0][0] if sa["efforts"] else "?"
|
|
g = grp[(sa["sid"][:8], str(meta.get("agentType") or "-"), mk, ef)]
|
|
g[0] += 1; g[1] += sa["turns"]; g[2] += sa["out"]; g[3] += sa["usd"]
|
|
for k, g in sorted(grp.items()):
|
|
L.append(f"| {k[0]} | {k[1]} | {k[2]} | {k[3]} | {g[0]} | {g[1]} | {g[2]} | {usd(g[3])} |")
|
|
L.append(f"\nTreffer: {len(treffer)} Hauptsitzungen.")
|
|
return "\n".join(L)
|
|
|
|
|
|
def bericht(t0):
|
|
L = []
|
|
L.append(f"# Transkript-Auswertung (Zeitraum seit {SEIT}, Stand {datetime.date.today().isoformat()}, Fassung 2)\n")
|
|
L.append(f"Quelle: {ROOT} — {agg['dateien']} JSONL-Dateien, {agg['zeilen']} Zeilen, Lauf {round(time.time()-t0)} s. Kosten sind Listenpreis-Äquivalente (USD) nach der Preistabelle im Skript; der Max-Plan rechnet nicht in USD ab, die Zahl dient dem Vergleich der Hebel. Fassung 2 zählt je Nachrichten-ID den größten Wert jedes usage-Felds (vor Claude Code 2.1.308 trug die erste Zeile einer Subagenten-Nachricht nur eine Teilzahl der Ausgabe; betroffen: {agg['nachrichten_mit_teilausgabe']} Nachrichten), überspringt Nachrichten, die schon in einer anderen Datei standen ({agg['doppelt_dateiuebergreifend']} Kopien), und rechnet Haiku 5.5 über 100k Prompt mit dem Fünffachen.\n")
|
|
L.append("Nachrichten mit Teilausgabe in der ersten Zeile nach Claude-Code-Version: "
|
|
+ (", ".join(f"{k}: {v}" for k, v in sorted(agg["teilausgabe_version"].items())) or "keine") + "\n")
|
|
if BEFUNDE:
|
|
L.append("Hinweise: " + "; ".join(BEFUNDE[:20]) + (f" (und {len(BEFUNDE) - 20} weitere)" if len(BEFUNDE) > 20 else "") + "\n")
|
|
|
|
for scope in ("seit", "gesamt"):
|
|
L.append(f"\n## Verbrauch nach Modell ({'seit ' + SEIT if scope == 'seit' else 'gesamt'})\n")
|
|
L.append("| Art | Modell | Runden | Eingabe Mio | Cache-Schreiben Mio | Cache-Lesen Mio | Ausgabe Mio | USD | Anteil USD | Runden ohne Cache |")
|
|
L.append("|---|---|---|---|---|---|---|---|---|---|")
|
|
rows = [(k, v) for k, v in agg["modell"].items() if k[0] == scope]
|
|
total = sum(v["usd_x1000"] for _, v in rows) / 1000 or 1
|
|
for (sc, art, mk), v in sorted(rows, key=lambda kv: -kv[1]["usd_x1000"]):
|
|
c = v["usd_x1000"] / 1000
|
|
L.append(f"| {art} | {mk} | {v['turns']} | {mio(v['in'])} | {mio(v['cc'])} | {mio(v['cr'])} | {mio(v['out'])} | {usd(c)} | {100*c/total:.0f} % | {v['kalt']} |")
|
|
L.append(f"| **Summe** | | | | | | | **{usd(total)}** | | |")
|
|
|
|
L.append(f"\n## Kostenzerlegung je Art und Modell (seit {SEIT}, Listenpreis)\n")
|
|
L.append("| Art | Modell | Eingabe | Cache-Schreiben | Cache-Lesen | Ausgabe |")
|
|
L.append("|---|---|---|---|---|---|")
|
|
for (sc, art, mk), v in sorted(((k, v) for k, v in agg["modell"].items() if k[0] == "seit"), key=lambda kv: -kv[1]["usd_x1000"])[:6]:
|
|
p = PREISE.get(mk, PREISE["claude-opus-5-5"])
|
|
ein = v["in"] * p[0] / 1e6; les = v["cr"] * p[2] / 1e6; aus = v["out"] * p[1] / 1e6
|
|
ges = v["usd_x1000"] / 1000 or 1
|
|
schr = max(ges - ein - les - aus, 0)
|
|
L.append(f"| {art} | {mk} | {100*ein/ges:.0f} % | {100*schr/ges:.0f} % | {100*les/ges:.0f} % | {100*aus/ges:.0f} % |")
|
|
|
|
L.append(f"\n## Hauptsitzungen nach Modell und Effort (seit {SEIT})\n")
|
|
L.append("| Modell | Effort | Runden | Ausgabe Mio | Ausgabe je Runde | USD | mittlerer Kontext je Runde |")
|
|
L.append("|---|---|---|---|---|---|---|")
|
|
for (sc, mk, eff), v in sorted(((k, v) for k, v in agg["effort"].items() if k[0] == "seit"), key=lambda kv: -kv[1]["usd_x1000"]):
|
|
L.append(f"| {mk} | {eff} | {v['turns']} | {mio(v['out'])} | {v['out']//max(v['turns'],1)} | {usd(v['usd_x1000']/1000)} | {v['ctx_sum']//max(v['turns'],1)} |")
|
|
|
|
L.append("\n## Je Kalenderwoche (UTC-Datum)\n")
|
|
L.append("| Woche | Runden Haupt | USD Haupt | Runden Sub | USD Sub | Ausgabe Mio Haupt | Ausgabe Mio Sub | Cache-Lesen Mio | Cache-Schreiben Mio |")
|
|
L.append("|---|---|---|---|---|---|---|---|---|")
|
|
wochen = collections.defaultdict(Z)
|
|
for (tag, art), v in agg["tag"].items():
|
|
try:
|
|
d = datetime.date.fromisoformat(tag)
|
|
except Exception:
|
|
continue
|
|
w = f"{d.isocalendar()[0]}-W{d.isocalendar()[1]:02d}"
|
|
z = wochen[w]
|
|
z[f"turns_{art}"] += v["turns"]; z[f"usd_{art}"] += v["usd_x1000"]; z[f"out_{art}"] += v["out"]; z["cr"] += v["cr"]; z["cc"] += v["cc"]
|
|
for w in sorted(wochen):
|
|
z = wochen[w]
|
|
L.append(f"| {w} | {z['turns_haupt']} | {usd(z['usd_haupt']/1000)} | {z['turns_sub']} | {usd(z['usd_sub']/1000)} | {mio(z['out_haupt'])} | {mio(z['out_sub'])} | {mio(z['cr'])} | {mio(z['cc'])} |")
|
|
|
|
L.append(f"\n## Teuerste Hauptsitzungen (seit {SEIT}, 25)\n")
|
|
L.append("| Beginn | Titel | Modell / Effort | Runden | USD | max. Kontext | Cache-Lesen Anteil | Subagenten (n / USD) |")
|
|
L.append("|---|---|---|---|---|---|---|---|")
|
|
sub_by_parent = collections.defaultdict(lambda: [0, 0.0])
|
|
for sa in subagents:
|
|
if (sa["last"] or "") >= SEIT:
|
|
sub_by_parent[(sa["proj"], sa["sid"])][0] += 1
|
|
sub_by_parent[(sa["proj"], sa["sid"])][1] += sa["usd"]
|
|
haupt = [s for s in sessions.values() if s["turns_seit"] > 0]
|
|
for s in sorted(haupt, key=lambda s: -s["usd_seit"])[:25]:
|
|
mk = ", ".join(f"{k}" for k, _ in s["modelle"].most_common(2))
|
|
ef = ", ".join(f"{k}" for k, _ in s["efforts"].most_common(2))
|
|
ges = s["in_seit"] + s["cc_seit"] + s["cr_seit"] or 1
|
|
n, c = sub_by_parent.get((s["proj"], s["sid"]), [0, 0.0])
|
|
L.append(f"| {(s['first'] or '')[:10]} | {(s['title'] or s['slug'] or s['sid'][:8])[:50]} | {mk} / {ef} | {s['turns_seit']} | {usd(s['usd_seit'])} | {s['max_ctx']} | {100*s['cr_seit']/ges:.0f} % | {n} / {usd(c)} |")
|
|
L.append(f"\nHauptsitzungen mit Runden seit {SEIT}: {len(haupt)}; davon mit maximalem Kontext über 400k: {sum(1 for s in haupt if s['max_ctx'] > 400000)}, über 700k: {sum(1 for s in haupt if s['max_ctx'] > 700000)}.")
|
|
L.append(f"Summe USD Hauptsitzungen seit {SEIT}: {usd(sum(s['usd_seit'] for s in haupt))}; Summe USD Subagenten seit {SEIT}: {usd(sum(c for n, c in sub_by_parent.values()))}.")
|
|
|
|
L.append(f"\n## Kontextgröße der Hauptsitzungen (max je Sitzung, seit {SEIT})\n")
|
|
L.append("| Stufe | Sitzungen | USD | Anteil USD |")
|
|
L.append("|---|---|---|---|")
|
|
tot = sum(s["usd_seit"] for s in haupt) or 1
|
|
for a, b, n in STUFEN:
|
|
grp = [s for s in haupt if a <= s["max_ctx"] < b]
|
|
L.append(f"| {n} | {len(grp)} | {usd(sum(s['usd_seit'] for s in grp))} | {100*sum(s['usd_seit'] for s in grp)/tot:.0f} % |")
|
|
|
|
L.append(f"\n## Subagenten (seit {SEIT}, aus Subagenten-Transkripten)\n")
|
|
sa_seit = [sa for sa in subagents if (sa["last"] or "") >= SEIT and sa["turns"] > 0]
|
|
L.append(f"Anzahl: {len(sa_seit)}; Meta-Schlüssel gesehen: {dict(meta_keys.most_common(12))}\n")
|
|
L.append("| Typ (meta) | Modell | Effort | Anzahl | Median Ausgabe | Median USD | Summe USD | Median Runden | Median max. Kontext |")
|
|
L.append("|---|---|---|---|---|---|---|---|---|")
|
|
grp = collections.defaultdict(list)
|
|
for sa in sa_seit:
|
|
meta = sa.get("meta") or {}
|
|
typ = str(meta.get("agentType") or meta.get("subagent_type") or meta.get("type") or meta.get("agent_type") or "-")
|
|
mk = sa["modelle"].most_common(1)[0][0] if sa["modelle"] else "?"
|
|
ef = sa["efforts"].most_common(1)[0][0] if sa["efforts"] else "?"
|
|
grp[(typ, mk, ef)].append(sa)
|
|
for k, lst in sorted(grp.items(), key=lambda kv: -sum(x["usd"] for x in kv[1])):
|
|
L.append(f"| {k[0]} | {k[1]} | {k[2]} | {len(lst)} | {int(statistics.median(x['out'] for x in lst))} | {usd(statistics.median(x['usd'] for x in lst))} | {usd(sum(x['usd'] for x in lst))} | {int(statistics.median(x['turns'] for x in lst))} | {int(statistics.median(x['max_ctx'] for x in lst))} |")
|
|
|
|
L.append(f"\n## Subagenten nach Workflow-Phase (meta.workflowPhase, seit {SEIT})\n")
|
|
L.append("| Phase | Anzahl | Summe USD | Mittel USD | Median Runden |")
|
|
L.append("|---|---|---|---|---|")
|
|
ph = collections.defaultdict(list)
|
|
for sa in sa_seit:
|
|
ph[str((sa.get("meta") or {}).get("workflowPhase") or "-")].append(sa)
|
|
for k, lst in sorted(ph.items(), key=lambda kv: -sum(x["usd"] for x in kv[1]))[:20]:
|
|
L.append(f"| {k} | {len(lst)} | {usd(sum(x['usd'] for x in lst))} | {usd(sum(x['usd'] for x in lst)/len(lst))} | {int(statistics.median(x['turns'] for x in lst))} |")
|
|
|
|
L.append(f"\n## Agent-Werkzeug: subagent_type und model (seit {SEIT})\n")
|
|
L.append("| subagent_type | model | Aufrufe |")
|
|
L.append("|---|---|---|")
|
|
for (st, mo), n in agg["agent_tool"].most_common(25):
|
|
L.append(f"| {st} | {mo} | {n} |")
|
|
L.append(f"\nWorkflow-Aufrufe seit {SEIT}: {agg['workflow_calls']}")
|
|
|
|
L.append(f"\n## Werkzeuge nach Ergebnisvolumen (seit {SEIT}, Zeichen der Werkzeugergebnisse)\n")
|
|
L.append("| Werkzeug | Aufrufe | Ergebnis-MB | Ø Zeichen | Max Zeichen | Ergebnisse > 50k | persistiert |")
|
|
L.append("|---|---|---|---|---|---|---|")
|
|
for n, z in sorted(agg["tool"].items(), key=lambda kv: -kv[1]["zeichen_seit"])[:30]:
|
|
L.append(f"| {n} | {z['aufrufe_seit']} | {z['zeichen_seit']/1e6:.1f} | {z['zeichen_seit']//max(z['ergebnisse_seit'],1)} | {z['max']} | {z['ueber50k_seit']} | {z['persistiert']} |")
|
|
|
|
L.append(f"\n## Kosten je Runde nach Kontextgröße der Runde (seit {SEIT})\n")
|
|
L.append("| Art | Kontext der Runde | Runden | USD gesamt | USD je Runde | Cache-Lesen je Runde | Cache-Schreiben je Runde | Ausgabe je Runde |")
|
|
L.append("|---|---|---|---|---|---|---|---|")
|
|
for art in ("haupt", "sub"):
|
|
for a, b, n in STUFEN:
|
|
z = agg["ctx_stufe"].get((art, n))
|
|
if not z or not z["turns"]:
|
|
continue
|
|
t = z["turns"]
|
|
L.append(f"| {art} | {n} | {t} | {usd(z['usd_x1000']/1000)} | {z['usd_x1000']/1000/t:.3f} | {z['cr']//t} | {z['cc']//t} | {z['out']//t} |")
|
|
|
|
L.append(f"\n## Fester Startaufwand je Subagent: Tokens der ersten Runde (Eingabe plus Cache-Schreiben), seit {SEIT}\n")
|
|
L.append("| Typ (meta) | Modell | Anzahl | Median erste Runde | Median Runden | Anteil erste Runde am Gesamtkontext-Maximum |")
|
|
L.append("|---|---|---|---|---|---|")
|
|
grp2 = collections.defaultdict(list)
|
|
for sa in sa_seit:
|
|
meta = sa.get("meta") or {}
|
|
typ = str(meta.get("agentType") or "-")
|
|
mk = sa["modelle"].most_common(1)[0][0] if sa["modelle"] else "?"
|
|
if sa.get("first_cc") is not None:
|
|
grp2[(typ, mk)].append(sa)
|
|
for k, lst in sorted(grp2.items(), key=lambda kv: -len(kv[1]))[:20]:
|
|
f = statistics.median(x["first_cc"] for x in lst)
|
|
m = statistics.median(x["max_ctx"] for x in lst) or 1
|
|
L.append(f"| {k[0]} | {k[1]} | {len(lst)} | {int(f)} | {int(statistics.median(x['turns'] for x in lst))} | {100*f/m:.0f} % |")
|
|
|
|
for titel, schl, n in (("nach Zeichen", "zeichen", 40), ("nach Aufrufen", "aufrufe", 25)):
|
|
L.append(f"\n## Meistgelesene Dateien (Read, seit {SEIT}, {titel})\n")
|
|
L.append("| Pfad (letzte drei Glieder) | Aufrufe | Zeichen gesamt Mio | Ø Zeichen |")
|
|
L.append("|---|---|---|---|")
|
|
for p, z in sorted(agg["read_pfad"].items(), key=lambda kv: -kv[1][schl])[:n]:
|
|
L.append(f"| {p} | {z['aufrufe']} | {z['zeichen']/1e6:.2f} | {z['zeichen']//max(z['aufrufe'],1)} |")
|
|
|
|
L.append("\n## Stop-Gründe (gesamt)\n")
|
|
L.append(", ".join(f"{k}: {v}" for k, v in agg["stop"].most_common()))
|
|
L.append("\n## Claude-Code-Versionen (Nachrichten, gesamt)\n")
|
|
L.append(", ".join(f"{k}: {v}" for k, v in agg["versionen"].most_common(12)))
|
|
return "\n".join(L)
|
|
|
|
|
|
def im_vault(pfad):
|
|
p = Path(pfad).resolve()
|
|
return p == VAULT or VAULT in p.parents
|
|
|
|
|
|
def main(argv=None):
|
|
global ROOT, OUT, SEIT, AUSWAHL
|
|
t0 = time.time()
|
|
ap = argparse.ArgumentParser(description="Tokenverbrauch aus Claude-Code-Transkripten")
|
|
ap.add_argument("projects", help="projects-Ordner von Claude Code, etwa D:\\ClaudeData\\config\\projects")
|
|
ap.add_argument("--ausgabe", required=True, help="Ordner für auswertung.md und auswertung.json, nie im Vault")
|
|
ap.add_argument("--seit", default="2026-09-15", help="Beginn des Zeitraums JJJJ-MM-TT (UTC), Vorgabe 2026-09-15")
|
|
ap.add_argument("--auswahl", default=None, help="regulärer Ausdruck auf Titel, Slug oder Kennung der Hauptsitzung; schreibt auswahl.md")
|
|
a = ap.parse_args(argv)
|
|
ROOT, OUT, SEIT, AUSWAHL = a.projects, a.ausgabe, a.seit, a.auswahl
|
|
try:
|
|
datetime.date.fromisoformat(SEIT)
|
|
except ValueError:
|
|
print(f"--seit {SEIT!r} ist kein Datum JJJJ-MM-TT")
|
|
return 2
|
|
if AUSWAHL is not None:
|
|
try:
|
|
re.compile(AUSWAHL)
|
|
except re.error as e:
|
|
print(f"--auswahl ist kein regulärer Ausdruck: {e}")
|
|
return 2
|
|
if not os.path.isdir(ROOT):
|
|
print(f"projects-Ordner fehlt: {ROOT}")
|
|
return 2
|
|
# Sitzungstitel können Kundennamen tragen, und der Vault wird teilweise veröffentlicht.
|
|
if im_vault(OUT):
|
|
print(f"Ausgabe im Vault abgelehnt: {OUT} liegt unter {VAULT}")
|
|
return 2
|
|
os.makedirs(OUT, exist_ok=True)
|
|
|
|
for proj, sid, agent_id, path in dateien():
|
|
agg["dateien"] += 1
|
|
try:
|
|
s = verarbeite(proj, sid, agent_id, path)
|
|
except Exception as e: # Eine kaputte Datei ist ein Hinweis im Bericht, kein Abbruch der ganzen Zählung.
|
|
BEFUNDE.append(f"{proj}/{os.path.basename(path)}: {type(e).__name__}")
|
|
continue
|
|
if agent_id is None:
|
|
sessions[(proj, sid)] = s
|
|
else:
|
|
s["meta"] = lese_meta(path)
|
|
subagents.append(s)
|
|
|
|
print("Dateien", agg["dateien"], "Zeilen", agg["zeilen"], "Sekunden", round(time.time() - t0),
|
|
"doppelt (andere Datei)", agg["doppelt_dateiuebergreifend"], "Nachrichten mit Teilausgabe in erster Zeile", agg["nachrichten_mit_teilausgabe"])
|
|
md = bericht(t0)
|
|
with open(os.path.join(OUT, "auswertung.md"), "w", encoding="utf-8") as f:
|
|
f.write(md)
|
|
with open(os.path.join(OUT, "auswertung.json"), "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"kennzahlen": {k: agg[k] for k in ("dateien", "zeilen", "doppelt_dateiuebergreifend", "nachrichten_mit_teilausgabe", "workflow_calls")},
|
|
"teilausgabe_version": dict(agg["teilausgabe_version"]),
|
|
"befunde": BEFUNDE,
|
|
"modell": {"|".join(k): dict(v) for k, v in agg["modell"].items()},
|
|
"effort": {"|".join(k): dict(v) for k, v in agg["effort"].items()},
|
|
"tag": {"|".join(k): dict(v) for k, v in agg["tag"].items()},
|
|
"tool": {k: dict(v) for k, v in agg["tool"].items()},
|
|
"ctx_stufe": {"|".join(k): dict(v) for k, v in agg["ctx_stufe"].items()},
|
|
"sessions": [{k: (dict(v) if isinstance(v, collections.Counter) else v) for k, v in s.items()} for s in sessions.values()],
|
|
"subagents": [{k: (dict(v) if isinstance(v, collections.Counter) else v) for k, v in s.items()} for s in subagents],
|
|
}, f, ensure_ascii=False, indent=0)
|
|
if AUSWAHL is not None:
|
|
with open(os.path.join(OUT, "auswahl.md"), "w", encoding="utf-8") as f:
|
|
f.write(auswahl_bericht())
|
|
print(f"Geschrieben: {os.path.join(OUT, 'auswertung.md')}" + (" und auswahl.md" if AUSWAHL is not None else ""))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|