Coverage for src/keel/cost.py: 100%
86 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Keel Analytics, Token Usage, and USD Cost Estimation Engine.
3Computes exact token expenditures and estimated USD costs per issue, PR, and Swarm wave
4using built-in model pricing tables without any external billing APIs.
5"""
7from __future__ import annotations
9from dataclasses import dataclass
10from pathlib import Path
11from typing import Any
13from . import activity
15# Model pricing in USD per 1,000,000 tokens: (prompt_price_per_m, completion_price_per_m)
16MODEL_PRICING: dict[str, tuple[float, float]] = {
17 # Anthropic
18 "claude-3-7-sonnet": (3.00, 15.00),
19 "claude-3-5-sonnet": (3.00, 15.00),
20 "claude-3-5-haiku": (0.80, 4.00),
21 "claude-3-opus": (15.00, 75.00),
22 "claude": (3.00, 15.00),
23 # Google Gemini
24 "gemini-2.5-pro": (1.25, 5.00),
25 "gemini-2.5-flash": (0.15, 0.60),
26 "gemini-1.5-pro": (1.25, 5.00),
27 "gemini-1.5-flash": (0.075, 0.30),
28 "gemini": (0.15, 0.60),
29 # OpenAI
30 "gpt-4o": (2.50, 10.00),
31 "gpt-4o-mini": (0.15, 0.60),
32 "o1": (15.00, 60.00),
33 "o3-mini": (1.10, 4.40),
34 "codex": (2.50, 10.00),
35 # DeepSeek
36 "deepseek-chat": (0.27, 1.10),
37 "deepseek-reasoner": (0.55, 2.19),
38 "deepseek": (0.27, 1.10),
39 # Local
40 "ollama": (0.00, 0.00),
41 "local": (0.00, 0.00),
42}
44DEFAULT_FALLBACK_PRICE = (1.00, 3.00)
45FRONTIER_BENCHMARK_PRICE = (15.00, 75.00) # Used for computing savings vs Claude Opus/o1
48def normalize_model_name(model: str) -> str:
49 """Normalize vendor:model strings to model pricing keys."""
50 raw = model.lower().strip()
51 if ":" in raw:
52 raw = raw.split(":", 1)[1]
53 raw = raw.replace("/", "-")
54 for key in MODEL_PRICING:
55 if key in raw:
56 return key
57 return raw or "default"
60def estimate_token_cost(
61 prompt_tokens: int, completion_tokens: int, model: str = ""
62) -> float:
63 """Calculate estimated USD cost for token usage based on model pricing."""
64 key = normalize_model_name(model)
65 prompt_rate, completion_rate = MODEL_PRICING.get(key, DEFAULT_FALLBACK_PRICE)
66 cost = (prompt_tokens * prompt_rate + completion_tokens * completion_rate) / 1_000_000.0
67 return round(cost, 6)
70def estimate_benchmark_cost(prompt_tokens: int, completion_tokens: int) -> float:
71 """Calculate benchmark frontier model cost for computing tiered routing savings."""
72 p_rate, c_rate = FRONTIER_BENCHMARK_PRICE
73 return round((prompt_tokens * p_rate + completion_tokens * c_rate) / 1_000_000.0, 6)
76@dataclass(frozen=True)
77class CostReport:
78 total_runs: int
79 total_prompt_tokens: int
80 total_completion_tokens: int
81 total_tokens: int
82 total_cost_usd: float
83 estimated_savings_usd: float
84 model_breakdown: dict[str, dict[str, Any]]
85 top_performer: str | None
87 def to_dict(self) -> dict[str, Any]:
88 return {
89 "total_runs": self.total_runs,
90 "total_prompt_tokens": self.total_prompt_tokens,
91 "total_completion_tokens": self.total_completion_tokens,
92 "total_tokens": self.total_tokens,
93 "total_cost_usd": round(self.total_cost_usd, 4),
94 "estimated_savings_usd": round(self.estimated_savings_usd, 4),
95 "model_breakdown": self.model_breakdown,
96 "top_performer": self.top_performer,
97 }
100def calculate_cost_report(records: list[dict[str, Any]]) -> CostReport:
101 """Aggregate token metrics and compute USD costs from activity records."""
102 total_prompt = 0
103 total_completion = 0
104 total_actual_cost = 0.0
105 total_benchmark_cost = 0.0
106 models_data: dict[str, dict[str, Any]] = {}
108 for rec in records:
109 p_tok = int(rec.get("prompt_tokens") or 0)
110 c_tok = int(rec.get("completion_tokens") or 0)
111 model = rec.get("model") or "gemini-2.5-flash"
113 if p_tok == 0 and c_tok == 0:
114 # Synthetic conservative estimate per activity phase (1,500 prompt, 400 completion)
115 p_tok = 1500
116 c_tok = 400
118 total_prompt += p_tok
119 total_completion += c_tok
121 cost = estimate_token_cost(p_tok, c_tok, model)
122 bench = estimate_benchmark_cost(p_tok, c_tok)
124 total_actual_cost += cost
125 total_benchmark_cost += bench
127 m_key = normalize_model_name(model)
128 if m_key not in models_data:
129 models_data[m_key] = {
130 "runs": 0,
131 "prompt_tokens": 0,
132 "completion_tokens": 0,
133 "cost_usd": 0.0,
134 }
135 models_data[m_key]["runs"] += 1
136 models_data[m_key]["prompt_tokens"] += p_tok
137 models_data[m_key]["completion_tokens"] += c_tok
138 models_data[m_key]["cost_usd"] = round(models_data[m_key]["cost_usd"] + cost, 4)
140 total_tokens = total_prompt + total_completion
141 savings = max(0.0, total_benchmark_cost - total_actual_cost)
143 top_perf = None
144 if models_data:
145 # Top performer is model with most runs
146 top_perf = max(models_data.items(), key=lambda item: item[1]["runs"])[0]
148 return CostReport(
149 total_runs=len(records),
150 total_prompt_tokens=total_prompt,
151 total_completion_tokens=total_completion,
152 total_tokens=total_tokens,
153 total_cost_usd=round(total_actual_cost, 4),
154 estimated_savings_usd=round(savings, 4),
155 model_breakdown=models_data,
156 top_performer=top_perf,
157 )
160def render_cost_report(report: CostReport) -> str:
161 """Render human-readable markdown / CLI report."""
162 lines = [
163 "Keel Efficiency & Cost Ledger",
164 "────────────────────────────────────────────────────────",
165 f" Total Runs Tracked : {report.total_runs}",
166 f" Total Tokens : {report.total_tokens:,} "
167 f"(Prompt: {report.total_prompt_tokens:,} / "
168 f"Completion: {report.total_completion_tokens:,})",
169 f" Estimated Spend (USD) : ${report.total_cost_usd:.4f}",
170 f" Estimated Savings : ${report.estimated_savings_usd:.4f} "
171 "(via Tiered Routing & Local/Flash Models)",
172 ]
173 if report.top_performer:
174 lines.append(f" Top Dispatched Model : {report.top_performer}")
176 if report.model_breakdown:
177 lines.append("")
178 lines.append(" Model Breakdown:")
179 for m, stats in sorted(report.model_breakdown.items(), key=lambda x: -x[1]["runs"]):
180 lines.append(
181 f" - {m:<18}: {stats['runs']:>3} runs | "
182 f"{stats['prompt_tokens'] + stats['completion_tokens']:>9,} tokens | "
183 f"${stats['cost_usd']:.4f}"
184 )
186 return "\n".join(lines)
189def generate_cost_report(root: str | Path = ".") -> CostReport:
190 """Read all activity records from .keel/activity and compile CostReport."""
191 root_path = Path(root).resolve()
192 act_dir = root_path / activity.DEFAULT_ACTIVITY_DIR
193 records: list[dict[str, Any]] = []
195 if act_dir.exists() and act_dir.is_dir():
196 records = activity.read_all_activity(act_dir)
198 return calculate_cost_report(records)