tokentab
§ Source-level Observatory

tokentab pricing aggregation

Pricing aggregation

★ 713
GitHub Stars
2.0K
LOC
24
Files
2026-08-27
Last commit
§ 00 · IMPORT CHAIN

Pricing aggregation的证据链

这一页拆价格表、成本计算、报表分组和活动分类如何串起来。为了避免变成行号清单,本页每个源码定位都绑定具体设计含义:在 sources/tokentab/tokentab/pricing/prices.py:16 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/pricing/prices.py:46 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/pricing/prices.py:54 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/pricing/prices.py:67 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/pricing/prices.py:81 这一处,页面把源码证据和设计判断绑定起来。

sources/tokentab/tokentab/pricing/prices.py · L16-90Python
16TABLE: dict[str, ModelPrice] = {
17    # --- Anthropic (Claude) ---
18    "claude-opus-4-6":   ModelPrice(15, 75, 1.5, 18.75),
19    "claude-opus-4-5":   ModelPrice(15, 75, 1.5, 18.75),
20    "claude-opus-4-1":   ModelPrice(15, 75, 1.5, 18.75),
21    "claude-opus-4":     ModelPrice(15, 75, 1.5, 18.75),
22    "claude-sonnet-4-6": ModelPrice(3, 15, 0.3, 3.75),
23    "claude-sonnet-4-5": ModelPrice(3, 15, 0.3, 3.75),
24    "claude-sonnet-4":   ModelPrice(3, 15, 0.3, 3.75),
25    "claude-haiku-4-5":  ModelPrice(1, 5, 0.1, 1.25),
26    "claude-3-5-haiku":  ModelPrice(0.8, 4, 0.08, 1),
27
28    # --- OpenAI (Codex / GPT) ---
29    "gpt-5.6":      ModelPrice(1.25, 10, 0.125, 0),
30    "gpt-5.1":      ModelPrice(1.25, 10, 0.125, 0),
31    "gpt-5":        ModelPrice(1.25, 10, 0.125, 0),
32    "gpt-5-mini":   ModelPrice(0.25, 2, 0.025, 0),
33    "gpt-5-codex":  ModelPrice(1.25, 10, 0.125, 0),
34    "gpt-4o":       ModelPrice(2.5, 10, 1.25, 0),
35    "gpt-4o-mini":  ModelPrice(0.15, 0.6, 0.075, 0),
36    "o3":           ModelPrice(2, 8, 0.5, 0),
37    "o4-mini":      ModelPrice(1.1, 4.4, 0.275, 0),
38
39    # --- Google (Gemini) ---
40    "gemini-3.5-pro":   ModelPrice(2.5, 15, 0.25, 0),
41    "gemini-3.5-flash": ModelPrice(0.3, 2.5, 0.03, 0),
42    "gemini-2.5-pro":   ModelPrice(1.25, 10, 0.125, 0),
43    "gemini-2.5-flash": ModelPrice(0.3, 2.5, 0.03, 0),
44}
45
46# When a log says "claude-opus-4-6-20260514" or "anthropic/claude-sonnet-4-6" we
47# won't find an exact key, so we try progressively looser matches. Longest keys
48# first, so "claude-opus-4-6" wins over "claude-opus-4".
49_KEYS_BY_LENGTH = sorted(TABLE.keys(), key=len, reverse=True)
50
51_unknown_models: set[str] = set()
52
53
54def price_for(model: str) -> ModelPrice | None:
55    if not model:
56        return None
57    m = model.lower()
58
59    if m in TABLE:
60        return TABLE[m]
61
62    # strip a leading vendor prefix like "anthropic/" or "openai/"
63    bare = m.split("/")[-1] if "/" in m else m
64    if bare in TABLE:
65        return TABLE[bare]
66
67    # substring match — handles date-stamped and suffixed names
68    for key in _KEYS_BY_LENGTH:
69        if key in bare:
70            return TABLE[key]
71
72    _unknown_models.add(model)
73    return None
74
75
76def unpriced_models() -> list[str]:
77    """Models we saw but couldn't price. The CLI warns so $0 rows aren't a silent lie."""
78    return sorted(_unknown_models)
79
80
81def cost_of(tokens: TokenCounts, model: str) -> float:
82    """Cost in dollars for one set of token counts on a given model."""
83    p = price_for(model)
84    if p is None:
85        return 0.0
86    return (
87        tokens.input * p.input
88        + tokens.output * p.output
89        + tokens.cache_read * p.cache_read
90        + tokens.cache_write * p.cache_write
§ 01 · CORE IMPL

核心实现,逐段拆

▸报表聚合器

在 sources/tokentab/tokentab/report/aggregate.py:102 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/aggregate.py:110 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/aggregate.py:124 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/aggregate.py:140 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/aggregate.py:162 这一处,页面把源码证据和设计判断绑定起来。这一段的关键不是单个语句,而是它和上下游契约如何相互约束:输入必须先被压到统一字段,后续 pricing、aggregate、serialize 才能保持无状态且可复用。

sources/tokentab/tokentab/report/aggregate.py · L102-175Python
102def build_report(
103    records: list[PricedRecord],
104    rng: DateRange,
105    project: str | None = None,
106    exclude: list[str] | None = None,
107) -> Report:
108    exclude = exclude or []
109
110    def keep(r: PricedRecord) -> bool:
111        ts = r.timestamp
112        if ts.tzinfo is None:
113            ts = ts.replace(tzinfo=timezone.utc)
114        if ts < rng.frm or ts > rng.to:
115            return False
116        if project and project.lower() not in r.project.lower():
117            return False
118        if any(x.lower() in r.project.lower() for x in exclude):
119            return False
120        return True
121
122    in_range = [r for r in records if keep(r)]
123
124    totals = Totals(calls=len(in_range))
125    tools: dict[str, Row] = {}
126    models: dict[str, Row] = {}
127    projects: dict[str, Row] = {}
128    activities: dict[str, Row] = {}
129    days: dict[str, DayRow] = {}
130
131    def bump(m: dict[str, Row], key: str, r: PricedRecord) -> None:
132        row = m.get(key)
133        if row is None:
134            row = Row(label=key)
135            m[key] = row
136        row.cost += r.cost
137        row.tokens += r.tokens.total()
138        row.calls += 1
139
140    for r in in_range:
141        totals.cost += r.cost
142        totals.tokens.input += r.tokens.input
143        totals.tokens.output += r.tokens.output
144        totals.tokens.cache_read += r.tokens.cache_read
145        totals.tokens.cache_write += r.tokens.cache_write
146
147        bump(tools, r.provider, r)
148        bump(models, r.model, r)
149        bump(projects, r.project, r)
150        bump(activities, r.activity, r)
151
152        ts = r.timestamp if r.timestamp.tzinfo else r.timestamp.replace(tzinfo=timezone.utc)
153        key = _day_key(ts)
154        d = days.get(key)
155        if d is None:
156            d = DayRow(date=key)
157            days[key] = d
158        d.cost += r.cost
159        d.tokens += r.tokens.total()
160        d.calls += 1
161
162    input_side = totals.tokens.input + totals.tokens.cache_read
163    totals.cache_hit_rate = (totals.tokens.cache_read / input_side) if input_side > 0 else 0.0
164
165    by_cost = lambda rows: sorted(rows, key=lambda x: x.cost, reverse=True)
166
167    return Report(
168        range=rng,
169        totals=totals,
170        by_tool=by_cost(tools.values()),
171        by_model=by_cost(models.values()),
172        by_project=by_cost(projects.values()),
173        by_activity=by_cost(activities.values()),
174        by_day=sorted(days.values(), key=lambda d: d.date),
175    )

▸活动分类规则

在 sources/tokentab/tokentab/classify.py:15 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/classify.py:25 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/classify.py:31 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/classify.py:38 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/classify.py:47 这一处,页面把源码证据和设计判断绑定起来。这一段的关键不是单个语句,而是它和上下游契约如何相互约束:输入必须先被压到统一字段,后续 pricing、aggregate、serialize 才能保持无状态且可复用。

sources/tokentab/tokentab/classify.py · L15-51Python
15def classify(tool_names: list[str], user_text: str) -> str:
16    tools = [t.lower() for t in tool_names]
17    text = user_text.lower()
18
19    def has(tool_set: set[str]) -> bool:
20        return any(t in tool_set for t in tools)
21
22    def said(*words: str) -> bool:
23        return any(w in text for w in words)
24
25    edited = has(_EDIT_TOOLS)
26    ran_shell = any(t in ("bash", "shell", "run") for t in tools)
27
28    if ran_shell and said("test", "pytest", "vitest", "jest", "npm test", "go test"):
29        return "testing"
30
31    if said("commit", "push", "merge", "rebase", "pull request", "git "):
32        return "git"
33
34    if said("plan", "how should", "what's the best way", "approach", "design the"):
35        if not edited:
36            return "planning"
37
38    if edited:
39        if said("bug", "fix", "error", "crash", "broken", "doesn't work", "not working"):
40            return "debugging"
41        if said("refactor", "rename", "clean up", "simplify", "tidy", "extract"):
42            return "refactor"
43        if said("add", "create", "implement", "build", "new feature", "support for"):
44            return "feature"
45        return "coding"
46
47    if has(_READ_TOOLS):
48        return "exploring"
49    if not tools:
50        return "chat"
51    return "general"

▸JSON 序列化形状

在 sources/tokentab/tokentab/report/serialize.py:12 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:20 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:30 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:34 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:38 这一处,页面把源码证据和设计判断绑定起来。这一段的关键不是单个语句,而是它和上下游契约如何相互约束:输入必须先被压到统一字段,后续 pricing、aggregate、serialize 才能保持无状态且可复用。

sources/tokentab/tokentab/report/serialize.py · L12-39Python
12def report_to_dict(report: Report) -> dict:
13    t = report.totals
14    return {
15        "range": {
16            "from": report.range.frm.isoformat(),
17            "to": report.range.to.isoformat(),
18        },
19        "totals": {
20            "cost": t.cost,
21            "tokens": {
22                "input": t.tokens.input,
23                "output": t.tokens.output,
24                "cacheRead": t.tokens.cache_read,
25                "cacheWrite": t.tokens.cache_write,
26            },
27            "calls": t.calls,
28            "cacheHitRate": t.cache_hit_rate,
29        },
30        "byTool": [_row(r) for r in report.by_tool],
31        "byModel": [_row(r) for r in report.by_model],
32        "byProject": [_row(r) for r in report.by_project],
33        "byActivity": [_row(r) for r in report.by_activity],
34        "byDay": [{"date": d.date, "cost": d.cost, "tokens": d.tokens, "calls": d.calls} for d in report.by_day],
35    }
36
37
38def _row(r) -> dict:
39    return {"label": r.label, "cost": r.cost, "tokens": r.tokens, "calls": r.calls}
§ 02 · KEY FLOW

关键流程

disk logs → tolerant parser → UsageRecord → PricedRecord → grouped Report → CLI table / browser invoice

在 sources/tokentab/tokentab/report/serialize.py:12 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:20 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:30 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:34 这一处,页面把源码证据和设计判断绑定起来;在 sources/tokentab/tokentab/report/serialize.py:38 这一处,页面把源码证据和设计判断绑定起来。这条流程的优点是每层只承认一种输入形状;风险是入口文件一旦没有走到这个链路,底层解析器再干净也无法被用户正常调用。

§ 03 · EDGE

边界与失败

边界主要集中在日志格式漂移、模型价格表滞后、缓存 token 含义不一致和 provider 文件不可读。源码用 try_json、walk_files、provider try/except 和 unknown model warning 做了软失败处理;这适合本地观测工具,因为用户更需要“尽量给出已有数据”,而不是因为一个工具坏了让整份账单失败。