Cost Tracking

Track per-request costs, enforce department budgets, and alert on spending anomalies.

Cost Tracker

# src/costs.py
from datetime import datetime, timedelta
from typing import Dict, Optional
import json

# Token pricing per 1M tokens (input/output)
PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "claude-sonnet-4-20250514": {"input": 3.00, "output": 15.00},
    "claude-3-5-haiku-20241022": {"input": 0.80, "output": 4.00},
}

class CostTracker:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.budgets = {}

    def calculate_cost(self, model, prompt_tokens, completion_tokens):
        pricing = PRICING.get(model, {"input": 5.0, "output": 15.0})
        input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
        output_cost = (completion_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)

    async def record(self, user_id, team_id, model,
                     prompt_tokens, completion_tokens):
        cost = self.calculate_cost(model, prompt_tokens, completion_tokens)
        now = datetime.now()
        record = {
            "user_id": user_id, "team_id": team_id, "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "cost": cost, "timestamp": now.isoformat(),
        }

        # Store in Redis sorted set by timestamp
        day_key = f"costs:{now.strftime('%Y-%m-%d')}"
        await self.redis.zadd(day_key, {json.dumps(record): now.timestamp()})
        await self.redis.expire(day_key, 86400 * 90)

        # Update running totals
        await self.redis.incrbyfloat(f"cost:user:{user_id}:daily", cost)
        await self.redis.incrbyfloat(f"cost:team:{team_id}:monthly", cost)

        return {"cost": cost, "record": record}

    def set_budget(self, team_id, monthly_budget):
        self.budgets[team_id] = monthly_budget

    async def check_budget(self, team_id):
        spent = await self.redis.get(f"cost:team:{team_id}:monthly")
        spent = float(spent) if spent else 0
        budget = self.budgets.get(team_id, float('inf'))
        return {
            "team_id": team_id,
            "spent": round(spent, 2),
            "budget": budget,
            "remaining": round(budget - spent, 2),
            "utilization": round(spent / budget * 100, 1) if budget else 0,
            "over_budget": spent > budget,
        }

    async def get_daily_summary(self, date=None):
        if date is None:
            date = datetime.now().strftime('%Y-%m-%d')
        day_key = f"costs:{date}"
        records = await self.redis.zrange(day_key, 0, -1)
        total = 0
        by_model = {}
        by_team = {}
        for r in records:
            rec = json.loads(r)
            total += rec["cost"]
            by_model[rec["model"]] = by_model.get(rec["model"], 0) + rec["cost"]
            by_team[rec["team_id"]] = by_team.get(rec["team_id"], 0) + rec["cost"]
        return {"date": date, "total_cost": round(total, 4),
                "by_model": by_model, "by_team": by_team,
                "total_requests": len(records)}