How to estimate monthly LLM API costs before you scale

Estimating monthly LLM API costs before you scale requires more than multiplying a per-token price by a guess. You need a model that accounts for input/output token ratios, caching behavior, retry overhead, provider fallback, and the inevitable growth in context length as features expand. This guide walks through building a defensible cost model you can validate against real traffic.

Step 1: Instrument your current token usage

You cannot estimate what you do not measure. Add token counting to every LLM call in your application — both the request you send and the response you receive. Most providers return usage in the response body; capture it.

# utils/token_tracker.py
from dataclasses import dataclass
from typing import Optional
import tiktoken

@dataclass
class TokenUsage:
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    model: str
    cached_prompt_tokens: int = 0  # For providers that report cache hits

class TokenTracker:
    def __init__(self):
        self._encoders = {}
    
    def _get_encoder(self, model: str):
        if model not in self._encoders:
            try:
                self._encoders[model] = tiktoken.encoding_for_model(model)
            except KeyError:
                # Fallback for unknown models
                self._encoders[model] = tiktoken.get_encoding("cl100k_base")
        return self._encoders[model]
    
    def count_tokens(self, text: str, model: str) -> int:
        return len(self._get_encoder(model).encode(text))
    
    def estimate_request_tokens(self, messages: list[dict], model: str) -> int:
        """Approximate tokens for a chat completion request."""
        encoder = self._get_encoder(model)
        total = 0
        for msg in messages:
            total += 4  # message overhead
            for key, value in msg.items():
                total += len(encoder.encode(str(value)))
        total += 2  # assistant primer
        return total

Wrap your LLM client to log usage automatically:

# llm/client_wrapper.py
import time
import logging
from typing import Any
from openai import OpenAI
from utils.token_tracker import TokenTracker, TokenUsage

logger = logging.getLogger(__name__)

class TrackedOpenAIClient:
    def __init__(self, api_key: str, base_url: str | None = None):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.tracker = TokenTracker()
    
    def chat_completion(self, **kwargs) -> Any:
        model = kwargs.get("model", "gpt-4o-mini")
        messages = kwargs.get("messages", [])
        
        # Estimate input tokens before sending
        est_input = self.tracker.estimate_request_tokens(messages, model)
        
        start = time.perf_counter()
        response = self.client.chat.completions.create(**kwargs)
        latency_ms = (time.perf_counter() - start) * 1000
        
        # Extract actual usage from response
        usage = response.usage
        actual_usage = TokenUsage(
            prompt_tokens=usage.prompt_tokens,
            completion_tokens=usage.completion_tokens,
            total_tokens=usage.total_tokens,
            model=model,
            cached_prompt_tokens=getattr(usage, "prompt_tokens_details", {}).get("cached_tokens", 0)
        )
        
        logger.info(
            "llm_call",
            extra={
                "model": model,
                "est_input_tokens": est_input,
                "actual_input_tokens": actual_usage.prompt_tokens,
                "output_tokens": actual_usage.completion_tokens,
                "cached_tokens": actual_usage.cached_prompt_tokens,
                "latency_ms": latency_ms,
                "finish_reason": response.choices[0].finish_reason,
            }
        )
        return response

Verify success: Deploy this wrapper for one week. Confirm your logs show actual_input_tokens, output_tokens, and cached_tokens for every call. The ratio of estimated to actual input tokens should be within 10% for standard chat formats.

Step 2: Calculate your token profile per feature

Aggregate the logged data by feature or user flow. You need the median and p90 for:

• Input tokens per request (split cached vs. uncached)
• Output tokens per request
• Requests per active user per day

# analysis/token_profile.py
import pandas as pd
import json
from pathlib import Path

def load_logs(log_path: str) -> pd.DataFrame:
    records = []
    with open(log_path) as f:
        for line in f:
            try:
                records.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return pd.DataFrame(records)

