Cost Attribution by Dimension

Capsule overview

Knowing that your AI system costs $2,400/month tells you nothing useful. It's like knowing your restaurant spent $10,000 on ingredients without knowing how much was meat, how much vegetables, and how much got thrown in the trash. To optimize, you need granularity: which endpoint generates the biggest spend? Which user consumes the most resources? Which model dominates the invoice? Embeddings or completions?

Cost attribution is the art of tagging every cent with the dimensions it came from. You don't change how much you spend — you change how much you understand about what you spend. And that understanding is what turns "I have to cut costs" into "I have to optimize the /summarize endpoint that generates 40% of the spend with only 15% of the traffic."

In this capsule you'll build a multidimensional cost attribution system. Every request to your AI system will be tagged with endpoint, user, model and operation type. By the end, you'll be able to identify your system's cost hotspots — that 20% of endpoints or users that generates 80% of the spend.

What is cost attribution?

From "how much" to "where"

Imagine you get the OpenAI invoice: $2,400. In Module 1 you learned to break it down by cost type: input tokens, output tokens, embeddings, retries. That answers how much in each category.

Cost attribution adds a completely different layer: where each cost originates. It's the difference between these two reports:

Report without attribution:
  Total: $2,400/mo
  - Completions: $1,500
  - Embeddings: $480
  - Retries: $300
  - Storage: $120

Report with attribution:
  Total: $2,400/mo

  By endpoint:
  - /chat:      $1,080 (45%)  ← cost hotspot
  - /summarize: $480  (20%)
  - /search:    $360  (15%)
  - /classify:  $240  (10%)
  - Others:     $240  (10%)

  By model:
  - GPT-4o:      $1,800 (75%)
  - GPT-4o-mini: $360   (15%)
  - Embeddings:  $240   (10%)

The first report tells you "I spend a lot on completions." The second tells you "the /chat endpoint on GPT-4o generates almost half my invoice." With the second one you can act.

The four fundamental dimensions

In an AI system, there are four dimensions that capture 95% of the attribution information you need:

Dimension        Question it answers             Example
──────────       ──────────────────────          ───────
Endpoint         Which feature spends most?      /chat vs /search vs /summarize
User             Who generates the most cost?    user_847 vs the average
Model            Which model dominates the bill? GPT-4o vs GPT-4o-mini
Operation type   Completions or embeddings?      completion vs embedding vs image

You can add more dimensions depending on your case (by team, by tenant, by region), but these four cover most scenarios.

Implementing the CostTracker

CostRecord and CostTracker

from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
import statistics


@dataclass
class CostRecord:
    """An individual cost record with all its dimensions."""
    timestamp: datetime
    endpoint: str
    user_id: str
    model: str
    operation_type: str
    input_tokens: int
    output_tokens: int
    cost: float
    metadata: dict = field(default_factory=dict)


