Last month, I watched my development budget spiral out of control without realizing it. Our team was running automated tests throughout the day, and by the 15th, we had already burned through our entire monthly budget. The surprise bill arrived at $847 — nearly 3x our planned $300 monthly spend. That painful experience taught me the critical importance of real-time AI API cost monitoring with automated alerts.
In this guide, I will walk you through building a robust cost monitoring and alerting system that integrates with HolySheep AI and other providers, ensuring you never face an unexpected bill again. The system tracks usage in real-time, sends instant notifications when spending approaches thresholds, and includes auto-shutdown capabilities for extreme overages.
Why Real-Time Cost Monitoring Matters
Modern AI API pricing can escalate rapidly. With models like GPT-4.1 at $8 per million tokens and Claude Sonnet 4.5 at $15 per million tokens, a single runaway loop can cost hundreds of dollars in hours. HolySheep AI offers rates starting at $1 per dollar equivalent — an 85%+ savings compared to ¥7.3 market rates — with support for WeChat and Alipay payments and latency under 50ms, but even budget-friendly providers require monitoring to prevent surprises.
Our monitoring system will cover:
- Real-time spend tracking per API endpoint and model
- Configurable budget thresholds with percentage-based alerts
- Multi-channel notifications (email, Slack, webhook)
- Automatic API key rotation or rate limiting on overage
- Daily and monthly budget summaries with forecasting
Building the Cost Monitoring System
Project Structure and Dependencies
First, set up a Python project with the necessary dependencies. Create a virtual environment and install the required packages:
# requirements.txt
requests>=2.28.0
python-dotenv>=1.0.0
slack-sdk>=3.21.0
APScheduler>=3.10.0
httpx>=0.24.0
pandas>=2.0.0
pip install -r requirements.txt
mkdir cost_monitor
cd cost_monitor
touch monitor.py config.py notifiers.py models.py
Configuration Management
Create a robust configuration system that supports multiple AI providers and budget rules:
# config.py
import os
from dataclasses import dataclass, field
from typing import List, Dict, Optional
@dataclass
class BudgetThreshold:
percentage: float # 0.5 = 50%, 0.75 = 75%, 1.0 = 100%
alert_channels: List[str]
action: Optional[str] = None # "email", "slack", "webhook", "shutdown"
@dataclass
class ProviderConfig:
name: str
base_url: str
api_key: str
monthly_budget: float
thresholds: List[BudgetThreshold] = field(default_factory=list)
class MonitoringConfig:
def __init__(self):
self.providers: List[ProviderConfig] = [
ProviderConfig(
name="HolySheep AI",
base_url="https://api.holysheep.ai/v1",
api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
monthly_budget=300.00,
thresholds=[
BudgetThreshold(percentage=0.50, alert_channels=["email", "slack"]),
BudgetThreshold(percentage=0.75, alert_channels=["slack", "webhook"]),
BudgetThreshold(percentage=0.90, alert_channels=["email", "slack"], action="warn"),
BudgetThreshold(percentage=1.00, alert_channels=["email", "slack", "webhook"], action="shutdown"),
]
),
]
self.slack_webhook = os.getenv("SLACK_WEBHOOK_URL")
self.slack_bot_token = os.getenv("SLACK_BOT_TOKEN")
self.alert_email = os.getenv("ALERT_EMAIL")
self.check_interval_minutes = 15
# Webhook for custom alerting systems
self.webhook_url = os.getenv("ALERT_WEBHOOK_URL")
config = MonitoringConfig()
HolySheep API Integration with Cost Tracking
The core of our system integrates with the HolySheep AI API to track actual usage. Here's a comprehensive client that handles authentication, usage retrieval, and cost calculation:
# monitor.py
import requests
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
@dataclass
class UsageRecord:
timestamp: datetime
prompt_tokens: int
completion_tokens: int
model: str
cost: float
@dataclass
class SpendSummary:
provider: str
current_spend: float
budget: float
percentage_used: float
usage_records: List[UsageRecord]
period_start: datetime
period_end: datetime
class HolySheepCostMonitor:
"""
Monitor HolySheep AI API costs with real-time usage tracking.
HolySheep AI offers $1 per rate unit with <50ms latency.
"""
# Pricing model mapping (2026 rates in USD per million tokens)
MODEL_PRICING = {
"gpt-4.1": {"prompt": 2.00, "completion": 6.00}, # $8 per 1M tokens total
"claude-sonnet-4.5": {"prompt": 3.75, "completion": 11.25}, # $15 per 1M tokens
"gemini-2.5-flash": {"prompt": 0.625, "completion": 1.875}, # $2.50 per 1M tokens
"deepseek-v3.2": {"prompt": 0.14, "completion": 0.28}, # $0.42 per 1M tokens
"default": {"prompt": 1.00, "completion": 3.00}, # HolySheep standard rate
}
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
self._usage_cache: List[UsageRecord] = []
def test_connection(self) -> Tuple[bool, Optional[str]]:
"""
Test API connection with a minimal request.
This catches 401 Unauthorized errors before they become budget problems.
"""
try:
# Test with a small completion request
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": "deepseek-v3.2", # Cheapest model for testing
"messages": [{"role": "user", "content": "test"}],
"max_tokens": 5
},
timeout=10
)
if response.status_code == 401:
return False, "401 Unauthorized - Invalid API key. Check your HOLYSHEEP_API_KEY."
elif response.status_code == 403:
return False, "403 Forbidden - API key may lack required permissions."
elif response.status_code == 429:
return True, "Connection successful but rate limited (429). Consider reducing request frequency."
elif response.status_code == 200:
return True, "Connection successful"
else:
return False, f"Unexpected response: {response.status_code} - {response.text}"
except requests.exceptions.Timeout:
return False, "Connection timeout - check network or firewall settings."
except requests.exceptions.ConnectionError as e:
return False, f"Connection refused - ensure {self.base_url} is accessible. Error: {str(e)}"
except Exception as e:
return False, f"Unexpected error: {str(e)}"
def calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""Calculate cost based on token usage and model pricing."""
pricing = self.MODEL_PRICING.get(model, self.MODEL_PRICING["default"])
prompt_cost = (prompt_tokens / 1_000_000) * pricing["prompt"]
completion_cost = (completion_tokens / 1_000_000) * pricing["completion"]
return round(prompt_cost + completion_cost, 6)
def get_usage_stats(self, start_date: Optional[datetime] = None) -> SpendSummary:
"""
Retrieve usage statistics for the billing period.
Falls back to estimated costs if API doesn't provide usage data.
"""
if start_date is None:
# Default to current billing period (monthly)
today = datetime.now()
start_date = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
try:
# Attempt to get usage from HolySheep API
response = self.session.get(
f"{self.base_url}/usage",
params={"start_date": start_date.isoformat()},
timeout=30
)
if response.status_code == 200:
data = response.json()
return self._parse_usage_response(data, start_date)
else:
# Fallback: estimate from cached usage
return self._estimate_spend_from_cache(start_date)
except requests.exceptions.RequestException as e:
print(f"Warning: Could not fetch usage stats: {e}")
return self._estimate_spend_from_cache(start_date)
def _parse_usage_response(self, data: Dict, period_start: datetime) -> SpendSummary:
"""Parse API response into SpendSummary."""
total_cost = 0.0
records = []
for item in data.get("data", []):
cost = self.calculate_cost(
item.get("model", "default"),
item.get("prompt_tokens", 0),
item.get("completion_tokens", 0)
)
total_cost += cost
records.append(UsageRecord(
timestamp=datetime.fromisoformat(item.get("timestamp", datetime.now().isoformat())),
prompt_tokens=item.get("prompt_tokens", 0),
completion_tokens=item.get("completion_tokens", 0),
model=item.get("model", "unknown"),
cost=cost
))
return SpendSummary(
provider="HolySheep AI",
current_spend=total_cost,
budget=300.00, # From config
percentage_used=0.0,
usage_records=records,
period_start=period_start,
period_end=datetime.now()
)
def _estimate_spend_from_cache(self, period_start: datetime) -> SpendSummary:
"""Estimate spend from locally cached usage records."""
total_cost = sum(r.cost for r in self._usage_cache
if r.timestamp >= period_start)
return SpendSummary(
provider="HolySheep AI",
current_spend=total_cost,
budget=300.00,
percentage_used=0.0,
usage_records=self._usage_cache,
period_start=period_start,
period_end=datetime.now()
)
def record_usage(self, model: str, prompt_tokens: int, completion_tokens: int):
"""Manually record usage for tracking when API doesn't provide it."""
cost = self.calculate_cost(model, prompt_tokens, completion_tokens)
self._usage_cache.append(UsageRecord(
timestamp=datetime.now(),
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
model=model,
cost=cost
))
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] "
f"Recorded usage: {model} - {prompt_tokens + completion_tokens} tokens - ${cost:.4f}")
Example usage
if __name__ == "__main__":
monitor = HolySheepCostMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
# Test connection first
success, message = monitor.test_connection()
print(f"Connection test: {message}")
if success:
# Record sample usage
monitor.record_usage("deepseek-v3.2", 1500, 350)
monitor.record_usage("gemini-2.5-flash", 8000, 1200)
# Get current spend
summary = monitor.get_usage_stats()
print(f"\nCurrent spend: ${summary.current_spend:.2f}")
print(f"Budget: ${summary.budget:.2f}")
Notification System with Multi-Channel Support
The alerting system supports email, Slack, and webhooks with rate limiting to prevent notification spam:
# notifiers.py
import smtplib
import json
import time
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import List, Dict, Optional
import requests
class AlertRateLimiter:
"""Prevent alert fatigue with smart rate limiting."""
def __init__(self, min_interval_seconds: int = 300): # 5 minutes minimum
self.min_interval = min_interval_seconds
self.last_alerts: Dict[str, float] = {}
def can_alert(self, alert_key: str) -> bool:
now = time.time()
if alert_key in self.last_alerts:
if now - self.last_alerts[alert_key] < self.min_interval:
return False
self.last_alerts[alert_key] = now
return True
class NotifierSystem:
def __init__(self, config):
self.config = config
self.rate_limiter = AlertRateLimiter(min_interval_seconds=300)
# Slack client
self.slack_client = None
if config.slack_bot_token:
try:
from slack_sdk import WebClient
self.slack_client = WebClient(token=config.slack_bot_token)
except ImportError:
print("Warning: slack-sdk not installed. Slack notifications disabled.")
def send_budget_alert(self, provider: str, current_spend: float,
budget: float, percentage: float, threshold: float,
action: Optional[str] = None):
"""Send budget alert through all configured channels."""
alert_key = f"{provider}_{threshold}"
if not self.rate_limiter.can_alert(alert_key):
print(f"Alert rate-limited for {alert_key}")
return
message = self._format_alert_message(provider, current_spend, budget, percentage, threshold)
# Determine which channels to use based on threshold
channels = self._get_channels_for_threshold(threshold)
for channel in channels:
try:
if channel == "email":
self._send_email_alert(message, provider, percentage)
elif channel == "slack":
self._send_slack_alert(message, provider, percentage)
elif channel == "webhook":
self._send_webhook_alert(message, provider, current_spend, budget)
except Exception as e:
print(f"Failed to send {channel} alert: {e}")
# Handle critical actions
if action == "shutdown":
self._trigger_emergency_shutdown(provider)
def _format_alert_message(self, provider: str, current: float,
budget: float, percentage: float, threshold: float) -> str:
return f"""
🚨 *BUDGET ALERT - {provider}*
Current Spend: ${current:.2f}
Monthly Budget: ${budget:.2f}
Usage: {percentage:.1%}
Threshold Hit: {threshold:.0%}
Remaining Budget: ${budget - current:.2f}
Daily Forecast: ${self._forecast_daily_spend(current):.2f}
Action Required: {"Immediate" if percentage >= 1.0 else "Review recommended"}
"""
def _forecast_daily_spend(self, current_spend: float) -> float:
"""Estimate daily spend based on current month usage."""
day = datetime.now().day
if day > 0:
return current_spend / day
return current_spend
def _get_channels_for_threshold(self, threshold: float) -> List[str]:
if threshold >= 1.0:
return ["email", "slack", "webhook"]
elif threshold >= 0.75:
return ["slack", "webhook"]
elif threshold >= 0.50:
return ["email", "slack"]
return ["slack"]
def _send_email_alert(self, message: str, provider: str, percentage: float):
"""Send email alert via SMTP."""
if not self.config.alert_email:
return
# In production, use a proper email configuration
# This is a simplified version
print(f"[EMAIL] Would send to {self.config.alert_email}: {message}")
def _send_slack_alert(self, message: str, provider: str, percentage: float):
"""Send Slack notification with rich formatting."""
if self.slack_client and self.config.slack_webhook:
# Use webhook for simplicity
requests.post(self.config.slack_webhook, json={
"text": message,
"blocks": [
{
"type": "header",
"text": {"type": "plain_text", "text": f"⚠️ Budget Alert: {provider}"}
},
{"type": "section", "text": {"type": "mrkdwn", "text": message}},
{
"type": "actions",
"elements": [
{"type": "button", "text": {"type": "plain_text", "text": "View Dashboard"},
"url": "https://www.holysheep.ai/dashboard"}
]
}
]
})
elif self.config.slack_webhook:
requests.post(self.config.slack_webhook, json={"text": message})
else:
print(f"[SLACK] Webhook not configured. Message: {message}")
def _send_webhook_alert(self, message: str, provider: str,
current: float, budget: float):
"""Send alert to custom webhook endpoint."""
if not self.config.webhook_url:
return
payload = {
"alert_type": "budget_threshold",
"provider": provider,
"current_spend": current,
"budget": budget,
"message": message.strip(),
"timestamp": datetime.now().isoformat()
}
try:
response = requests.post(
self.config.webhook_url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=10
)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Webhook delivery failed: {e}")
def _trigger_emergency_shutdown(self, provider: str):
"""Emergency action: disable API access when budget is exceeded."""
# In production, implement actual key revocation or rate limiting
print(f"""
🚨🚨🚨 EMERGENCY SHUTDOWN INITIATED 🚨🚨🚨
Provider: {provider}
Action: API key rotation/revocation recommended
""")
from datetime import datetime
Main Monitoring Loop with Scheduled Checks
Here's the main monitoring script that ties everything together with scheduled checks:
# main_monitor.py
import os
import time
import logging
from datetime import datetime
from threading import Thread
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from config import config, ProviderConfig
from monitor import HolySheepCostMonitor
from notifiers import NotifierSystem
Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("CostMonitor")
class CostMonitoringService:
"""
Main service for monitoring AI API costs across providers.
Runs scheduled checks and triggers alerts when thresholds are crossed.
"""
def __init__(self):
self.monitors: Dict[str, HolySheepCostMonitor] = {}
self.notifier = NotifierSystem(config)
self.scheduler = BackgroundScheduler()
self.last_alert_state: Dict[str, float] = {} # Track which thresholds already alerted
# Initialize monitors for each provider
for provider_config in config.providers:
self.monitors[provider_config.name] = HolySheepCostMonitor(
api_key=provider_config.api_key,
base_url=provider_config.base_url
)
def check_all_providers(self):
"""Check spending for all configured providers."""
logger.info("Running scheduled cost check...")
for provider_config in config.providers:
try:
self._check_provider_spending(provider_config)
except Exception as e:
logger.error(f"Error checking {provider_config.name}: {e}")
def _check_provider_spending(self, provider_config: ProviderConfig):
"""Check spending for a single provider and trigger alerts if needed."""
monitor = self.monitors[provider_config.name]
# Test connection first
success, message = monitor.test_connection()
if not success:
logger.warning(f"Connection issue with {provider_config.name}: {message}")
# Could send alert here for connection issues
# Get current usage
summary = monitor.get_usage_stats()
summary.budget = provider_config.monthly_budget
summary.percentage_used = summary.current_spend / provider_config.monthly_budget
logger.info(f"{provider_config.name}: ${summary.current_spend:.2f} / "
f"${summary.budget:.2f} ({summary.percentage_used:.1%})")
# Check each threshold
for threshold in provider_config.thresholds:
percentage = threshold.percentage
# Only alert if we cross this threshold (not if we're still above it)
last_state = self.last_alert_state.get(f"{provider_config.name}_{percentage}", 0)
if summary.percentage_used >= percentage and last_state < percentage:
logger.warning(f"Threshold crossed: {percentage:.0%} - ${summary.current_spend:.2f}")
self.notifier.send_budget_alert(
provider=provider_config.name,
current_spend=summary.current_spend,
budget=provider_config.monthly_budget,
percentage=summary.percentage_used,
threshold=percentage,
action=threshold.action
)
self.last_alert_state[f"{provider_config.name}_{percentage}"] = summary.percentage_used
def start(self):
"""Start the monitoring service."""
logger.info("Starting Cost Monitoring Service...")
# Schedule regular checks
self.scheduler.add_job(
self.check_all_providers,
IntervalTrigger(minutes=config.check_interval_minutes),
id='cost_check',
name='Check API spending',
replace_existing=True
)
# Run initial check
self.check_all_providers()
# Start scheduler in background thread
self.scheduler.start()
logger.info(f"Monitoring started. Checks every {config.check_interval_minutes} minutes.")
# Keep main thread alive
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
self.stop()
def stop(self):
"""Stop the monitoring service."""
logger.info("Stopping Cost Monitoring Service...")
self.scheduler.shutdown()
logger.info("Service stopped.")
Entry point
if __name__ == "__main__":
service = CostMonitoringService()
service.start()
Setting Up Environment Variables
Create a .env file to store your sensitive configuration:
# .env file - DO NOT COMMIT TO VERSION CONTROL
HOLYSHEEP_API_KEY=your_holysheep_api_key_here
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL
SLACK_BOT_TOKEN=xoxb-your-slack-bot-token
[email protected]
ALERT_WEBHOOK_URL=https://your-webhook-endpoint.com/alerts
Budget configuration
DEFAULT_MONTHLY_BUDGET=300
CHECK_INTERVAL_MINUTES=15
# .gitignore
.env
__pycache__/
*.pyc
*.log
Docker Deployment for Production Use
For production environments, containerize the monitoring system:
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
Copy application code
COPY *.py .
Run as non-root user
RUN useradd -m -u 1000 monitor
USER monitor
Health check
HEALTHCHECK --interval=5m --timeout=10s --start-period=30s \
CMD python -c "import requests; requests.get('http://localhost:8000/health')" || exit 1
CMD ["python", "main_monitor.py"]
# docker-compose.yml for production deployment
version: '3.8'
services:
cost-monitor:
build: .
container_name: ai-cost-monitor
restart: unless-stopped
env_file:
- .env
volumes:
- ./logs:/app/logs
networks:
- monitoring
# Optional: Prometheus metrics exporter
prometheus-exporter:
image: prometheus/node-exporter:latest
container_name: prometheus-exporter
ports:
- "9100:9100"
networks:
- monitoring
networks:
monitoring:
driver: bridge
Common Errors and Fixes
Error 1: "401 Unauthorized - Invalid API key"
This is the most common issue when setting up cost monitoring. The HolySheep API returns 401 when authentication fails.
# Quick diagnosis and fix
import requests
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
BASE_URL = "https://api.holysheep.ai/v1"
Test authentication
response = requests.get(
f"{BASE_URL}/auth/status",
headers={"Authorization": f"Bearer {API_KEY}"}
)
print(f"Status: {response.status_code}")
print(f"Response: {response.text}")
Common fixes:
1. Check for leading/trailing whitespace in key
2. Verify key is active in dashboard: https://www.holysheep.ai/dashboard
3. Regenerate key if compromised
4. Ensure environment variable is loaded: source .env
Fix: Verify your API key is correctly set in the environment variable and has not expired. Regenerate from the HolySheep dashboard if necessary.
Error 2: "Connection refused - [Errno 111] Connection refused"
This network-level error indicates the API endpoint is unreachable.
# Debug connection issues
import socket
import requests
BASE_URL = "api.holysheep.ai"
PORT = 443
Test DNS resolution
try:
ip = socket.gethostbyname(BASE_URL)
print(f"DNS resolved to: {ip}")
except socket.gaierror as e:
print(f"DNS resolution failed: {e}")
Test TCP connection
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
result = sock.connect_ex((BASE_URL, PORT))
sock.close()
if result == 0:
print("Port 443 is open")
else:
print(f"Port 443 is blocked (error code: {result})")
except Exception as e:
print(f"Connection test failed: {e}")
Verify no firewall/proxy issues
response = requests.get(
"https://api.holysheep.ai/v1/models",
timeout=10,
headers={"Authorization": "Bearer YOUR_KEY"}
)
print(f"API response: {response.status_code}")
Fix: Check firewall rules, corporate proxy settings, or VPN configurations. Ensure outbound connections to api.holysheep.ai on port 443 are allowed.
Error 3: "429 Too Many Requests - Rate Limit Exceeded"
Excessive monitoring requests can trigger rate limiting.
# Implement exponential backoff for rate limit handling
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry(max_retries=3, backoff_factor=1):
session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
Usage in monitor
session = create_session_with_retry(max_retries=3, backoff_factor=2)
try:
response = session.get(
"https://api.holysheep.ai/v1/usage",
headers={"Authorization": "Bearer YOUR_KEY"},
timeout=30
)
response.raise_for_status()
except requests.exceptions.RetryError as e:
print(f"All retries exhausted: {e}")
# Fall back to cached data
Fix: Reduce monitoring frequency, implement exponential backoff, and cache responses to minimize API calls. Consider using webhooks instead of polling.
Error 4: "AttributeError: 'NoneType' object has no attribute 'get'"
This occurs when parsing API responses that don't match expected format.
# Safe response parsing with defaults
def safe_get_usage_data(response_json):
"""
Safely extract usage data with fallback defaults.
Handles missing fields and None values gracefully.
"""
default_response = {
"data": [],
"total_cost": 0.0,
"billing_period": {
"start": None,
"end": None
}
}
if response_json is None:
return default_response
return {
"data": response_json.get("data") or [],
"total_cost": response_json.get("total_cost") or 0.0,
"billing_period": response_json.get("billing_period") or {},
"currency": response_json.get("currency", "USD")
}
Usage
response = requests.get(url, headers=headers)
data = safe_get_usage_data(response.json())
for item in data["data"]:
tokens = (item.get("prompt_tokens") or 0) + (item.get("completion_tokens") or 0)
print(f"Model: {item.get('model', 'unknown')}, Tokens: {tokens}")
Fix: Always use .get() with default values when parsing JSON responses. Implement schema validation for API responses.
Advanced: Cost Forecasting and Anomaly Detection
Beyond basic threshold alerts, implement predictive monitoring to catch issues before they occur:
# anomaly_detection.py
import statistics
from datetime import datetime, timedelta
from typing import List, Dict
class CostForecaster:
"""Predict future costs based on historical usage patterns."""
def __init__(self, history_days: int = 30):
self.history_days = history_days
self.daily_spends: List[float] = []
def add_daily_spend(self, date: datetime, amount: float):
"""Record daily spend for analysis."""
self.daily_spends.append(amount)
# Keep only relevant history
cutoff = datetime.now() - timedelta(days=self.history_days)
# Filter logic here
def predict_monthly_end(self) -> Dict:
"""Predict end-of-month spend based on current trajectory."""
if not self.daily_spends:
return {"predicted_spend": 0, "confidence": "low"}
avg_daily = statistics.mean(self.daily_spends)
std_daily = statistics.stdev(self.daily_spends) if len(self.daily_spends) > 1 else 0
days_remaining = 30 - datetime.now().day
predicted_additional = avg_daily * days_remaining
current_month = sum(self.daily_spends)
total_predicted = current_month + predicted_additional
# Spike detection
if self.daily_spends:
recent = self.daily_spends[-7:] # Last 7 days
if statistics.mean(recent) > avg_daily * 1.5:
return {
"predicted_spend": total_predicted * 1.3, # Account for spike
"confidence": "medium",
"anomaly_detected": True,
"recommendation": "Usage spike detected - investigate cause"
}
return {
"predicted_spend": total_predicted,
"confidence": "high" if len(self.daily_spends) >= 14 else "medium",
"avg_daily_spend": avg_daily,
"days_remaining": days_remaining
}
Testing Your Monitoring System
Before deploying to production, thoroughly test your monitoring setup:
# test_monitor.py
import unittest
from unittest.mock import Mock, patch, MagicMock
from monitor import HolySheepCostMonitor
from notifiers import NotifierSystem, AlertRateLimiter
class TestCostMonitor(unittest.TestCase):
def setUp(self):
self.monitor = HolySheepCostMonitor(api_key="test_key")
def test_cost_calculation_deepseek(self):
"""Test DeepSeek V3.2 pricing ($0.42/M tokens total)."""
cost = self.monitor.calculate_cost("deepseek-v3.2", 1000, 500)
expected = (1000 / 1_000_000) * 0.14 + (500 / 1_000_000) * 0.28
self.assertAlmostEqual(cost, expected, places=5)
def test_cost_calculation_gpt41(self):
"""Test GPT-4.1 pricing ($8/M tokens total)."""
cost = self.monitor.calculate_cost("gpt-4.1", 100000, 50000)
expected = (100000 / 1_000_000) * 2 + (50000 / 1_000_000) * 6
self.assertAlmostEqual(cost, expected, places=4)
def test_connection_handles_401(self):
"""Test that 401 errors are properly handled."""
with patch('requests.Session') as mock_session:
mock_response = Mock()
mock_response.status_code = 401
mock_response.text = "Invalid API key"
mock_session.return_value.post.return_value = mock_response
success, message = self.monitor.test_connection()
self.assertFalse(success)
self.assertIn("401", message)
class TestRateLimiter(unittest.TestCase):
def test_rate_limit_prevents_spam(self):
"""Test that rate limiter prevents duplicate alerts."""
limiter = AlertRateLimiter(min_interval_seconds=5)
# First alert should pass
self.assertTrue(lim