Skip to main content

Cost Tracking & Budget Management

Advanced cost tracking with predictive analysis, budget management, and real-time monitoring across all LLM providers and operations.

Features​

  • Real-time Cost Monitoring: Track costs across all operations
  • Predictive Analysis: ML-based cost predictions
  • Budget Management: Set and enforce budget limits
  • Cost Optimization: Automatic cost reduction recommendations
  • Multi-Provider Support: Track costs across all LLM providers
  • Alert System: Budget alerts and notifications

Quick Start​

from packages.observability import CostTracker, CostCategory, BudgetAlertLevel

# Initialize cost tracker
tracker = CostTracker(
    enable_predictions=True,
    prediction_window_days=30,
    alert_thresholds={
        "info": 0.5,      # 50% of budget
        "warning": 0.75,  # 75% of budget
        "critical": 0.9,  # 90% of budget
        "exceeded": 1.0   # 100% of budget
    }
)

# Set budget
tracker.set_budget(
    budget_name="monthly_llm_budget",
    amount_usd=1000.0,
    category=CostCategory.LLM_TOKENS,
    period="monthly"
)

# Track costs
tracker.add_cost_entry(
    category=CostCategory.LLM_TOKENS,
    provider="openai",
    model="gpt-4",
    operation="chat_completion",
    cost_usd=0.05,
    tokens_input=100,
    tokens_output=50
)

Cost Categories​

LLM Tokens​

# Track LLM token usage
tracker.add_cost_entry(
    category=CostCategory.LLM_TOKENS,
    provider="openai",
    model="gpt-4",
    operation="chat_completion",
    cost_usd=0.05,
    tokens_input=100,
    tokens_output=50,
    metadata={"user_id": "user123"}
)

LLM Requests​

# Track LLM requests
tracker.add_cost_entry(
    category=CostCategory.LLM_REQUESTS,
    provider="anthropic",
    model="claude-3",
    operation="completion",
    cost_usd=0.02,
    metadata={"request_type": "generation"}
)
# Track vector search costs
tracker.add_cost_entry(
    category=CostCategory.VECTOR_SEARCH,
    provider="pinecone",
    model="embedding-model",
    operation="vector_search",
    cost_usd=0.001,
    metadata={"query_count": 1}
)

Tool Execution​

# Track tool execution costs
tracker.add_cost_entry(
    category=CostCategory.TOOL_EXECUTION,
    provider="custom",
    model="tool",
    operation="web_search",
    cost_usd=0.01,
    metadata={"tool_name": "web_search"}
)

Budget Management​

Setting Budgets​

# Monthly budget
tracker.set_budget(
    budget_name="monthly_budget",
    amount_usd=1000.0,
    period="monthly"
)

# Daily budget
tracker.set_budget(
    budget_name="daily_budget",
    amount_usd=50.0,
    period="daily"
)

# Category-specific budget
tracker.set_budget(
    budget_name="llm_budget",
    amount_usd=500.0,
    category=CostCategory.LLM_TOKENS,
    period="monthly"
)

Budget Status​

# Check budget status
status = tracker.get_budget_status()

for budget_name, budget_info in status.items():
    print(f"Budget: {budget_name}")
    print(f"Amount: ${budget_info['amount']}")
    print(f"Current Spend: ${budget_info['current_spend']}")
    print(f"Percentage Used: {budget_info['percentage_used']:.1f}%")
    print(f"Status: {budget_info['status']}")
    print(f"Remaining: ${budget_info['remaining']}")

Budget Alerts​

# Get alerts
alerts = tracker.get_alerts()

for alert in alerts:
    print(f"Alert Level: {alert.level}")
    print(f"Message: {alert.message}")
    print(f"Current Spend: ${alert.current_spend}")
    print(f"Budget Limit: ${alert.budget_limit}")
    print(f"Percentage Used: {alert.percentage_used:.1f}%")

Cost Analysis​

Current Spending​

# Get current spending
current_spend = tracker.get_current_spend(days=30)
print(f"Last 30 days: ${current_spend:.2f}")

# By category
llm_spend = tracker.get_current_spend(
    category=CostCategory.LLM_TOKENS,
    days=30
)
print(f"LLM costs: ${llm_spend:.2f}")

# By provider
openai_spend = tracker.get_current_spend(
    provider="openai",
    days=30
)
print(f"OpenAI costs: ${openai_spend:.2f}")

Cost Breakdown​

# Get detailed breakdown
breakdown = tracker.get_cost_breakdown(days=30)

print(f"Total Cost: ${breakdown['total_cost']:.2f}")
print(f"By Category: {breakdown['by_category']}")
print(f"By Provider: {breakdown['by_provider']}")
print(f"By Model: {breakdown['by_model']}")
print(f"By Operation: {breakdown['by_operation']}")
print(f"Token Usage: {breakdown['token_usage']}")
# Get daily cost trends
breakdown = tracker.get_cost_breakdown(days=30)
daily_trend = breakdown['daily_trend']

for date, cost in daily_trend.items():
    print(f"{date}: ${cost:.2f}")

Predictive Analysis​

Cost Predictions​

# Get cost predictions
prediction = tracker.predict_costs()

if prediction:
    print(f"Predicted Daily Cost: ${prediction.predicted_daily_cost:.2f}")
    print(f"Predicted Weekly Cost: ${prediction.predicted_weekly_cost:.2f}")
    print(f"Predicted Monthly Cost: ${prediction.predicted_monthly_cost:.2f}")
    print(f"Confidence: {prediction.confidence:.2f}")
    print(f"Trend: {prediction.trend}")
    print(f"Based on {prediction.based_on_days} days of data")

Trend Analysis​

# Analyze trends
prediction = tracker.predict_costs()

if prediction.trend == "increasing":
    print("⚠️ Costs are increasing - consider optimization")
