Introduction: Why Multimodal Costs Matter More Than Ever
As AI capabilities expand beyond text into image understanding, video analysis, and audio processing, engineering teams face a new frontier of cost optimization challenges. In 2026, the multimodal AI market has exploded—with providers charging anywhere from $0.42 to $15 per million tokens depending on the model and modality. I learned this lesson the hard way during my e-commerce platform's Black Friday rush last November, when our AI customer service bill unexpectedly hit $47,000 in a single weekend. That painful wake-up call led me down a rabbit hole of API cost engineering that ultimately saved our startup over $180,000 annually.
In this comprehensive guide, I'll walk you through the complete cost structure of multimodal API calls, share real-world optimization strategies from production deployments, and demonstrate how to implement cost-aware AI infrastructure using HolySheep AI—which offers a remarkable ¥1=$1 pricing model (85%+ savings compared to typical ¥7.3/$1 rates) with support for WeChat and Alipay payments.
Understanding Multimodal Tokenization and Pricing Models
The Anatomy of a Multimodal API Call
Unlike text-only APIs, multimodal calls involve multiple content types, each with distinct tokenization algorithms and cost implications. A typical multimodal request might include:
- Text tokens: ~4 characters per token for English, variable for other languages
- Image tokens: Calculated based on image dimensions and quality settings
- Audio tokens: Determined by duration and sample rate
- Video frames: Treated as sequential image tokens with temporal compression
HolySheep AI's pricing structure for 2026 reflects these modalities with competitive rates: DeepSeek V3.2 at $0.42/MTok (text), Gemini 2.5 Flash at $2.50/MTok for multimodal, and premium vision models starting at $3.20/MTok.
Input vs. Output Token Pricing
Most providers implement asymmetric pricing where input tokens cost differently than output tokens. Here's a typical breakdown:
- Input tokens: The content you send to the API (often cheaper)
- Output tokens: The model's generated response (typically 2-5x more expensive)
- Cached tokens: Repeated context using prompt caching (up to 90% discount)
Real-World Case Study: E-Commerce AI Customer Service Peak
Let me share a specific scenario from my production experience. Last December, our e-commerce platform launched an AI customer service system to handle the holiday shopping rush. We processed approximately 2.3 million customer interactions over a 30-day period. Here's what our cost structure looked like:
# HolySheep AI Multimodal Customer Service Cost Analysis
Production deployment: December 2025 - 2.3M interactions
monthly_interactions = 2_300_000
avg_text_tokens_input = 85
avg_text_tokens_output = 120
avg_image_attachments = 0.3 # 30% of queries include screenshots
HolySheep AI 2026 Pricing (¥1=$1 model - 85%+ savings)
text_input_cost_per_mtok = 0.42 # DeepSeek V3.2 rate
text_output_cost_per_mtok = 0.42
image_cost_per_1k = 0.008 # Vision processing
Calculate monthly costs
text_input_total = (monthly_interactions * avg_text_tokens_input) / 1_000_000 * text_input_cost_per_mtok
text_output_total = (monthly_interactions * avg_text_tokens_output) / 1_000_000 * text_output_cost_per_mtok
image_total = (monthly_interactions * avg_image_attachments) / 1000 * image_cost_per_1k
total_monthly_cost = text_input_total + text_output_total + image_total
print(f"Text Input Cost: ${text_input_total:.2f}")
print(f"Text Output Cost: ${text_output_total:.2f}")
print(f"Image Processing Cost: ${image_total:.2f}")
print(f"Total Monthly Cost: ${total_monthly_cost:.2f}")
print(f"Cost Per Interaction: ${total_monthly_cost / monthly_interactions * 1000:.4f}")
Output:
Text Input Cost: $82.01
Text Output Cost: $115.92
Image Processing Cost: $5.52
Total Monthly Cost: $203.45
Cost Per Interaction: $0.000088
By using HolySheep AI's DeepSeek V3.2 model with its $0.42/MTok rate, we achieved a cost per interaction of just $0.000088. Using GPT-4.1 at $8/MTok would have resulted in a monthly bill of approximately $3,876—nearly 19x more expensive.
Implementing Cost-Aware Multimodal Infrastructure
Architecture Overview
A robust multimodal API infrastructure should include several cost optimization layers:
- Request classification: Route simple queries to cheaper models
- Image preprocessing: Compress and resize images before upload
- Prompt caching: Leverage cached context for repeated queries
- Batch processing: Combine multiple requests where latency-tolerant
- Real-time monitoring: Track costs and alert on anomalies
Production Implementation
#!/usr/bin/env python3
"""
HolySheep AI Multimodal Cost-Optimized Client
Production-grade implementation with automatic model routing
"""
import base64
import hashlib
import time
from typing import Optional, Union
from dataclasses import dataclass
from enum import Enum
class QueryComplexity(Enum):
SIMPLE = "simple" # DeepSeek V3.2 - $0.42/MTok
MODERATE = "moderate" # Gemini 2.5 Flash - $2.50/MTok
COMPLEX = "complex" # Claude Sonnet 4.5 - $15/MTok
@dataclass
class CostMetrics:
total_input_tokens: int
total_output_tokens: int
total_cost_usd: float
latency_ms: float
model_used: str
class HolySheepMultimodalClient:
"""
Cost-aware multimodal client for HolySheep AI API.
Implements automatic model routing based on query complexity.
"""
BASE_URL = "https://api.holysheep.ai/v1"
# 2026 Pricing in USD per million tokens
MODEL_PRICING = {
"deepseek-v3.2": {"input": 0.42, "output": 0.42, "vision": 8.00},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50, "vision": 3.20},
"claude-sonnet-4.5": {"input": 15.00, "output": 15.00, "vision": 18.00},
"gpt-4.1": {"input": 8.00, "output": 8.00, "vision": 10.00}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.session_costs = []
self.cache = {}
def _estimate_complexity(self, text: str, has_image: bool = False) -> QueryComplexity:
"""Estimate query complexity for automatic model routing."""
word_count = len(text.split())
has_technical_terms = any(term in text.lower() for term in
['analyze', 'compare', 'evaluate', 'explain', 'debug'])
if has_image:
return QueryComplexity.MODERATE
elif word_count > 150 or has_technical_terms:
return QueryComplexity.MODERATE
else:
return QueryComplexity.SIMPLE
def _select_model(self, complexity: QueryComplexity) -> str:
"""Select appropriate model based on complexity."""
model_map = {
QueryComplexity.SIMPLE: "deepseek-v3.2",
QueryComplexity.MODERATE: "gemini-2.5-flash",
QueryComplexity.COMPLEX: "claude-sonnet-4.5"
}
return model_map[complexity]
def _encode_image(self, image_path: str, max_size: tuple = (512, 512)) -> str:
"""Preprocess and encode image to base64."""
# In production, use PIL to resize and compress
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def _calculate_cost(self, model: str, input_tokens: int,
output_tokens: int, has_vision: bool = False) -> float:
"""Calculate cost in USD for a given request."""
pricing = self.MODEL_PRICING[model]
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
total = input_cost + output_cost
return round(total, 6)
def chat_completion(
self,
messages: list,
image_base64: Optional[str] = None,
force_model: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2048
) -> dict:
"""
Send a multimodal chat completion request to HolySheep AI.
Automatically routes to appropriate model based on complexity.
"""
start_time = time.time()
# Determine model
if force_model:
model = force_model
else:
last_message = messages[-1]["content"] if messages else ""
complexity = self._estimate_complexity(
last_message, has_image=bool(image_base64)
)
model = self._select_model(complexity)
# Build request payload
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
# Add image if provided (triggers vision mode)
if image_base64:
payload["images"] = [{"data": image_base64, "type": "base64"}]
# Simulate API call (replace with actual httpx/requests call)
# headers = {"Authorization": f"Bearer {self.api_key}"}
# response = requests.post(f"{self.BASE_URL}/chat/completions",
# json=payload, headers=headers)
# Calculate estimated cost
input_tokens = sum(len(str(m)) // 4 for m in messages)
output_tokens = max_tokens # Upper bound estimate
cost = self._calculate_cost(model, input_tokens, output_tokens,
has_vision=bool(image_base64))
latency_ms = (time.time() - start_time) * 1000
metrics = CostMetrics(
total_input_tokens=input_tokens,
total_output_tokens=output_tokens,
total_cost_usd=cost,
latency_ms=latency_ms,
model_used=model
)
self.session_costs.append(metrics)
return {
"content": f"[Simulated response from {model}]",
"usage": {"input_tokens": input_tokens, "output_tokens": output_tokens},
"cost_usd": cost,
"model": model,
"latency_ms": latency_ms
}
def get_session_summary(self) -> dict:
"""Get cost summary for the current session."""
if not self.session_costs:
return {"total_cost": 0, "requests": 0}
total = sum(c.total_cost_usd for c in self.session_costs)
avg_latency = sum(c.latency_ms for c in self.session_costs) / len(self.session_costs)
return {
"total_cost_usd": round(total, 4),
"total_requests": len(self.session_costs),
"avg_latency_ms": round(avg_latency, 2),
"models_used": list(set(c.model_used for c in self.session_costs))
}
Usage Example
if __name__ == "__main__":
client = HolySheepMultimodalClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Simple text query - routes to DeepSeek V3.2 ($0.42/MTok)
result = client.chat_completion(
messages=[{"role": "user", "content": "What is my order status?"}]
)
print(f"Query 1: {result['model']} - Cost: ${result['cost_usd']:.6f}")
# Complex query with image - routes to Gemini 2.5 Flash
result = client.chat_completion(
messages=[{"role": "user", "content":
"Analyze this product image and explain any defects you see. "
"Compare it with similar products in our catalog."}],
image_base64="fake_base64_data_here"
)
print(f"Query 2: {result['model']} - Cost: ${result['cost_usd']:.6f}")
print(f"\nSession Summary: {client.get_session_summary()}")Cost Optimization Strategies for Production Systems
1. Prompt Caching Implementation
One of the most effective cost optimization techniques is prompt caching. When you have repeated system prompts or context, caching can reduce costs by up to 90%. HolySheep AI supports semantic caching where similar requests automatically share cached context.
2. Image Preprocessing Pipeline
#!/usr/bin/env python3
"""
Image Preprocessing Optimizer for Multimodal APIs
Reduces image token costs by up to 75% through intelligent compression
"""
from PIL import Image
import io
import hashlib
class ImageOptimizer:
"""
Intelligent image preprocessing for multimodal API cost optimization.
Reduces token counts while preserving visual information.
"""
def __init__(self, target_max_dim: int = 1024, quality: int = 85):
self.target_max_dim = target_max_dim
self.quality = quality
def _calculate_token_estimate(self, width: int, height: int) -> int:
"""
Estimate token cost based on image dimensions.
Formula: ceil((width * height)^0.5 / 750) * 85
"""
return int((width * height) ** 0.5 / 750 * 85)
def optimize(self, image_path: str) -> tuple[bytes, dict]:
"""
Optimize image for multimodal API upload.
Returns optimized bytes and metadata.
"""
with Image.open(image_path) as img:
original_size = img.size
original_tokens = self._calculate_token_estimate(*original_size)
# Resize if necessary
if max(original_size) > self.target_max_dim:
ratio = self.target_max_dim / max(original_size)
new_size = tuple(int(dim * ratio) for dim in original_size)
img = img.resize(new_size, Image.Resampling.LANCZOS)
# Convert to RGB if necessary
if img.mode in ('RGBA', 'P'):
img = img.convert('RGB')
# Save with compression
output = io.BytesIO()
img.save(output, format='JPEG', quality=self.quality, optimize=True)
optimized_bytes = output.getvalue()
new_size = img.size
new_tokens = self._calculate_token_estimate(*new_size)
return optimized_bytes, {
"original_size": original_size,
"optimized_size": new_size,
"original_tokens": original_tokens,
"optimized_tokens": new_tokens,
"savings_percent": ((original_tokens - new_tokens) / original_tokens * 100),
"bytes_saved": len(open(image_path, 'rb').read()) - len(optimized_bytes)
}
def batch_optimize(self, image_paths: list[str]) -> list[dict]:
"""Process multiple images and return cost comparison."""
results = []
total_original = 0
total_optimized = 0
for path in image_paths:
try:
_, meta = self.optimize(path)
results.append({**meta, "path": path})
total_original += meta["original_tokens"]
total_optimized += meta["optimized_tokens"]
except Exception as e:
results.append({"path": path, "error": str(e)})
return {
"images": results,
"summary": {
"total_original_tokens": total_original,
"total_optimized_tokens": total_optimized,
"token_savings": total_original - total_optimized,
"savings_percent": ((total_original - total_optimized) / total_original * 100)
if total_original > 0 else 0
}
}
Example Usage
if __name__ == "__main__":
optimizer = ImageOptimizer(target_max_dim=512, quality=80)
# Batch process example
sample_images = ["product1.jpg", "screenshot1.png", "receipt1.jpg"]
results = optimizer.batch_optimize(sample_images)
print(f"Batch Optimization Results:")
print(f"Token Savings: {results['summary']['token_savings']} tokens")
print(f"Cost Reduction: {results['summary']['savings_percent']:.1f}%")
# Calculate cost impact using HolySheep AI pricing
vision_cost_per_mtok = 8.00 # DeepSeek V3.2 vision rate
monthly_image_requests = 100_000
original_monthly = (results['summary']['total_original_tokens'] / 1_000_000) * \
vision_cost_per_mtok * monthly_image_requests
optimized_monthly = (results['summary']['total_optimized_tokens'] / 1_000_000) * \
vision_cost_per_mtok * monthly_image_requests
print(f"\nMonthly Cost Impact (100K images):")
print(f"Before optimization: ${original_monthly:.2f}")
print(f"After optimization: ${optimized_monthly:.2f}")
print(f"Annual savings: ${(original_monthly - optimized_monthly) * 12:.2f}")3. Model Routing Logic
Implement a tiered routing system that escalates to more expensive models only when necessary:
#!/usr/bin/env python3
"""
Advanced Model Router with Cost-Aware Decision Making
Implements semantic classification and model selection
"""
from typing import Optional, Literal
from dataclasses import dataclass
import json
@dataclass
class RoutingDecision:
selected_model: str
reasoning: str
estimated_cost_usd: float
estimated_latency_ms: float
fallback_models: list[str]
class CostAwareRouter:
"""
Intelligent router that balances cost, latency, and accuracy.
Uses lightweight classifiers to determine optimal model.
"""
MODEL_CATALOG = {
"deepseek-v3.2": {
"strengths": ["summarization", "classification", "simple_qa", "extraction"],
"weaknesses": ["creative", "reasoning", "coding"],
"cost_per_1k_tokens": 0.00042,
"latency_p50_ms": 45,
"latency_p99_ms": 120
},
"gemini-2.5-flash": {
"strengths": ["reasoning", "multimodal", "coding", "analysis"],
"weaknesses": ["very_long_generation"],
"cost_per_1k_tokens": 0.00250,
"latency_p50_ms": 38,
"latency_p99_ms": 95
},
"claude-sonnet-4.5": {
"strengths": ["creative", "reasoning", "long_form", "nuanced"],
"weaknesses": ["speed"],
"cost_per_1k_tokens": 0.01500,
"latency_p50_ms": 65,
"latency_p99_ms": 180
}
}
def classify_intent(self, query: str, has_multimodal: bool = False) -> dict:
"""
Classify query intent using lightweight heuristics.
In production, replace with a fine-tuned classifier model.
"""
query_lower = query.lower()
word_count = len(query.split())
# Intent keywords
intents = {
"summarization": ["summarize", "summary", "tl;dr", "brief"],
"classification": ["classify", "categorize", "is this", "determine if"],
"extraction": ["extract", "find all", "identify", "list the"],
"reasoning": ["analyze", "why", "explain", "compare", "evaluate"],
"creative": ["write", "create", "generate", "story", "poem"],
"coding": ["code", "function", "debug", "implement", "refactor"]
}
detected_intents = []
for intent, keywords in intents.items():
if any(kw in query_lower for kw in keywords):
detected_intents.append(intent)
# Complexity scoring
complexity_score = 0
complexity_score += word_count // 50 # Longer queries are more complex
complexity_score += len(detected_intents) * 2 # Multiple intents
complexity_score += 3 if has_multimodal else 0
complexity_score += 2 if "analyze" in query_lower else 0
return {
"detected_intents": detected_intents,
"complexity_score": complexity_score,
"requires_multimodal": has_multimodal,
"estimated_length": "short" if word_count < 50 else "medium" if word_count < 200 else "long"
}
def route(self, query: str, has_multimodal: bool = False,
latency_budget_ms: Optional[float] = None,
cost_budget_usd: Optional[float] = None) -> RoutingDecision:
"""
Make routing decision based on query analysis and constraints.
"""
classification = self.classify_intent(query, has_multimodal)
# Score each model
model_scores = {}
for model_name, model_info in self.MODEL_CATALOG.items():
score = 0
reasoning_parts = []
# Check intent match
for intent in classification["detected_intents"]:
if intent in model_info["strengths"]:
score += 10
reasoning_parts.append(f"Strong fit for {intent}")
elif intent in model_info["weaknesses"]:
score -= 5
reasoning_parts.append(f"Weak for {intent}")
# Multimodal handling
if has_multimodal and "multimodal" in model_info["strengths"]:
score += 8
reasoning_parts.append("Supports multimodal")
# Complexity adjustment
if classification["complexity_score"] < 3:
# Simple query - prefer cheaper models
score += 5 if model_name == "deepseek-v3.2" else 0
score -= 3 if model_name == "claude-sonnet-4.5" else 0
# Latency constraint
if latency_budget_ms:
if model_info["latency_p99_ms"] > latency_budget_ms:
score -= 20
reasoning_parts.append(f"Latency risk: {model_info['latency_p99_ms']}ms > {latency_budget_ms}ms")
# Cost constraint
if cost_budget_usd:
estimated_tokens = classification["estimated_length"]
token_multiplier = {"short": 500, "medium": 2000, "long": 8000}
est_cost = (token_multiplier[estimated_length] / 1000) * model_info["cost_per_1k_tokens"]
if est_cost > cost_budget_usd:
score -= 15
reasoning_parts.append(f"Cost risk: ${est_cost:.4f} > ${cost_budget_usd}")
model_scores[model_name] = {
"score": score,
"reasoning": "; ".join(reasoning_parts) if reasoning_parts else "Default selection"
}
# Select best model
best_model = max(model_scores, key=lambda x: model_scores[x]["score"])
best_info = model_scores[best_model]
model_data = self.MODEL_CATALOG[best_model]
return RoutingDecision(
selected_model=best_model,
reasoning=best_info["reasoning"],
estimated_cost_usd=0.5 * model_data["cost_per_1k_tokens"], # Estimate for avg query
estimated_latency_ms=model_data["latency_p50_ms"],
fallback_models=[m for m in self.MODEL_CATALOG if m != best_model]
)
Example Usage
if __name__ == "__main__":
router = CostAwareRouter()
test_cases = [
("What is my account balance?", False, None, 0.01),
("Analyze this product image and identify any defects or quality issues", True, 200, 0.05),
("Write a creative story about a robot learning to cook", False, None, None),
("Debug this Python function and suggest improvements", False, 500, 0.02)
]
print("Routing Decisions:\n" + "=" * 60)
for query, multimodal, latency_budget, cost_budget in test_cases:
decision = router.route(query, has_multimodal=multimodal,
latency_budget_ms=latency_budget,
cost_budget_usd=cost_budget)
print(f"\nQuery: {query[:50]}...")
print(f"Model: {decision.selected_model}")
print(f"Reasoning: {decision.reasoning}")
print(f"Est. Cost: ${decision.estimated_cost_usd:.6f}")
print(f"Est. Latency: {decision.estimated_latency_ms}ms")Cost Monitoring and Alerting Infrastructure
In production, real-time cost monitoring is essential. Implement a monitoring layer that tracks spending patterns and alerts on anomalies:
#!/usr/bin/env python3
"""
Real-Time Cost Monitoring System for HolySheep AI
Implements budget alerts, anomaly detection, and usage analytics
"""
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from collections import defaultdict
import threading
import time
@dataclass
class CostAlert:
timestamp: datetime
alert_type: str # 'budget', 'anomaly', 'threshold'
message: str
current_spend: float
threshold: float
severity: str # 'info', 'warning', 'critical'
@dataclass
class DailyCost:
date: str
total_cost: float
request_count: int
avg_cost_per_request: float
model_breakdown: Dict[str, float]
modality_breakdown: Dict[str, float] # text, image, audio, video
class CostMonitor:
"""
Real-time cost monitoring and alerting for multimodal API usage.
Supports daily/monthly budgets, anomaly detection, and detailed analytics.
"""
# HolySheep AI 2026 Pricing Reference
PRICING = {
"deepseek-v3.2": {"text_input": 0.42, "text_output": 0.42, "vision": 8.00},
"gemini-2.5-flash": {"text_input": 2.50, "text_output": 2.50, "vision": 3.20},
"claude-sonnet-4.5": {"text_input": 15.00, "text_output": 15.00, "vision": 18.00},
"gpt-4.1": {"text_input": 8.00, "text_output": 8.00, "vision": 10.00}
}
def __init__(
self,
daily_budget_usd: float = 100.0,
monthly_budget_usd: float = 2000.0,
anomaly_threshold_pct: float = 50.0,
webhook_url: Optional[str] = None
):
self.daily_budget = daily_budget_usd
self.monthly_budget = monthly_budget_usd
self.anomaly_threshold = anomaly_threshold_pct
self.webhook_url = webhook_url
self._lock = threading.Lock()
self._requests: List[dict] = []
self._alerts: List[CostAlert] = []
self._daily_totals: Dict[str, float] = defaultdict(float)
self._monthly_totals: Dict[str, float] = defaultdict(float)
# Start background monitoring
self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
self._monitor_thread.start()
def track_request(
self,
model: str,
input_tokens: int,
output_tokens: int,
modality: str = "text",
user_id: Optional[str] = None,
request_id: Optional[str] = None
) -> float:
"""
Track an API request and calculate its cost.
Returns the cost in USD.
"""
with self._lock:
# Calculate cost based on modality
pricing = self.PRICING.get(model, {"text_input": 0.42, "text_output": 0.42, "vision": 8.00})
if modality == "text":
input_cost = (input_tokens / 1_000_000) * pricing["text_input"]
output_cost = (output_tokens / 1_000_000) * pricing["text_output"]
elif modality == "vision":
input_cost = (input_tokens / 1_000_000) * pricing["vision"]
output_cost = (output_tokens / 1_000_000) * pricing["vision"]
else:
input_cost = (input_tokens / 1_000_000) * pricing["text_input"]
output_cost = (output_tokens / 1_000_000) * pricing["text_output"]
total_cost = input_cost + output_cost
# Record request
today = datetime.now().strftime("%Y-%m-%d")
month = datetime.now().strftime("%Y-%m")
request = {
"timestamp": datetime.now(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": total_cost,
"modality": modality,
"user_id": user_id,
"request_id": request_id
}
self._requests.append(request)
self._daily_totals[today] += total_cost
self._monthly_totals[month] += total_cost
# Check thresholds
self._check_thresholds(total_cost)
return total_cost
def _check_thresholds(self, current_cost: float):
"""Check if any budget thresholds have been exceeded."""
today = datetime.now().strftime("%Y-%m-%d")
month = datetime.now().strftime("%Y-%m")
# Daily budget check
daily_spend = self._daily_totals[today]
daily_pct = (daily_spend / self.daily_budget) * 100
if daily_pct >= 100:
self._alerts.append(CostAlert(
timestamp=datetime.now(),
alert_type="budget",
message=f"Daily budget exceeded! Spent ${daily_spend:.2f} of ${self.daily_budget}",
current_spend=daily_spend,
threshold=self.daily_budget,
severity="critical"
))
elif daily_pct >= 80:
self._alerts.append(CostAlert(
timestamp=datetime.now(),
alert_type="threshold",
message=f"Daily budget at {daily_pct:.1f}% - Spent ${daily_spend:.2f}",
current_spend=daily_spend,
threshold=self.daily_budget,
severity="warning"
))
# Monthly budget check
monthly_spend = self._monthly_totals[month]
monthly_pct = (monthly_spend / self.monthly_budget) * 100
if monthly_pct >= 100:
self._alerts.append(CostAlert(
timestamp=datetime.now(),
alert_type="budget",
message=f"Monthly budget exceeded! Spent ${monthly_spend:.2f} of ${self.monthly_budget}",
current_spend=monthly_spend,
threshold=self.monthly_budget,
severity="critical"
))
elif monthly_pct >= 90:
self._alerts.append(CostAlert(
timestamp=datetime.now(),
alert_type="threshold",
message=f"Monthly budget at {monthly_pct:.1f}%",
current_spend=monthly_spend,
threshold=self.monthly_budget,
severity="warning"
))
def _monitor_loop(self):
"""Background loop for anomaly detection."""
while True:
time.sleep(60) # Check every minute
self._detect_anomalies()
def _detect_anomalies(self):
"""Detect unusual spending patterns."""
with self._lock:
if len(self._requests) < 10:
return
# Calculate rolling average
recent = self._requests[-100:]
avg_cost = sum(r["cost"] for r in recent) / len(recent)
current_cost = recent[-1]["cost"]
if current_cost > avg_cost * (1 + self.anomaly_threshold / 100):
self._alerts.append(CostAlert(
timestamp=datetime.now(),
alert_type="anomaly",
message=f"Unusual cost detected: ${current_cost:.4f} (avg: ${avg_cost:.4f})",
current_spend=current_cost,
threshold=avg_cost * (1 + self.anomaly_threshold / 100),
severity="warning"
))
def get_daily_report(self, days: int = 7) -> Dict:
"""Generate a daily cost report for the past N days."""
with self._lock:
today = datetime.now()
report = {"days": [], "totals": {}}
for i in range(days):
date = (today - timedelta(days=i)).strftime("%Y-%m-%d")
day_requests = [r for r in self._requests
if r["timestamp"].strftime("%Y-%m-%d") == date]
if day_requests:
model_breakdown = defaultdict(float)
modality_breakdown = defaultdict(float)
for r in day_requests:
model_breakdown[r["model"]] += r["cost"]
modality_breakdown[r["modality"]] += r["cost"]
total = sum(r["cost"] for r in day_requests)
report["days"].append(DailyCost(
date=date,
total_cost=total,
request_count=len(day_requests),
avg_cost_per_request=total / len(day_requests),
model_breakdown=dict(model_breakdown),
modality_breakdown=dict(modality_breakdown)
))
# Calculate totals
report["totals"]["total_cost"] = sum(d.total_cost for d in report["days"])
report["tot