#!/usr/bin/env python3 """Cost per million tokens for a local model service. Purpose: turn measurements you made - watts at idle, watts under load, output tokens per second - plus your own electricity tariff and what the machine cost, into a cost per million tokens you can defend, and compare it against a hosted price you looked up yourself. Platform: all (spark, strix, mac, nvidia). Pure Python, no dependencies. Minimum memory: 8 GB, which is what the service being costed needs; this script needs almost nothing. Assumes: a JSON file of inputs in the shape of cost-inputs-example.json. Every number in it is yours: this script invents nothing and has no price of any kind built into it. Where a value is missing, the report says so rather than guessing. Usage: python3 cost-model.py --inputs cost-inputs.json python3 cost-model.py --inputs cost-inputs.json --labbook labbook.md python3 cost-model.py --inputs cost-inputs.json --json """ import argparse import json import sys from datetime import date from pathlib import Path def die(message): print(f"cost-model: {message}", file=sys.stderr) sys.exit(1) def require(mapping, key, where): if key not in mapping: die(f"{where} is missing the required field {key!r}.") return mapping[key] def money(value): return f"{value:,.4f}" def cost_workload(workload, machine, tariff, api_price): """Cost of one million output tokens for one workload on one machine. The arithmetic, written out so that you can check it by hand: hours per million tokens H = 1e6 / tokens_per_second / 3600 energy while working E = load_watts / 1000 * H * tariff marginal energy M = (load_watts - idle_watts) / 1000 * H * tariff capital C = purchase_price / (lifetime_years * service_hours) * H idle carried by the work I = idle_watts / 1000 * (powered_hours - service_hours) * tariff / service_hours * H E is what the work costs in electricity. M is what it costs over leaving the machine switched on doing nothing, which is the honest number when the machine was going to be on anyway. C and I are the costs of ownership, and they both divide by the hours the machine actually spends generating: a machine used for an hour a day carries the same purchase price as one used all day, so its hour is dearer. That is the whole lesson. """ where = f"workload {workload.get('id')!r}" machine_where = f"machine {machine.get('id')!r}" tps = float(require(workload, "output_tokens_per_second", where)) if tps <= 0: die(f"{where}: output_tokens_per_second must be above zero.") price = float(require(machine, "purchase_price", machine_where)) years = float(require(machine, "lifetime_years", machine_where)) duty = float(require(machine, "service_hours_per_year", machine_where)) idle_w = float(require(machine, "idle_watts", machine_where)) load_w = float(require(machine, "load_watts", machine_where)) powered = float(machine.get("powered_hours_per_year", duty)) if years <= 0 or duty <= 0: die(f"{machine_where}: lifetime_years and service_hours_per_year must be above zero.") if powered < duty: die(f"{machine_where}: powered_hours_per_year cannot be below service_hours_per_year.") if load_w < idle_w: die(f"{machine_where}: load_watts is below idle_watts; check which is which.") share = float(workload.get("share_of_machine", 1.0)) if not 0 < share <= 1: die(f"{where}: share_of_machine must be above zero and at most one.") hours = 1_000_000.0 / tps / 3600.0 kwh = load_w / 1000.0 * hours marginal_kwh = (load_w - idle_w) / 1000.0 * hours energy = kwh * tariff marginal = marginal_kwh * tariff capital = price / (years * duty) * hours * share idle = idle_w / 1000.0 * (powered - duty) * tariff / duty * hours * share total = energy + capital + idle row = { "id": workload.get("id"), "machine": machine.get("id"), "model": workload.get("model"), "quant": workload.get("quant"), "context_length": workload.get("context_length"), "concurrency": workload.get("concurrency"), "output_tokens_per_second": tps, "hours_per_million_tokens": hours, "kwh_per_million_tokens": kwh, "marginal_kwh_per_million_tokens": marginal_kwh, "marginal_cost_per_million": marginal, "energy_cost_per_million": energy, "capital_cost_per_million": capital, "idle_cost_per_million": idle, "total_cost_per_million": total, } if api_price is None: return row # Utilisation is the variable that decides this, so the useful answer is not "which is # cheaper today" but "how many hours a year would this machine have to generate for # the two to cost the same". Rearranging total(duty) = api_price for duty: # # total(duty) = E + (K / duty) - B where # K = H * share * (purchase_price / lifetime_years + idle_kW * tariff * powered) # B = H * share * idle_kW * tariff # # so duty = K / (api_price - E + B). A denominator at or below zero means no amount of # use gets there: the electricity alone already costs more than the hosted price. k = hours * share * (price / years + idle_w / 1000.0 * tariff * powered) b = hours * share * idle_w / 1000.0 * tariff denominator = api_price - energy + b break_even_hours = k / denominator if denominator > 0 else None row.update({ "api_price_per_million": api_price, "cheaper_on_total_cost": "local" if total < api_price else "hosted", "cheaper_on_marginal_cost": "local" if marginal < api_price else "hosted", "break_even_service_hours_per_year": break_even_hours, }) return row def render(rows, currency, tariff, api_meta): print(f"Cost per million output tokens, in {currency}, at {tariff} per kWh") print() header = (f"{'workload':<22}{'machine':<10}{'kWh/Mtok':>10}{'energy':>10}" f"{'capital':>10}{'idle':>10}{'total':>10}{'marginal':>10}") print(header) print("-" * len(header)) for r in rows: print(f"{str(r['id'])[:21]:<22}{str(r['machine'])[:9]:<10}" f"{r['kwh_per_million_tokens']:>10.3f}" f"{money(r['energy_cost_per_million']):>10}" f"{money(r['capital_cost_per_million']):>10}" f"{money(r['idle_cost_per_million']):>10}" f"{money(r['total_cost_per_million']):>10}" f"{money(r['marginal_cost_per_million']):>10}") print() if not any("api_price_per_million" in r for r in rows): print("No hosted price was supplied, so no comparison was made. Look one up on the") print("provider's own pricing page, write it and the date into api_comparison, and") print("run this again.") return source = (api_meta or {}).get("source", "not stated") checked = (api_meta or {}).get("checked_on", "not stated") print(f"Against a hosted price of {money(rows[0]['api_price_per_million'])} per million " f"output tokens ({source}, checked {checked}):") print() for r in rows: if "api_price_per_million" not in r: continue print(f" {r['id']}") print(f" cheaper counting everything: {r['cheaper_on_total_cost']}") print(f" cheaper counting only the extra kWh: {r['cheaper_on_marginal_cost']}") hours = r["break_even_service_hours_per_year"] if hours is None: print(" break-even: never. The electricity this workload burns already") print(" costs more than the hosted price.") elif hours > 8760: print(f" break-even: about {hours:,.0f} generating hours a year, which is") print(" more hours than a year has. At these numbers the") print(" machine cannot be used enough to pay for itself on") print(" this workload alone.") else: print(f" break-even: about {hours:,.0f} generating hours a year, out of") print(" the 8,760 a year contains.") print() def main(): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--inputs", required=True, help="JSON file of your own measurements") parser.add_argument("--labbook", default=None, help="append one JSON line per workload to this file") parser.add_argument("--json", dest="as_json", action="store_true", help="print the rows as JSON instead of a table") args = parser.parse_args() path = Path(args.inputs) if not path.is_file(): die(f"no such file: {path}") try: data = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: die(f"{path} is not valid JSON: {exc}") placeholder = bool(data.get("_placeholder", False)) tariff_block = require(data, "tariff", "the input file") currency = tariff_block.get("currency", "your currency") tariff = float(require(tariff_block, "price_per_kwh", "tariff")) machines = {m["id"]: m for m in require(data, "machines", "the input file")} if not machines: die("the input file lists no machines.") api_meta = data.get("api_comparison") or {} api_price = api_meta.get("price_per_million_output_tokens") api_price = None if api_price is None else float(api_price) rows = [] for workload in require(data, "workloads", "the input file"): machine_id = require(workload, "machine", f"workload {workload.get('id')!r}") if machine_id not in machines: die(f"workload {workload.get('id')!r} names machine {machine_id!r}, " "which is not in the machines list.") rows.append(cost_workload(workload, machines[machine_id], tariff, api_price)) if args.as_json: print(json.dumps({"currency": currency, "price_per_kwh": tariff, "rows": rows}, indent=2, sort_keys=True)) else: render(rows, currency, tariff, api_meta) if placeholder: print("WARNING: this input file is still the shipped example. Every number in it is") print(" a placeholder chosen to make the arithmetic visible, not a") print(" measurement of anything. Replace them with your own, then set") print(' "_placeholder" to false.') if not args.labbook: return if placeholder: die("refusing to write placeholder numbers into the lab notebook.") out = Path(args.labbook) with out.open("a", encoding="utf-8") as handle: for r in rows: record = dict(r) record.update({"lab": "part-23/capacity-planning", "currency": currency, "price_per_kwh": tariff, "recorded": date.today().isoformat()}) handle.write(json.dumps(record, sort_keys=True) + "\n") print(f"recorded {len(rows)} row(s) in {out}") if __name__ == "__main__": main()