Token Usage Analytics and Monitoring for Claude Code

Overview

Effective token usage monitoring is crucial for managing costs and optimizing performance in Claude Code applications. This guide provides implementation patterns for tracking, analyzing, and optimizing token consumption across single and multi-agent systems.

Token Economics (2025)

Token-to-Cost Mapping

const TOKEN_COSTS = {
  "claude-4-opus": {
    input: 15.00,      // per million tokens
    output: 75.00,     // per million tokens
    cacheWrite: 18.75, // 25% premium
    cacheRead: 1.50    // 90% discount
  },
  "claude-4-sonnet": {
    input: 3.00,
    output: 15.00,
    cacheWrite: 3.75,
    cacheRead: 0.30
  },
  "claude-3-5-haiku": {
    input: 0.80,
    output: 4.00,
    cacheWrite: 1.00,
    cacheRead: 0.08
  }
};

Implementation Patterns

1. Basic Token Tracking System

import json
import sqlite3
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
import anthropic
 
@dataclass
class TokenUsage:
    """Token usage record."""
    timestamp: str
    model: str
    request_id: str
    input_tokens: int
    output_tokens: int
    cache_creation_tokens: int
    cache_read_tokens: int
    total_cost: float
    latency_ms: int
    user_id: Optional[str] = None
    session_id: Optional[str] = None
    prompt_category: Optional[str] = None
 