def compute_profile(df: pd.DataFrame) -> dict:
    # Filter successful calls
    df = df[df["finish_reason"] == "stop"].copy()
    
    profile = {}
    for feature in df["feature"].unique():
        subset = df[df["feature"] == feature]
        profile[feature] = {
            "requests_per_day_per_user": len(subset) / subset["user_id"].nunique() / 7,
            "input_tokens": {
                "median": subset["actual_input_tokens"].median(),
                "p90": subset["actual_input_tokens"].quantile(0.9),
                "cached_median": subset["cached_tokens"].median(),
                "cached_p90": subset["cached_tokens"].quantile(0.9),
            },
            "output_tokens": {
                "median": subset["output_tokens"].median(),
                "p90": subset["output_tokens"].quantile(0.9),
            },
            "uncached_input_median": subset["actual_input_tokens"].median() - subset["cached_tokens"].median(),
        }
    return profile

if __name__ == "__main__":
    df = load_logs("logs/llm_usage.jsonl")
    profile = compute_profile(df)
    print(json.dumps(profile, indent=2, default=str))

Verify success: Run this against a week of production logs. Each feature should have a complete profile with non-zero request rates. If a feature shows zero cached tokens, verify your provider supports prompt caching and that your prompts have stable prefixes.

Step 3: Build a pricing matrix per model and provider

Create a maintainable pricing table. Include input, output, and cached input prices. Providers change prices quarterly; version this file.

// config/pricing.json
{
  "version": "2025-01-15",
  "providers": {
    "openai": {
      "gpt-4o": {
        "input_per_1m": 2.50,
        "output_per_1m": 10.00,
        "cached_input_per_1m": 1.25
      },
      "gpt-4o-mini": {
        "input_per_1m": 0.15,
        "output_per_1m": 0.60,
        "cached_input_per_1m": 0.075
      },
      "o1-preview": {
        "input_per_1m": 15.00,
        "output_per_1m": 60.00,
        "cached_input_per_1m": 7.50
      }
    },
    "anthropic": {
      "claude-3-5-sonnet-20241022": {
        "input_per_1m": 3.00,
        "output_per_1m": 15.00,
        "cached_input_per_1m": 0.30
      },
      "claude-3-5-haiku-20241022": {
        "input_per_1m": 0.25,
        "output_per_1m": 1.25,
        "cached_input_per_1m": 0.025
      }
    }
  }
}

Load it into a calculator:

# analysis/cost_calculator.py
import json
from dataclasses import dataclass
from typing import Literal

@dataclass
class ModelPricing:
    input_per_1m: float
    output_per_1m: float
    cached_input_per_1m: float

class PricingMatrix:
    def __init__(self, path: str = "config/pricing.json"):
        with open(path) as f:
            data = json.load(f)
        self.providers = {}
        for provider, models in data["providers"].items():
            self.providers[provider] = {
                model: ModelPricing(**pricing) for model, pricing in models.items()
            }
    
    def get_pricing(self, provider: str, model: str) -> ModelPricing:
        return self.providers[provider][model]
    
    def cost_per_request(
        self,
        provider: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        cached_tokens: int = 0
    ) -> float:
        pricing = self.get_pricing(provider, model)
        uncached_input = input_tokens - cached_tokens
        return (
            (uncached_input * pricing.input_per_1m / 1_000_000) +
            (cached_tokens * pricing.cached_input_per_1m / 1_000_000) +
            (output_tokens * pricing.output_per_1m / 1_000_000)
        )

Verify success: Unit test the calculator against known examples. For instance, 1,000 input tokens (100 cached) + 500 output tokens on gpt-4o-mini should equal $0.000525.

Step 4: Model monthly cost at multiple growth scenarios

Combine your token profile with pricing across three scenarios: conservative (median tokens, current MAU), expected (p90 tokens, 3x MAU), and stress (p90 tokens, 10x MAU + 20% context growth).

# analysis/monthly_projection.py
from dataclasses import dataclass
from typing import Literal
from analysis.cost_calculator import PricingMatrix, ModelPricing

@dataclass
class Scenario:
    name: str
    mau_multiplier: float
    token_percentile: Literal["median", "p90"]
    context_growth_factor: float = 1.0

SCENARIOS = [
    Scenario("conservative", 1.0, "median", 1.0),
    Scenario("expected", 3.0, "p90", 1.0),
    Scenario("stress", 10.0, "p90", 1.2),
]

