tokentab pricing aggregation
Pricing aggregation
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 这一处,页面把源码证据和设计判断绑定起来。
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
核心实现,逐段拆
▸报表聚合器
在 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 才能保持无状态且可复用。
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 才能保持无状态且可复用。
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 才能保持无状态且可复用。
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}
关键流程
在 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 这一处,页面把源码证据和设计判断绑定起来。这条流程的优点是每层只承认一种输入形状;风险是入口文件一旦没有走到这个链路,底层解析器再干净也无法被用户正常调用。
边界与失败
边界主要集中在日志格式漂移、模型价格表滞后、缓存 token 含义不一致和 provider 文件不可读。源码用 try_json、walk_files、provider try/except 和 unknown model warning 做了软失败处理;这适合本地观测工具,因为用户更需要“尽量给出已有数据”,而不是因为一个工具坏了让整份账单失败。