PRICING_PER_1M = {
    "gpt-4o":      {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gpt-4-turbo": {"input": 10.00, "output": 30.00},
    "o3-mini":     {"input": 1.10, "output": 4.40},
    "text-embedding-3-small": {"input": 0.02, "output": 0.0},
    "text-embedding-3-large": {"input": 0.13, "output": 0.0},
}


class CostTracker:
    """Cost tracker with multidimensional attribution."""

    def __init__(self):
        self.records: list[CostRecord] = []

    def record(
        self,
        endpoint: str,
        user_id: str,
        model: str,
        operation_type: str,
        input_tokens: int,
        output_tokens: int = 0,
        metadata: dict | None = None,
    ) -> CostRecord:
        """Record an operation with its calculated cost."""
        pricing = PRICING_PER_1M.get(model)
        if not pricing:
            raise ValueError(f"Unsupported model: {model}")

        cost = (
            (input_tokens / 1_000_000) * pricing["input"]
            + (output_tokens / 1_000_000) * pricing["output"]
        )

        record = CostRecord(
            timestamp=datetime.now(),
            endpoint=endpoint,
            user_id=user_id,
            model=model,
            operation_type=operation_type,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost=round(cost, 8),
            metadata=metadata or {},
        )
        self.records.append(record)
        return record

    def total_cost(self) -> float:
        return round(sum(r.cost for r in self.records), 6)

    def cost_by(self, dimension: str) -> dict[str, float]:
        """Group costs by any dimension."""
        grouped: dict[str, float] = defaultdict(float)
        for r in self.records:
            key = getattr(r, dimension)
            grouped[key] += r.cost
        return dict(sorted(grouped.items(), key=lambda x: x[1], reverse=True))

    def cost_by_two(self, dim1: str, dim2: str) -> dict[str, dict[str, float]]:
        """Group costs by two dimensions (cross-tabulation)."""
        grouped: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
        for r in self.records:
            k1 = getattr(r, dim1)
            k2 = getattr(r, dim2)
            grouped[k1][k2] += r.cost
        return {k1: dict(v) for k1, v in grouped.items()}

tracker = CostTracker()

tracker.record("/chat", "user_847", "gpt-4o", "completion", 1500, 400)
tracker.record("/chat", "user_312", "gpt-4o", "completion", 1200, 350)
tracker.record("/summarize", "user_847", "gpt-4o", "completion", 3000, 800)
tracker.record("/search", "user_091", "gpt-4o-mini", "completion", 800, 200)
tracker.record("/search", "user_312", "text-embedding-3-small", "embedding", 500)
tracker.record("/classify", "user_558", "gpt-4o-mini", "completion", 200, 10)

print(f"Total cost: ${tracker.total_cost()}")
print(f"\nBy endpoint: {tracker.cost_by('endpoint')}")
print(f"\nBy user: {tracker.cost_by('user_id')}")
print(f"\nBy model: {tracker.cost_by('model')}")

# Expected output:
Total cost: $0.030036

By endpoint: {'/summarize': 0.0155, '/chat': 0.014249999999999999, '/search': 0.00025, '/classify': 3.6e-05}
By user: {'user_847': 0.02325, 'user_312': 0.006509999999999999, 'user_091': 0.00024, 'user_558': 3.6e-05}
By model: {'gpt-4o': 0.02975, 'gpt-4o-mini': 0.000276, 'text-embedding-3-small': 1e-05}

You can already see the patterns: /summarize is the most expensive endpoint despite having fewer requests, user_847 consumes 77% of the total spend, and GPT-4o dominates the invoice.

Cross-tabulation: crossing dimensions

Attribution by one dimension is useful. Crossing two dimensions is where the analysis becomes powerful.

cross = tracker.cost_by_two("endpoint", "model")
for endpoint, models in cross.items():
    print(f"\n{endpoint}:")
    for model, cost in models.items():
        print(f"  {model}: ${cost:.6f}")

# Expected output:
/chat:
  gpt-4o: $0.014250

/summarize:
  gpt-4o: $0.015500

/search:
  gpt-4o-mini: $0.000240
  text-embedding-3-small: $0.000010

/classify:
  gpt-4o-mini: $0.000036

Now you can see that /summarize uses GPT-4o exclusively. An immediate question: could /summarize work with GPT-4o-mini for certain document types? If the answer is yes, the savings are significant.

Automation: a cost tracking decorator

Recording costs manually is useful for understanding the concept, but in production you need automation. A decorator intercepts every API call and records the cost automatically.

import functools
from typing import Callable


def track_cost(
    tracker: CostTracker,
    endpoint: str,
    model: str = "gpt-4o",
    operation_type: str = "completion",
):
    """Decorator that automatically records the cost of a function."""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            user_id = kwargs.get("user_id", "anonymous")
            result = func(*args, **kwargs)

            input_tokens = result.get("usage", {}).get("prompt_tokens", 0)
            output_tokens = result.get("usage", {}).get("completion_tokens", 0)

            record = tracker.record(
                endpoint=endpoint, user_id=user_id, model=model,
                operation_type=operation_type,
                input_tokens=input_tokens, output_tokens=output_tokens,
            )
            result["_cost_record"] = {"cost": record.cost, "endpoint": endpoint}
            return result
        return wrapper
    return decorator


tracked_tracker = CostTracker()


@track_cost(tracked_tracker, endpoint="/chat", model="gpt-4o")
def chat_completion(message: str, user_id: str = "anonymous") -> dict:
    return {"content": f"Response for: {message}",
            "usage": {"prompt_tokens": 500, "completion_tokens": 200}}


@track_cost(tracked_tracker, endpoint="/summarize", model="gpt-4o")
def summarize_text(text: str, user_id: str = "anonymous") -> dict:
    return {"content": "Summary of the text...",
            "usage": {"prompt_tokens": 3000, "completion_tokens": 600}}


@track_cost(tracked_tracker, endpoint="/classify", model="gpt-4o-mini")
def classify_text(text: str, user_id: str = "anonymous") -> dict:
    return {"content": "positive",
            "usage": {"prompt_tokens": 200, "completion_tokens": 5}}


chat_completion("How does Python work?", user_id="user_100")
chat_completion("Explain decorators", user_id="user_200")
summarize_text("Long example text...", user_id="user_100")
classify_text("Great product", user_id="user_300")

print(f"Total: ${tracked_tracker.total_cost()}")
for endpoint, cost in tracked_tracker.cost_by("endpoint").items():
    pct = cost / tracked_tracker.total_cost() * 100
    print(f"  {endpoint}: ${cost:.6f} ({pct:.1f}%)")

# Expected output:
Total: $0.020033
  /summarize: $0.013500 (67.4%)
  /chat: $0.006500 (32.4%)
  /classify: $0.000033 (0.2%)

The decorator automatically captures the tokens reported by the API and records the cost with all its dimensions. You don't need to modify the business logic — you just add the decorator.

Identifying cost hotspots

The 80/20 rule in AI costs

In most AI systems, 20% of the endpoints (or users, or models) generate 80% of the cost. Identifying those hotspots is priority number one before optimizing.

def find_hotspots(
    tracker: CostTracker,
    dimension: str,
    threshold: float = 0.8,
) -> dict:
    """Identify the elements of a dimension that accumulate threshold% of the cost."""
    costs = tracker.cost_by(dimension)
    total = sum(costs.values())

    if total == 0:
        return {"hotspots": [], "coverage": 0, "total": 0}

    hotspots = []
    cumulative = 0.0

    for key, cost in costs.items():
        cumulative += cost
        hotspots.append({
            "key": key,
            "cost": round(cost, 6),
            "percentage": round(cost / total * 100, 1),
            "cumulative_pct": round(cumulative / total * 100, 1),
        })
        if cumulative / total >= threshold:
            break

    return {
        "hotspots": hotspots,
        "concentration": f"{len(hotspots)}/{len(costs)} generate {threshold*100:.0f}% of the cost",
        "total_cost": round(total, 6),
    }


result = find_hotspots(tracked_tracker, "endpoint")
print(f"Concentration: {result['concentration']}")
for h in result["hotspots"]:
    print(f"  {h['key']}: ${h['cost']} ({h['percentage']}%) — cumulative: {h['cumulative_pct']}%")

# Expected output:
Concentration: 2/3 generate 80% of the cost
  /summarize: $0.0135 (67.4%) — cumulative: 67.4%
  /chat: $0.0065 (32.4%) — cumulative: 99.8%

Now you know that optimizing /summarize and /chat would cover 99.8% of the potential impact. /classify is irrelevant for optimization: even if you removed it entirely, you'd save 0.2%.

Comparison: Flat tracking vs Multi-dimensional attribution

When flat tracking is enough

If your system has a single endpoint, a single model, and few users, flat tracking may be enough. But any system with more than 2-3 endpoints benefits from multidimensional attribution.

Connection to the project

Attributed data = actionable dashboards

The Cost Dashboard Integration (this module's project) needs data with dimensions to be useful. A dashboard that only shows "total cost per day" is a chart nobody looks at. A dashboard that shows "cost per endpoint with a weekly trend" generates decisions.

The CostTracker you built here produces exactly the data that capsule 03 (Designing Cost Dashboards) needs to create actionable visualizations. And find_hotspots is the function that identifies where to focus the optimization effort.

Connection to the baseline (capsule 07)

The formal cost baseline requires attribution. "My system costs $2,400/month" isn't a rigorous baseline. "My system costs $2,400/month, distributed 45% on /chat, 20% on /summarize, 15% on /search" — that's a baseline that lets you measure the impact of each individual optimization.

Troubleshooting

Problem 1: "The costs I calculate don't match my OpenAI invoice"

Cause: Your calculation uses the base prices per token, but the real invoice includes additional overhead. OpenAI charges cached input tokens at 50%, and there can be discrepancies from rounding at high volumes.

Fix: Use the calculated costs as a relative reference (to compare across endpoints/users) rather than as an absolute value. To reconcile with the invoice, add an adjustment factor of 1.05-1.10x.

Problem 2: "The user_id isn't always available in the request"

Cause: Unauthenticated requests or systems where the user ID arrives in inconsistent headers.

Fix: Define a fallback user_id. In the CostTracker, use "anonymous" as the default. If "anonymous" becomes a hotspot, you need to improve your user identification.

Problem 3: "I have too many endpoints, the report is unreadable"

Cause: APIs with many parameterized routes (/users/123/tasks/456) generate a unique key for every combination.

Fix: Normalize the paths before recording:

import re

def normalize_path(path: str) -> str:
    """Replace numeric IDs and UUIDs with placeholders."""
    path = re.sub(r'/\d+', '/{id}', path)
    path = re.sub(
        r'/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
        '/{uuid}', path
    )
    return path

print(normalize_path("/users/123/tasks/456"))
print(normalize_path("/docs/a1b2c3d4-e5f6-7890-abcd-ef1234567890/summary"))

# Expected output:
/users/{id}/tasks/{id}
/docs/{uuid}/summary

Problem 4: "The CostTracker eats a lot of memory with millions of records"

Cause: Storing every CostRecord in an in-memory list doesn't scale to production.

Fix: For development, the in-memory list is enough. For production, write the records to a database or metrics system (Prometheus counters, PostgreSQL, logs). The CostTracker becomes a facade that emits events instead of accumulating data.

Exercises

Exercise 1: Basic attribution with given data (Easy)

You have a system with this data from one day:

/chat:      200 requests, GPT-4o, avg 1000 input / 300 output
/search:    500 requests, GPT-4o-mini, avg 600 input / 150 output
/summarize: 50 requests, GPT-4o, avg 4000 input / 1000 output

Without code, calculate the daily cost per endpoint and determine which one is the hotspot.

/chat (GPT-4o: $2.50/1M input, $10.00/1M output):
  Input:  200 × 1000 = 200K tokens → 200K/1M × $2.50 = $0.50
  Output: 200 × 300 = 60K tokens → 60K/1M × $10.00 = $0.60
  Total: $1.10

/search (GPT-4o-mini: $0.15/1M input, $0.60/1M output):
  Input:  500 × 600 = 300K tokens → 300K/1M × $0.15 = $0.045
  Output: 500 × 150 = 75K tokens → 75K/1M × $0.60 = $0.045
  Total: $0.09

/summarize (GPT-4o: $2.50/1M input, $10.00/1M output):
  Input:  50 × 4000 = 200K tokens → 200K/1M × $2.50 = $0.50
  Output: 50 × 1000 = 50K tokens → 50K/1M × $10.00 = $0.50
  Total: $1.00

Hotspot: /chat and /summarize (95.9% of the cost).
/search has 2.5x more requests but costs 12x less thanks to GPT-4o-mini.

Exercise 2: Build an attribution report (Easy)

Use the CostTracker to record 8 mixed operations (at least 3 endpoints, 3 users, 2 models) and generate a report showing cost_by endpoint, user and model.

tracker = CostTracker()

tracker.record("/chat", "alice", "gpt-4o", "completion", 1000, 300)
tracker.record("/chat", "bob", "gpt-4o", "completion", 1200, 400)
tracker.record("/chat", "alice", "gpt-4o", "completion", 900, 250)
tracker.record("/summarize", "charlie", "gpt-4o", "completion", 5000, 1200)
tracker.record("/summarize", "alice", "gpt-4o", "completion", 4000, 900)
tracker.record("/search", "bob", "gpt-4o-mini", "completion", 500, 100)
tracker.record("/search", "charlie", "gpt-4o-mini", "completion", 600, 120)
tracker.record("/classify", "bob", "gpt-4o-mini", "completion", 200, 5)

total = tracker.total_cost()
print(f"=== Cost Attribution Report ===\nTotal: ${total}\n")

for dim_name, dim_field in [("Endpoint", "endpoint"), ("User", "user_id"), ("Model", "model")]:
    print(f"--- By {dim_name} ---")
    for key, cost in tracker.cost_by(dim_field).items():
        print(f"  {key}: ${cost:.6f} ({cost/total*100:.1f}%)")
    print()

Exercise 3: Cross-tabulation endpoint × model (Medium)

Use cost_by_two("endpoint", "model") to generate a cross table from the previous exercise. Identify the most expensive combination and propose an optimization.

cross = tracker.cost_by_two("endpoint", "model")
total = tracker.total_cost()

print(f"{'Endpoint':<15} {'Model':<25} {'Cost':>12} {'%':>8}")
print("-" * 62)

rows = []
for ep, models in cross.items():
    for model, cost in models.items():
        rows.append((ep, model, cost))
rows.sort(key=lambda x: x[2], reverse=True)

for ep, model, cost in rows:
    print(f"{ep:<15} {model:<25} ${cost:>10.6f} {cost/total*100:>6.1f}%")

print(f"\n🔥 Hotspot: {rows[0][0]} + {rows[0][1]}")
print(f"Optimization: Evaluate whether {rows[0][0]} can use GPT-4o-mini for simple requests.")

Exercise 4: Simulate a day and detect anomalies (Hard)

Simulate 100 requests spread across 4 endpoints. Make one user (user_999) generate 30 expensive requests to the /summarize endpoint with GPT-4o (simulating abuse). Use find_hotspots to identify them.

import random

random.seed(42)
sim_tracker = CostTracker()

normal_users = [f"user_{i:03d}" for i in range(1, 21)]
endpoints = [
    ("/chat", "gpt-4o", 1000, 300),
    ("/search", "gpt-4o-mini", 500, 100),
    ("/summarize", "gpt-4o", 3000, 800),
    ("/classify", "gpt-4o-mini", 200, 5),
]

for _ in range(70):
    ep, model, inp, out = random.choice(endpoints)
    user = random.choice(normal_users)
    sim_tracker.record(ep, user, model, "completion",
                       int(inp * random.uniform(0.8, 1.2)),
                       int(out * random.uniform(0.8, 1.2)))

for _ in range(30):
    sim_tracker.record("/summarize", "user_999", "gpt-4o", "completion",
                       int(4000 * random.uniform(0.9, 1.1)),
                       int(1000 * random.uniform(0.9, 1.1)))

total = sim_tracker.total_cost()
print(f"Total: ${total:.4f} ({len(sim_tracker.records)} requests)\n")

hotspots = find_hotspots(sim_tracker, "user_id", threshold=0.5)
print(f"Concentration (50% of the cost): {hotspots['concentration']}")
for h in hotspots["hotspots"]:
    print(f"  {h['key']}: ${h['cost']} ({h['percentage']}%)")

user_costs = sim_tracker.cost_by("user_id")
values = list(user_costs.values())
avg = statistics.mean(values)
std = statistics.stdev(values) if len(values) > 1 else 0

print(f"\n🚨 Anomalies (> avg + 2*std = ${avg + 2*std:.6f}):")
for user, cost in user_costs.items():
    if cost > avg + 2 * std:
        print(f"  {user}: ${cost:.6f} ({cost/avg:.1f}x the average)")

Summary

• ✅ Cost attribution turns "how much do I spend" into "where do I spend" — the difference between a useless number and an optimization map
• ✅ The four fundamental dimensions (endpoint, user, model, operation type) capture 95% of the information you need
• ✅ The CostTracker records every request with its calculated cost and all its dimensions
• ✅ Cross-tabulation (crossing two dimensions) reveals patterns a single dimension doesn't show
• ✅ Decorators automate the tracking without modifying the business logic
• ✅ The 80/20 rule applies consistently: a few endpoints/users generate most of the cost
• ✅ find_hotspots automatically identifies the elements that accumulate X% of the cost
• ✅ Normalizing parameterized paths is essential for useful attribution in RESTful APIs

Next capsule: Designing Cost Dashboards — turning this attributed data into visualizations that generate optimization decisions.

Additional resources

1. OpenAI Usage API - Official API to get usage and cost data
2. OpenAI Pricing - Up-to-date prices to validate your calculations
3. Prometheus Python Client - To expose cost metrics as counters
4. LiteLLM Cost Tracking - Multi-provider cost tracking
5. Pareto Principle in Software (Martin Fowler) - The 80/20 rule in software
6. FastAPI Middleware - Official middleware documentation in FastAPI
7. Datadog AI Cost Tracking - Managed alternative for cost attribution
8. OpenTelemetry Python - Open standard for instrumentation with cost dimensions

Created: March 2026 / Version: 1.0