"""Turn hardware prices, power draw and measured throughput into cost per million tokens. Purpose: the arithmetic that makes "cheaper" mean something. Every input is yours: the price you would actually pay, the working life you would actually plan for, how much of that life the machine is actually busy, what electricity costs where you live, and the throughput you measured yourself. The script contains no prices and no speeds, because a price the course looked up would be wrong for your country within a month and a speed it looked up would be wrong for your machine on the day. It reports cost per million output tokens for each machine you describe, and appends the whole calculation to the lab notebook so the inputs travel with the answer. Platform: all (spark, strix, mac, nvidia). Pure standard library. Minimum memory: none of consequence; this is arithmetic. Assumes: an inputs file in the shape of cost-inputs-example.json, holding one entry per machine or cluster you are comparing. Throughput figures come from your own measurements: on Track M from measure-pair.py in this part, elsewhere from the load generator in Part 9. Currency is whatever you type; the script never converts and never assumes one. Usage: python3 cost-per-million-tokens.py --inputs cost-inputs.json python3 cost-per-million-tokens.py --inputs cost-inputs.json --labbook labbook.md python3 cost-per-million-tokens.py --inputs cost-inputs.json --print """ from __future__ import annotations import argparse import json from datetime import date, datetime, timezone from pathlib import Path HERE = Path(__file__).resolve().parent HOURS_PER_YEAR = 8760 REQUIRED = ( "name", "purchase_price", "lifetime_years", "busy_fraction", "average_watts", "electricity_price_per_kwh", "output_tokens_per_second", ) def parse_args(): p = argparse.ArgumentParser(description="Cost per million output tokens") p.add_argument( "--inputs", default=str(HERE / "cost-inputs.json"), help="JSON file describing each machine or cluster being compared.", ) p.add_argument("--labbook", default=str(HERE / "labbook.md")) p.add_argument("--label", default="", help="A name for this comparison.") p.add_argument( "--print", dest="print_only", action="store_true", help="Show the result and write nothing.", ) return p.parse_args() def cost_per_million(entry: dict) -> dict: """One machine's cost per million output tokens, with every intermediate kept.""" price = float(entry["purchase_price"]) years = float(entry["lifetime_years"]) busy = float(entry["busy_fraction"]) watts = float(entry["average_watts"]) kwh_price = float(entry["electricity_price_per_kwh"]) tps = float(entry["output_tokens_per_second"]) if years <= 0 or busy <= 0 or tps <= 0: raise ValueError( f"{entry['name']}: lifetime_years, busy_fraction and " "output_tokens_per_second must all be greater than zero." ) if busy > 1: raise ValueError(f"{entry['name']}: busy_fraction is a fraction, not a percentage.") # Spread the purchase over the hours the machine is actually working. A machine # busy a tenth of the time carries ten times the capital cost per busy hour. busy_hours = years * HOURS_PER_YEAR * busy capital_per_busy_hour = price / busy_hours # Energy is charged for the busy hours too. Idle draw is deliberately excluded: # it is a cost of owning the machine, not a cost of the tokens. energy_per_busy_hour = (watts / 1000.0) * kwh_price tokens_per_busy_hour = tps * 3600.0 total_per_busy_hour = capital_per_busy_hour + energy_per_busy_hour per_million = total_per_busy_hour / tokens_per_busy_hour * 1_000_000 return { "name": entry["name"], "currency": entry.get("currency", "unstated"), "purchase_price": price, "lifetime_years": years, "busy_fraction": busy, "average_watts": watts, "electricity_price_per_kwh": kwh_price, "output_tokens_per_second": tps, "workload": entry.get("workload", "unstated"), "busy_hours_over_life": round(busy_hours, 1), "capital_cost_per_busy_hour": round(capital_per_busy_hour, 4), "energy_cost_per_busy_hour": round(energy_per_busy_hour, 4), "tokens_per_busy_hour": round(tokens_per_busy_hour, 1), "cost_per_million_output_tokens": round(per_million, 4), "capital_share": round(capital_per_busy_hour / total_per_busy_hour, 4), } def main() -> int: args = parse_args() path = Path(args.inputs) if not path.exists(): print(f"No {path}. Copy cost-inputs-example.json and fill it in.") return 1 payload = json.loads(path.read_text(encoding="utf-8")) machines = payload.get("machines", []) if not machines: print("The inputs file has no machines in it.") return 1 results = [] for entry in machines: missing = [k for k in REQUIRED if k not in entry] if missing: print(f"{entry.get('name', 'an entry')} is missing: {', '.join(missing)}") return 1 try: results.append(cost_per_million(entry)) except (ValueError, KeyError, TypeError) as exc: print(f"Cannot compute a cost for this entry: {exc}") print("Every value in cost-inputs-example.json is a placeholder; the") print("file has to be filled in with your own figures before it means") print("anything.") return 1 width = max(len(r["name"]) for r in results) print("") print(f"{'machine'.ljust(width)} cost per million output tokens capital share") for r in results: cost = f"{r['cost_per_million_output_tokens']:.2f} {r['currency']}" share = f"{r['capital_share'] * 100:.0f}%" print(f"{r['name'].ljust(width)} {cost.rjust(30)} {share.rjust(13)}") print("") print("Capital share is how much of the cost is the machine rather than the") print("electricity. A high share means the answer is mostly about how busy you") print("keep it, and a low share means it is mostly about how fast it is.") cheapest = min(results, key=lambda r: r["cost_per_million_output_tokens"]) print("") print(f"Cheapest per million output tokens, on these inputs: {cheapest['name']}") print("That is a result about this workload and these assumptions, and it changes") print("if you change the busy fraction, the lifetime or the workload.") record = { "lab": "part-21/reality-check-macs-replace-a-gpu-server", "record": "cost-per-million-tokens", "date": date.today().isoformat(), "recorded_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), "label": args.label, "machines": results, "cheapest": cheapest["name"], } text = json.dumps(record, sort_keys=True) if args.print_only: print("") print(text) return 0 with open(args.labbook, "a", encoding="utf-8") as fh: fh.write(text + "\n") print("") print(f"Appended to {args.labbook}") return 0 if __name__ == "__main__": raise SystemExit(main())