def project_monthly_cost(
    token_profile: dict,
    pricing: PricingMatrix,
    provider: str,
    model: str,
    current_mau: int
) -> dict:
    results = {}
    for scenario in SCENARIOS:
        total = 0.0
        for feature, profile in token_profile.items():
            req_per_user_day = profile["requests_per_day_per_user"]
            input_tokens = profile["input_tokens"][scenario.token_percentile] * scenario.context_growth_factor
            output_tokens = profile["output_tokens"][scenario.token_percentile] * scenario.context_growth_factor
            cached_tokens = profile["input_tokens"][f"cached_{scenario.token_percentile}"] * scenario.context_growth_factor
            
            cost_per_req = pricing.cost_per_request(
                provider, model, input_tokens, output_tokens, cached_tokens
            )
            monthly_requests = req_per_user_day * current_mau * scenario.mau_multiplier * 30
            total += cost_per_req * monthly_requests
        results[scenario.name] = round(total, 2)
    return results

if __name__ == "__main__":
    from analysis.token_profile import compute_profile, load_logs
    
    df = load_logs("logs/llm_usage.jsonl")
    profile = compute_profile(df)
    pricing = PricingMatrix()
    
    projections = project_monthly_cost(
        token_profile=profile,
        pricing=pricing,
        provider="openai",
        model="gpt-4o-mini",
        current_mau=5000
    )
    
    for scenario, cost in projections.items():
        print(f"{scenario}: ${cost:,.2f}/month")

Verify success: Run projections for each model you’re considering. The spread between conservative and stress should feel uncomfortable but not absurd — if stress is 100x conservative, your token profile variance is too high or your MAU assumptions need tightening.

Step 5: Factor in infrastructure overhead

LLM API costs are only part of the bill. Add:

• Retry overhead: Failed calls consume tokens. Multiply by 1 / (1 - error_rate).
• Fallback chains: If you route from premium to cheaper models on degradation, weight the blend.
• Embedding costs: RAG pipelines add embedding calls per document chunk.
• Moderation/classifier calls: Often overlooked, these run on every request.

# analysis/overhead.py
def apply_overhead(base_cost: float, config: dict) -> float:
    """
    config keys:
    - error_rate: fraction of calls that retry (e.g., 0.02 for 2%)
    - fallback_blend: dict of {model: weight} for fallback routing
    - embedding_calls_per_request: float
    - embedding_model: str
    - moderation_calls_per_request: float (usually 1.0)
    """
    # Retry multiplier
    retry_multiplier = 1 / (1 - config.get("error_rate", 0.0))
    
    # Fallback blend (simplified: assume weighted average of model costs)
    fallback_multiplier = 1.0
    if "fallback_blend" in config:
        # This requires pricing for each fallback model
        # Simplified: assume fallback is 40% cheaper on average
        fallback_multiplier = sum(
            weight * 0.6 for weight in config["fallback_blend"].values()
        ) + (1 - sum(config["fallback_blend"].values())) * 1.0
    
    total = base_cost * retry_multiplier * fallback_multiplier
    
    # Add embedding costs
    if config.get("embedding_calls_per_request", 0) > 0:
        # Would need embedding pricing — placeholder
        embedding_cost_per_call = 0.0001  # text-embedding-3-small approx
        total += config["embedding_calls_per_request"] * embedding_cost_per_call * 30 * config.get("mau", 1000)
    
    # Add moderation
    if config.get("moderation_calls_per_request", 0) > 0:
        moderation_cost_per_call = 0.00005  # free on some providers
        total += config["moderation_calls_per_request"] * moderation_cost_per_call * 30 * config.get("mau", 1000)
    
    return round(total, 2)

Verify success: Compare your overhead-adjusted projection against last month’s actual bill (if you have one). The model should be within 15%. If not, audit which overhead factor is misestimated — usually retry rate or fallback blend.

Step 6: Set up continuous cost monitoring

A static projection rots. Build a dashboard that compares projected vs. actual daily spend, alerting when actual exceeds 120% of expected.

# monitoring/cost_alert.py
import os
from datetime import datetime, timedelta
from dataclasses import dataclass
from analysis.cost_calculator import PricingMatrix

@dataclass
class DailySpend:
    date: str
    projected: float
    actual: float
    variance_pct: float