elif prediction.trend == "decreasing":
    print("✅ Costs are decreasing - good optimization")
else:
    print("📊 Costs are stable")

Cost Optimization​

Cost Summary​

# Get comprehensive cost summary
summary = tracker.get_cost_summary()

print("=== Cost Summary ===")
print(f"Total Cost: ${summary['summary']['total_cost']:.2f}")
print(f"Total Tokens: {summary['summary']['total_tokens']:,}")
print(f"Total Requests: {summary['summary']['total_requests']:,}")
print(f"Avg Cost/Request: ${summary['summary']['avg_cost_per_request']:.4f}")
print(f"Avg Cost/Token: ${summary['summary']['avg_cost_per_token']:.6f}")

print("\n=== Breakdown ===")
for category, cost in summary['breakdown']['by_category'].items():
    print(f"{category}: ${cost:.2f}")

print("\n=== Budget Status ===")
for budget_name, budget_info in summary['budget_status'].items():
    print(f"{budget_name}: {budget_info['percentage_used']:.1f}% used")

print("\n=== Predictions ===")
if summary['prediction']:
    pred = summary['prediction']
    print(f"Daily: ${pred['predicted_daily_cost']:.2f}")
    print(f"Weekly: ${pred['predicted_weekly_cost']:.2f}")
    print(f"Monthly: ${pred['predicted_monthly_cost']:.2f}")

Export Data​

# Export to JSON
json_data = tracker.export_data(format="json", days=30)
with open("cost_data.json", "w") as f:
    json.dump(json_data, f, indent=2)

# Export to CSV (requires pandas)
csv_data = tracker.export_data(format="csv", days=30)
with open("cost_data.csv", "w") as f:
    f.write(csv_data)

Integration Examples​

With LiteLLM​

from packages.llm import LiteLLMProvider, LiteLLMConfig
from packages.observability import track_cost, CostCategory

# Configure LLM with cost tracking
config = LiteLLMConfig(
    model="gpt-4",
    enable_cost_tracking=True
)
llm = LiteLLMProvider(config)

# Track costs automatically
@track_cost(
    category=CostCategory.LLM_TOKENS,
    provider="openai",
    model="gpt-4",
    operation="chat_completion"
)
async def chat_completion(messages):
    return await llm.ainvoke(messages)

With Workflows​

from packages.observability import trace_workflow, get_cost_tracker

@trace_workflow(name="cost_tracked_workflow")
async def my_workflow():
    tracker = get_cost_tracker()
    
    # Track workflow costs
    tracker.add_cost_entry(
        category=CostCategory.LLM_TOKENS,
        provider="openai",
        model="gpt-4",
        operation="workflow_execution",
        cost_usd=0.10,
        metadata={"workflow_id": "workflow_123"}
    )
    
    return result

With FastAPI​

from fastapi import FastAPI, Depends
from packages.observability import get_cost_tracker

app = FastAPI()

@app.get("/costs/summary")
async def get_cost_summary():
    tracker = get_cost_tracker()
    return tracker.get_cost_summary()

@app.get("/costs/budget-status")
async def get_budget_status():
    tracker = get_cost_tracker()
    return tracker.get_budget_status()

@app.get("/costs/predictions")
async def get_cost_predictions():
    tracker = get_cost_tracker()
    return tracker.predict_costs()

Advanced Configuration​

Custom Alert Thresholds​

tracker = CostTracker(
    alert_thresholds={
        "info": 0.3,      # 30% of budget
        "warning": 0.6,   # 60% of budget
        "critical": 0.8,  # 80% of budget
        "exceeded": 1.0   # 100% of budget
    }
)

Prediction Configuration​

tracker = CostTracker(
    enable_predictions=True,
    prediction_window_days=60,  # Use 60 days for predictions
    max_history_days=365       # Keep 1 year of history
)

Budget Alerts​

# Set up automatic alerts
tracker.set_budget(
    budget_name="monthly_budget",
    amount_usd=1000.0,
    period="monthly"
)

# Alerts are automatically generated when thresholds are reached
alerts = tracker.get_alerts(level=BudgetAlertLevel.CRITICAL)
for alert in alerts:
    print(f"CRITICAL: {alert.message}")

Best Practices​

  1. Set Realistic Budgets: Base budgets on historical data and business needs
  2. Monitor Trends: Watch for cost trends and optimize accordingly
  3. Use Categories: Categorize costs for better analysis
  4. Set Alerts: Configure appropriate alert thresholds
  5. Regular Reviews: Review costs and budgets regularly
  6. Optimize Based on Data: Use cost data to optimize operations
  7. Export Data: Regularly export data for external analysis

Troubleshooting​

Common Issues​

  1. Missing Cost Data: Ensure cost tracking is enabled
  2. Inaccurate Predictions: Check data quality and quantity
  3. Alert Spam: Adjust alert thresholds
  4. Performance Issues: Optimize data storage and queries

Debug Mode​

# Enable debug logging
import logging
logging.getLogger('packages.observability.cost_tracker').setLevel(logging.DEBUG)

API Reference​

CostTracker​

ParameterTypeDescription
enable_predictionsboolEnable cost predictions
prediction_window_daysintDays to use for predictions
alert_thresholdsDictAlert threshold percentages
max_history_daysintMaximum days to keep history

CostCategory​

ValueDescription
LLM_TOKENSLLM token usage costs
LLM_REQUESTSLLM request costs
VECTOR_SEARCHVector search costs
TOOL_EXECUTIONTool execution costs
STORAGEStorage costs
API_CALLSAPI call costs
OTHEROther costs

BudgetAlertLevel​

ValueDescription
INFOInformational alert
WARNINGWarning alert
CRITICALCritical alert
EXCEEDEDBudget exceeded alert