class TokenAnalytics:
    """Comprehensive token usage tracking and analytics."""
    
    def __init__(self, db_path: str = "token_usage.db"):
        self.db_path = db_path
        self.client = None
        self._init_db()
        
    def _init_db(self):
        """Initialize SQLite database for token tracking."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS token_usage (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                model TEXT NOT NULL,
                request_id TEXT NOT NULL,
                input_tokens INTEGER NOT NULL,
                output_tokens INTEGER NOT NULL,
                cache_creation_tokens INTEGER DEFAULT 0,
                cache_read_tokens INTEGER DEFAULT 0,
                total_cost REAL NOT NULL,
                latency_ms INTEGER NOT NULL,
                user_id TEXT,
                session_id TEXT,
                prompt_category TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # Create indexes for common queries
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON token_usage(timestamp)")
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_user_id ON token_usage(user_id)")
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_model ON token_usage(model)")
        
        conn.commit()
        conn.close()
    
    def track_usage(
        self,
        response: anthropic.types.Message,
        latency_ms: int,
        user_id: Optional[str] = None,
        session_id: Optional[str] = None,
        prompt_category: Optional[str] = None
    ):
        """Track token usage from API response."""
        usage = response.usage
        model = response.model
        
        # Calculate costs
        model_costs = self._get_model_costs(model)
        input_cost = (usage.input_tokens / 1_000_000) * model_costs["input"]
        output_cost = (usage.output_tokens / 1_000_000) * model_costs["output"]
        
        cache_creation_cost = 0
        cache_read_cost = 0
        
        if hasattr(usage, 'cache_creation_input_tokens'):
            cache_creation_cost = (usage.cache_creation_input_tokens / 1_000_000) * model_costs["cacheWrite"]
        
        if hasattr(usage, 'cache_read_input_tokens'):
            cache_read_cost = (usage.cache_read_input_tokens / 1_000_000) * model_costs["cacheRead"]
        
        total_cost = input_cost + output_cost + cache_creation_cost + cache_read_cost
        
        # Create usage record
        usage_record = TokenUsage(
            timestamp=datetime.utcnow().isoformat(),
            model=model,
            request_id=response.id,
            input_tokens=usage.input_tokens,
            output_tokens=usage.output_tokens,
            cache_creation_tokens=getattr(usage, 'cache_creation_input_tokens', 0),
            cache_read_tokens=getattr(usage, 'cache_read_input_tokens', 0),
            total_cost=total_cost,
            latency_ms=latency_ms,
            user_id=user_id,
            session_id=session_id,
            prompt_category=prompt_category
        )
        
        self._save_usage(usage_record)
        return usage_record
    
    def _save_usage(self, usage: TokenUsage):
        """Save usage record to database."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            INSERT INTO token_usage (
                timestamp, model, request_id, input_tokens, output_tokens,
                cache_creation_tokens, cache_read_tokens, total_cost, latency_ms,
                user_id, session_id, prompt_category
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            usage.timestamp, usage.model, usage.request_id,
            usage.input_tokens, usage.output_tokens,
            usage.cache_creation_tokens, usage.cache_read_tokens,
            usage.total_cost, usage.latency_ms,
            usage.user_id, usage.session_id, usage.prompt_category
        ))
        
        conn.commit()
        conn.close()
    
    def _get_model_costs(self, model: str) -> Dict[str, float]:
        """Get cost structure for a model."""
        # Simplified model mapping
        if "opus" in model:
            return {"input": 15.00, "output": 75.00, "cacheWrite": 18.75, "cacheRead": 1.50}
        elif "sonnet" in model:
            return {"input": 3.00, "output": 15.00, "cacheWrite": 3.75, "cacheRead": 0.30}
        elif "haiku" in model:
            return {"input": 0.80, "output": 4.00, "cacheWrite": 1.00, "cacheRead": 0.08}
        else:
            return {"input": 3.00, "output": 15.00, "cacheWrite": 3.75, "cacheRead": 0.30}
    
    def get_usage_summary(
        self,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        user_id: Optional[str] = None
    ) -> Dict[str, any]:
        """Get usage summary statistics."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Build query
        query = "SELECT * FROM token_usage WHERE 1=1"
        params = []
        
        if start_date:
            query += " AND timestamp >= ?"
            params.append(start_date)
        
        if end_date:
            query += " AND timestamp <= ?"
            params.append(end_date)
        
        if user_id:
            query += " AND user_id = ?"
            params.append(user_id)
        
        cursor.execute(query, params)
        rows = cursor.fetchall()
        
        # Calculate statistics
        total_requests = len(rows)
        total_input_tokens = sum(row[4] for row in rows)
        total_output_tokens = sum(row[5] for row in rows)
        total_cache_tokens = sum(row[6] + row[7] for row in rows)
        total_cost = sum(row[8] for row in rows)
        avg_latency = sum(row[9] for row in rows) / total_requests if total_requests > 0 else 0
        
        # Model breakdown
        model_usage = {}
        for row in rows:
            model = row[2]
            if model not in model_usage:
                model_usage[model] = {"requests": 0, "cost": 0}
            model_usage[model]["requests"] += 1
            model_usage[model]["cost"] += row[8]
        
        conn.close()
        
        return {
            "total_requests": total_requests,
            "total_input_tokens": total_input_tokens,
            "total_output_tokens": total_output_tokens,
            "total_cache_tokens": total_cache_tokens,
            "total_cost": round(total_cost, 2),
            "average_latency_ms": round(avg_latency),
            "cost_per_request": round(total_cost / total_requests, 4) if total_requests > 0 else 0,
            "cache_hit_rate": round(total_cache_tokens / total_input_tokens * 100, 2) if total_input_tokens > 0 else 0,
            "model_breakdown": model_usage
        }

2. Real-time Monitoring Dashboard

import { EventEmitter } from 'events';
 
interface TokenMetrics {
  timestamp: number;
  model: string;
  inputTokens: number;
  outputTokens: number;
  cacheTokens: number;
  cost: number;
  latency: number;
}
 
interface AlertRule {
  id: string;
  type: 'cost' | 'tokens' | 'latency' | 'rate';
  threshold: number;
  window: number; // seconds
  action: (metrics: TokenMetrics[]) => void;
}
 
class TokenMonitor extends EventEmitter {
  private metrics: TokenMetrics[] = [];
  private alerts: Map<string, AlertRule> = new Map();
  private costAccumulator = 0;
  private tokenAccumulator = 0;
  
  constructor() {
    super();
    // Clean old metrics every hour
    setInterval(() => this.cleanOldMetrics(), 3600000);
  }
  
  recordMetrics(metrics: TokenMetrics) {
    this.metrics.push(metrics);
    this.costAccumulator += metrics.cost;
    this.tokenAccumulator += metrics.inputTokens + metrics.outputTokens;
    
    // Emit real-time events
    this.emit('metrics', metrics);
    this.emit('cost-update', this.costAccumulator);
    
    // Check alerts
    this.checkAlerts();
  }
  
  addAlert(rule: AlertRule) {
    this.alerts.set(rule.id, rule);
  }
  
  removeAlert(id: string) {
    this.alerts.delete(id);
  }
  
  private checkAlerts() {
    const now = Date.now();
    
    for (const [id, rule] of this.alerts) {
      const windowStart = now - (rule.window * 1000);
      const windowMetrics = this.metrics.filter(m => m.timestamp >= windowStart);
      
      let triggered = false;
      
      switch (rule.type) {
        case 'cost':
          const windowCost = windowMetrics.reduce((sum, m) => sum + m.cost, 0);
          triggered = windowCost > rule.threshold;
          break;
          
        case 'tokens':
          const windowTokens = windowMetrics.reduce(
            (sum, m) => sum + m.inputTokens + m.outputTokens, 0
          );
          triggered = windowTokens > rule.threshold;
          break;
          
        case 'latency':
          const avgLatency = windowMetrics.reduce((sum, m) => sum + m.latency, 0) / 
                           windowMetrics.length;
          triggered = avgLatency > rule.threshold;
          break;
          
        case 'rate':
          const requestRate = windowMetrics.length / rule.window;
          triggered = requestRate > rule.threshold;
          break;
      }
      
      if (triggered) {
        rule.action(windowMetrics);
        this.emit('alert', { rule, metrics: windowMetrics });
      }
    }
  }
  
  getRealtimeStats(windowSeconds: number = 300) {
    const now = Date.now();
    const windowStart = now - (windowSeconds * 1000);
    const windowMetrics = this.metrics.filter(m => m.timestamp >= windowStart);
    
    if (windowMetrics.length === 0) {
      return {
        requests: 0,
        totalCost: 0,
        avgLatency: 0,
        tokensPerSecond: 0,
        costPerMinute: 0
      };
    }
    
    const totalCost = windowMetrics.reduce((sum, m) => sum + m.cost, 0);
    const totalTokens = windowMetrics.reduce(
      (sum, m) => sum + m.inputTokens + m.outputTokens, 0
    );
    const avgLatency = windowMetrics.reduce((sum, m) => sum + m.latency, 0) / 
                      windowMetrics.length;
    
    return {
      requests: windowMetrics.length,
      totalCost: totalCost.toFixed(4),
      avgLatency: Math.round(avgLatency),
      tokensPerSecond: Math.round(totalTokens / windowSeconds),
      costPerMinute: (totalCost / windowSeconds * 60).toFixed(4),
      modelBreakdown: this.getModelBreakdown(windowMetrics)
    };
  }
  
  private getModelBreakdown(metrics: TokenMetrics[]) {
    const breakdown: Record<string, any> = {};
    
    for (const metric of metrics) {
      if (!breakdown[metric.model]) {
        breakdown[metric.model] = {
          requests: 0,
          cost: 0,
          avgLatency: 0,
          latencies: []
        };
      }
      
      breakdown[metric.model].requests++;
      breakdown[metric.model].cost += metric.cost;
      breakdown[metric.model].latencies.push(metric.latency);
    }
    
    // Calculate average latencies
    for (const model in breakdown) {
      const latencies = breakdown[model].latencies;
      breakdown[model].avgLatency = Math.round(
        latencies.reduce((a, b) => a + b, 0) / latencies.length
      );
      delete breakdown[model].latencies;
    }
    
    return breakdown;
  }
  
  private cleanOldMetrics() {
    const oneHourAgo = Date.now() - 3600000;
    this.metrics = this.metrics.filter(m => m.timestamp > oneHourAgo);
  }
}
 
// Usage example
const monitor = new TokenMonitor();
 
// Add cost alert
monitor.addAlert({
  id: 'high-cost',
  type: 'cost',
  threshold: 10.00, // $10 in 5 minutes
  window: 300,
  action: (metrics) => {
    console.error(`⚠️ High cost alert: $${
      metrics.reduce((sum, m) => sum + m.cost, 0).toFixed(2)
    } in last 5 minutes`);
    // Send notification, pause requests, etc.
  }
});
 
// Add latency alert
monitor.addAlert({
  id: 'high-latency',
  type: 'latency',
  threshold: 5000, // 5 seconds average
  window: 60,
  action: (metrics) => {
    const avgLatency = metrics.reduce((sum, m) => sum + m.latency, 0) / metrics.length;
    console.warn(`⚠️ High latency: ${avgLatency}ms average`);
  }
});
 
// Real-time dashboard updates
monitor.on('metrics', (metrics: TokenMetrics) => {
  console.log(`📊 ${metrics.model}: ${metrics.cost.toFixed(4)} | ${metrics.latency}ms`);
});
 
monitor.on('cost-update', (totalCost: number) => {
  console.log(`💰 Total cost: $${totalCost.toFixed(2)}`);
});

3. Advanced Analytics with Anomaly Detection

import numpy as np
from sklearn.ensemble import IsolationForest
from collections import deque
import pandas as pd
 
class TokenAnomalyDetector:
    """Detect anomalies in token usage patterns."""
    
    def __init__(self, window_size: int = 100):
        self.window_size = window_size
        self.token_history = deque(maxlen=window_size)
        self.cost_history = deque(maxlen=window_size)
        self.latency_history = deque(maxlen=window_size)
        self.model = IsolationForest(contamination=0.1, random_state=42)
        self.is_trained = False
        
    def add_observation(self, tokens: int, cost: float, latency: int):
        """Add new observation to history."""
        self.token_history.append(tokens)
        self.cost_history.append(cost)
        self.latency_history.append(latency)
        
        # Train model once we have enough data
        if len(self.token_history) >= 50 and not self.is_trained:
            self._train_model()
            
    def _train_model(self):
        """Train anomaly detection model."""
        # Prepare features
        features = np.column_stack([
            list(self.token_history),
            list(self.cost_history),
            list(self.latency_history)
        ])
        
        # Normalize features
        features = (features - features.mean(axis=0)) / features.std(axis=0)
        
        # Train model
        self.model.fit(features)
        self.is_trained = True
        
    def is_anomaly(self, tokens: int, cost: float, latency: int) -> Dict[str, any]:
        """Check if current observation is anomalous."""
        if not self.is_trained:
            return {"is_anomaly": False, "reason": "Model not trained"}
            
        # Prepare feature
        feature = np.array([[tokens, cost, latency]])
        
        # Normalize using history statistics
        history_features = np.column_stack([
            list(self.token_history),
            list(self.cost_history),
            list(self.latency_history)
        ])
        
        mean = history_features.mean(axis=0)
        std = history_features.std(axis=0)
        feature_norm = (feature - mean) / std
        
        # Predict
        prediction = self.model.predict(feature_norm)[0]
        anomaly_score = self.model.score_samples(feature_norm)[0]
        
        # Determine reason if anomaly
        reason = None
        if prediction == -1:  # Anomaly detected
            # Check which dimension is most anomalous
            z_scores = np.abs(feature_norm[0])
            max_z_idx = np.argmax(z_scores)
            dimensions = ['tokens', 'cost', 'latency']
            reason = f"Unusual {dimensions[max_z_idx]} value"
            
        return {
            "is_anomaly": prediction == -1,
            "anomaly_score": float(anomaly_score),
            "reason": reason,
            "z_scores": {
                "tokens": float(z_scores[0]),
                "cost": float(z_scores[1]),
                "latency": float(z_scores[2])
            }
        }
    
    def get_usage_trends(self) -> Dict[str, any]:
        """Analyze usage trends."""
        if len(self.token_history) < 10:
            return {"error": "Insufficient data"}
            
        df = pd.DataFrame({
            'tokens': list(self.token_history),
            'cost': list(self.cost_history),
            'latency': list(self.latency_history)
        })
        
        # Calculate rolling statistics
        window = min(20, len(df) // 2)
        
        return {
            "token_trend": {
                "mean": float(df['tokens'].mean()),
                "std": float(df['tokens'].std()),
                "rolling_mean": list(df['tokens'].rolling(window).mean().dropna()),
                "trend": "increasing" if df['tokens'].iloc[-window:].mean() > df['tokens'].iloc[:window].mean() else "decreasing"
            },
            "cost_trend": {
                "mean": float(df['cost'].mean()),
                "std": float(df['cost'].std()),
                "rolling_mean": list(df['cost'].rolling(window).mean().dropna()),
                "trend": "increasing" if df['cost'].iloc[-window:].mean() > df['cost'].iloc[:window].mean() else "decreasing"
            },
            "latency_trend": {
                "mean": float(df['latency'].mean()),
                "std": float(df['latency'].std()),
                "rolling_mean": list(df['latency'].rolling(window).mean().dropna()),
                "trend": "increasing" if df['latency'].iloc[-window:].mean() > df['latency'].iloc[:window].mean() else "decreasing"
            }
        }

4. Multi-Agent Token Optimization

interface AgentConfig {
  id: string;
  name: string;
  modelPreference: string[];
  maxTokensPerRequest: number;
  costBudgetPerHour: number;
}
 
interface TokenBudget {
  agentId: string;
  allocated: number;
  used: number;
  remaining: number;
  resetTime: number;
}
 
class MultiAgentTokenOptimizer {
  private agents: Map<string, AgentConfig> = new Map();
  private budgets: Map<string, TokenBudget> = new Map();
  private usageHistory: Map<string, TokenMetrics[]> = new Map();
  
  constructor(private client: Anthropic) {
    // Reset budgets hourly
    setInterval(() => this.resetBudgets(), 3600000);
  }
  
  registerAgent(config: AgentConfig) {
    this.agents.set(config.id, config);
    this.budgets.set(config.id, {
      agentId: config.id,
      allocated: config.costBudgetPerHour,
      used: 0,
      remaining: config.costBudgetPerHour,
      resetTime: Date.now() + 3600000
    });
    this.usageHistory.set(config.id, []);
  }
  
  async executeRequest(
    agentId: string,
    messages: any[],
    preferredModel?: string
  ) {
    const agent = this.agents.get(agentId);
    const budget = this.budgets.get(agentId);
    
    if (!agent || !budget) {
      throw new Error(`Agent ${agentId} not registered`);
    }
    
    // Select optimal model based on budget and preference
    const model = this.selectOptimalModel(agent, budget, preferredModel);
    
    // Estimate cost before request
    const estimatedCost = this.estimateRequestCost(messages, model);
    
    if (estimatedCost > budget.remaining) {
      // Try cheaper model or reject
      const cheaperModel = this.findCheaperModel(agent, budget, estimatedCost);
      if (!cheaperModel) {
        throw new Error(`Insufficient budget for agent ${agentId}`);
      }
      model = cheaperModel;
    }
    
    // Execute request with monitoring
    const startTime = Date.now();
    const response = await this.client.messages.create({
      model,
      messages,
      max_tokens: agent.maxTokensPerRequest
    });
    const latency = Date.now() - startTime;
    
    // Track usage
    const actualCost = this.calculateActualCost(response);
    budget.used += actualCost;
    budget.remaining -= actualCost;
    
    const metrics: TokenMetrics = {
      timestamp: Date.now(),
      model,
      inputTokens: response.usage.input_tokens,
      outputTokens: response.usage.output_tokens,
      cacheTokens: response.usage.cache_read_input_tokens || 0,
      cost: actualCost,
      latency
    };
    
    this.usageHistory.get(agentId)!.push(metrics);
    
    return {
      response,
      metrics,
      budgetStatus: {
        used: budget.used.toFixed(4),
        remaining: budget.remaining.toFixed(4),
        percentUsed: (budget.used / budget.allocated * 100).toFixed(1)
      }
    };
  }
  
  private selectOptimalModel(
    agent: AgentConfig,
    budget: TokenBudget,
    preferred?: string
  ): string {
    // If preferred model is within budget, use it
    if (preferred && agent.modelPreference.includes(preferred)) {
      return preferred;
    }
    
    // Otherwise, select based on remaining budget
    const budgetPercentRemaining = budget.remaining / budget.allocated;
    
    if (budgetPercentRemaining > 0.5) {
      // Plenty of budget - use best model
      return agent.modelPreference[0];
    } else if (budgetPercentRemaining > 0.2) {
      // Conservative - use middle tier
      return agent.modelPreference[1] || agent.modelPreference[0];
    } else {
      // Low budget - use cheapest
      return agent.modelPreference[agent.modelPreference.length - 1];
    }
  }
  
  private estimateRequestCost(messages: any[], model: string): number {
    // Rough token estimation (4 chars = 1 token)
    const textLength = JSON.stringify(messages).length;
    const estimatedTokens = Math.ceil(textLength / 4);
    
    const modelCosts = this.getModelCosts(model);
    return (estimatedTokens / 1_000_000) * modelCosts.input;
  }
  
  private calculateActualCost(response: any): number {
    const model = response.model;
    const usage = response.usage;
    const costs = this.getModelCosts(model);
    
    const inputCost = (usage.input_tokens / 1_000_000) * costs.input;
    const outputCost = (usage.output_tokens / 1_000_000) * costs.output;
    const cacheCost = (usage.cache_read_input_tokens || 0) / 1_000_000 * costs.cacheRead;
    
    return inputCost + outputCost + cacheCost;
  }
  
  private findCheaperModel(
    agent: AgentConfig,
    budget: TokenBudget,
    requiredCost: number
  ): string | null {
    // Try models in reverse order (cheapest first)
    for (let i = agent.modelPreference.length - 1; i >= 0; i--) {
      const model = agent.modelPreference[i];
      const modelCosts = this.getModelCosts(model);
      
      // Estimate if this model would fit in budget
      const estimatedCost = requiredCost * (modelCosts.input / 15.00); // Relative to opus
      
      if (estimatedCost <= budget.remaining) {
        return model;
      }
    }
    
    return null;
  }
  
  private resetBudgets() {
    const now = Date.now();
    
    for (const [agentId, budget] of this.budgets) {
      if (now >= budget.resetTime) {
        budget.allocated = this.agents.get(agentId)!.costBudgetPerHour;
        budget.used = 0;
        budget.remaining = budget.allocated;
        budget.resetTime = now + 3600000;
        
        // Clean old history
        const history = this.usageHistory.get(agentId)!;
        const oneHourAgo = now - 3600000;
        this.usageHistory.set(
          agentId,
          history.filter(m => m.timestamp > oneHourAgo)
        );
      }
    }
  }
  
  getAgentReport(agentId: string) {
    const agent = this.agents.get(agentId);
    const budget = this.budgets.get(agentId);
    const history = this.usageHistory.get(agentId);
    
    if (!agent || !budget || !history) {
      return null;
    }
    
    const totalRequests = history.length;
    const avgCostPerRequest = totalRequests > 0 
      ? history.reduce((sum, m) => sum + m.cost, 0) / totalRequests 
      : 0;
    
    const modelUsage: Record<string, number> = {};
    for (const metric of history) {
      modelUsage[metric.model] = (modelUsage[metric.model] || 0) + 1;
    }
    
    return {
      agent: agent.name,
      budget: {
        allocated: budget.allocated.toFixed(2),
        used: budget.used.toFixed(2),
        remaining: budget.remaining.toFixed(2),
        percentUsed: (budget.used / budget.allocated * 100).toFixed(1)
      },
      usage: {
        totalRequests,
        avgCostPerRequest: avgCostPerRequest.toFixed(4),
        modelBreakdown: modelUsage,
        projectedHourlyCost: (avgCostPerRequest * totalRequests).toFixed(2)
      }
    };
  }
  
  private getModelCosts(model: string) {
    // Implementation same as earlier examples
    if (model.includes('opus')) {
      return { input: 15.00, output: 75.00, cacheRead: 1.50 };
    } else if (model.includes('sonnet')) {
      return { input: 3.00, output: 15.00, cacheRead: 0.30 };
    } else {
      return { input: 0.80, output: 4.00, cacheRead: 0.08 };
    }
  }
}

Monitoring Best Practices

1. Set Up Tiered Alerts

ALERT_THRESHOLDS = {
    "cost": {
        "warning": 10.00,    # $10/hour
        "critical": 50.00,   # $50/hour
        "emergency": 100.00  # $100/hour
    },
    "tokens": {
        "warning": 1_000_000,    # 1M tokens/hour
        "critical": 5_000_000,   # 5M tokens/hour
        "emergency": 10_000_000  # 10M tokens/hour
    },
    "latency": {
        "warning": 3000,    # 3 seconds
        "critical": 5000,   # 5 seconds
        "emergency": 10000  # 10 seconds
    }
}

2. Implement Cost Controls

class CostController {
  private dailyLimit: number;
  private hourlyLimit: number;
  private currentDailySpend = 0;
  private currentHourlySpend = 0;
  
  constructor(dailyLimit: number, hourlyLimit: number) {
    this.dailyLimit = dailyLimit;
    this.hourlyLimit = hourlyLimit;
    
    // Reset counters
    setInterval(() => this.currentHourlySpend = 0, 3600000);
    setInterval(() => this.currentDailySpend = 0, 86400000);
  }
  
  canProceed(estimatedCost: number): {
    allowed: boolean;
    reason?: string;
  } {
    if (this.currentHourlySpend + estimatedCost > this.hourlyLimit) {
      return { allowed: false, reason: "Hourly limit exceeded" };
    }
    
    if (this.currentDailySpend + estimatedCost > this.dailyLimit) {
      return { allowed: false, reason: "Daily limit exceeded" };
    }
    
    return { allowed: true };
  }
  
  recordSpend(amount: number) {
    this.currentHourlySpend += amount;
    this.currentDailySpend += amount;
  }
}

3. Optimize Token Usage

def optimize_prompt(prompt: str, max_tokens: int = 1000) -> str:
    """Optimize prompt to reduce token usage while maintaining quality."""
    
    # Remove excessive whitespace
    prompt = ' '.join(prompt.split())
    
    # Use abbreviations for common terms
    replacements = {
        "artificial intelligence": "AI",
        "machine learning": "ML",
        "natural language processing": "NLP",
        "for example": "e.g.",
        "that is": "i.e.",
    }
    
    for full, abbr in replacements.items():
        prompt = prompt.replace(full, abbr)
    
    # Truncate if still too long (rough estimate: 4 chars = 1 token)
    if len(prompt) > max_tokens * 4:
        prompt = prompt[:max_tokens * 4] + "..."
    
    return prompt

Dashboard Integration

Sample Grafana Query

-- Token usage over time
SELECT 
  DATE_TRUNC('hour', timestamp) as time,
  SUM(input_tokens + output_tokens) as total_tokens,
  SUM(total_cost) as total_cost,
  AVG(latency_ms) as avg_latency
FROM token_usage
WHERE timestamp >= NOW() - INTERVAL '24 hours'
GROUP BY time
ORDER BY time;
 
-- Cost by model
SELECT 
  model,
  COUNT(*) as requests,
  SUM(total_cost) as total_cost,
  AVG(total_cost) as avg_cost_per_request
FROM token_usage
WHERE timestamp >= NOW() - INTERVAL '24 hours'
GROUP BY model
ORDER BY total_cost DESC;
 
-- Cache efficiency
SELECT 
  DATE_TRUNC('hour', timestamp) as time,
  SUM(cache_read_tokens)::float / NULLIF(SUM(input_tokens), 0) * 100 as cache_hit_rate,
  SUM(cache_read_tokens) * 2.70 / 1000000 as cache_savings
FROM token_usage
WHERE timestamp >= NOW() - INTERVAL '24 hours'
GROUP BY time
ORDER BY time;

ROI Calculation

def calculate_optimization_roi(
    baseline_metrics: Dict[str, float],
    optimized_metrics: Dict[str, float],
    implementation_hours: float = 40,
    developer_rate: float = 150.0
) -> Dict[str, float]:
    """Calculate ROI for token optimization efforts."""
    
    # Implementation cost
    implementation_cost = implementation_hours * developer_rate
    
    # Monthly savings
    baseline_monthly = baseline_metrics["daily_cost"] * 30
    optimized_monthly = optimized_metrics["daily_cost"] * 30
    monthly_savings = baseline_monthly - optimized_monthly
    
    # Payback period
    payback_months = implementation_cost / monthly_savings if monthly_savings > 0 else float('inf')
    
    # Annual ROI
    annual_savings = monthly_savings * 12
    annual_roi = ((annual_savings - implementation_cost) / implementation_cost) * 100
    
    return {
        "implementation_cost": implementation_cost,
        "monthly_savings": monthly_savings,
        "annual_savings": annual_savings,
        "payback_months": payback_months,
        "annual_roi_percent": annual_roi,
        "cost_reduction_percent": ((baseline_monthly - optimized_monthly) / baseline_monthly) * 100
    }

Conclusion

Effective token usage monitoring and analytics are essential for:

1. Cost Control: Prevent budget overruns with real-time alerts
2. Performance Optimization: Identify and fix latency issues
3. Capacity Planning: Understand usage patterns for scaling
4. ROI Demonstration: Quantify the value of optimization efforts

Implement these patterns progressively, starting with basic tracking and gradually adding advanced analytics as your usage scales.