class CostMonitor:
    def __init__(self, pricing: PricingMatrix, provider: str, model: str):
        self.pricing = pricing
        self.provider = provider
        self.model = model
        self.daily_budget = float(os.getenv("DAILY_LLM_BUDGET_USD", "100"))
    
    def record_actual_spend(self, date: str, usage_logs: list[dict]) -> DailySpend:
        actual = 0.0
        for log in usage_logs:
            actual += self.pricing.cost_per_request(
                self.provider,
                self.model,
                log["actual_input_tokens"],
                log["output_tokens"],
                log.get("cached_tokens", 0)
            )
        
        # Projected based on current MAU trend
        projected = self.daily_budget
        
        variance = ((actual - projected) / projected * 100) if projected > 0 else 0
        
        return DailySpend(
            date=date,
            projected=round(projected, 2),
            actual=round(actual, 2),
            variance_pct=round(variance, 1)
        )
    
    def check_alert(self, spend: DailySpend) -> bool:
        if spend.variance_pct > 20:
            # Send to PagerDuty, Slack, etc.
            print(f"ALERT: {spend.date} spend ${spend.actual} vs projected ${spend.projected} ({spend.variance_pct}% over)")
            return True
        return False

Verify success: Run this against the last 7 days of logs. The variance should hover near 0% with occasional spikes on high-traffic days. If variance is consistently positive, your token profile underestimates real usage — go back to Step 2.

Step 7: Validate against provider invoices

The ultimate test: reconcile your model against the actual invoice. Providers bill on slightly different token counting rules (some count whitespace differently, some include system prompt overhead). Do this monthly.

# analysis/invoice_reconciliation.py
def reconcile_invoice(
    invoice_path: str,
    usage_logs: list[dict],
    pricing: PricingMatrix,
    provider: str,
    model: str
) -> dict:
    """
    invoice_path: CSV or JSON from provider billing export
    usage_logs: your tracked calls for the same period
    """
    import csv
    
    # Parse invoice (format varies by provider)
    invoice_total = 0.0
    invoice_tokens = {"input": 0, "output": 0, "cached": 0}
    
    with open(invoice_path) as f:
        reader = csv.DictReader(f)
        for row in reader:
            invoice_total += float(row.get("amount", 0))
            invoice_tokens["input"] += int(row.get("prompt_tokens", 0))
            invoice_tokens["output"] += int(row.get("completion_tokens", 0))
            invoice_tokens["cached"] += int(row.get("cached_prompt_tokens", 0))
    
    # Calculate from our logs
    our_total = 0.0
    our_tokens = {"input": 0, "output": 0, "cached": 0}
    for log in usage_logs:
        our_total += pricing.cost_per_request(
            provider, model,
            log["actual_input_tokens"],
            log["output_tokens"],
            log.get("cached_tokens", 0)
        )
        our_tokens["input"] += log["actual_input_tokens"]
        our_tokens["output"] += log["output_tokens"]
        our_tokens["cached"] += log.get("cached_tokens", 0)
    
    return {
        "invoice_total": round(invoice_total, 2),
        "our_total": round(our_total, 2),
        "difference_pct": round((our_total - invoice_total) / invoice_total * 100, 1),
        "token_diff": {
            k: our_tokens[k] - invoice_tokens[k] for k in invoice_tokens
        }
    }

Verify success: Run reconciliation monthly. Target: within 5% on total cost, within 3% on token counts. Persistent gaps indicate:

• Unlogged calls (background jobs, batch processing)
• Provider counting differences (file uploads, image tokens, tool call overhead)
• Fallback routing you didn’t model

Putting it into practice

Start with Steps 1 and 2 this week. Deploy the tracker, let it run for 5-7 days, then generate your first token profile. The code above is deliberately framework-agnostic — adapt the logging format to your observability stack (Datadog, Honeycomb, CloudWatch, plain JSONL).

Once you have real token distributions, the rest becomes arithmetic. The engineers who skip instrumentation and guess at “about 2k tokens per request” are the ones explaining a $50k surprise invoice to finance. The ones who measure sleep better.

If you’re routing across multiple providers and want automatic fallback with unified usage metering, n4n.ai surfaces provider cache-control hints and honors client routing directives through a single OpenAI-compatible endpoint — which keeps your cost model honest even when traffic shifts between models.