Als ich vor sechs Monaten ein Enterprise RAG-System für einen E-Commerce-Kunden mit 2 Millionen monatlichen Nutzern launchte, erlebte ich einen Albtraum: Innerhalb von 3 Stunden explodierten die API-Kosten von geplanten 800€ auf über 12.000€ — ein einzelner fehlerhafter Loop im System sendete 47.000 Requests pro Minute. Dieser Vorfall motivierte mich, ein robustes System zur Anomalie-Erkennung aufzubauen, das ich heute mit Ihnen teilen möchte.
Warum API-Nutzungsüberwachung entscheidend ist
Bei HolySheep AI sehe ich täglich, wie Entwickler ohne Monitoring ihre Budgets überschreiten. Die Plattform bietet mit unter 50ms Latenz und Preisen ab $0.42 pro Million Tokens (DeepSeek V3.2) bereits extreme Kosteneffizienz — aber selbst die günstigste API wird teuer, wenn Sie Anomalien nicht frühzeitig erkennen.
Architektur des Anomalie-Erkennungssystems
Mein System basiert auf drei Säulen: Echtzeit-Metriken-Sammlung, statistische Anomalie-Erkennung und mehrstufige Alerting-Kanäle. Die folgende Architektur können Sie direkt übernehmen:
1. Grundlegendes API-Usage-Tracking
Der erste Schritt ist die kontinuierliche Erfassung aller API-Aufrufe. Mit HolySheep AI können Sie Ihre Requests tracken, indem Sie jede Interaktion protokollieren:
# requirements: pip install requests redis prometheus-client
import requests
import json
import time
from datetime import datetime
from collections import defaultdict
class HolySheepUsageTracker:
"""
tracks API usage in Echtzeit für HolySheep AI
Preise gültig 2026: DeepSeek V3.2 $0.42/MTok, GPT-4.1 $8/MTok
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.usage_stats = defaultdict(lambda: {
"requests": 0,
"input_tokens": 0,
"output_tokens": 0,
"cost": 0.0,
"latencies": []
})
self.anomaly_thresholds = {
"requests_per_minute": 1000,
"cost_per_hour": 50.0, # $50/Stunde als Alert-Schwelle
"p99_latency_ms": 2000
}
def call_chat_completion(self, messages: list, model: str = "deepseek-v3.2"):
"""
Wrapper für HolySheep Chat Completions mit Usage-Tracking
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048
}
start_time = time.time()
try:
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
end_time = time.time()
latency_ms = (end_time - start_time) * 1000
result = response.json()
# Usage-Daten extrahieren
usage = result.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
# Kosten berechnen basierend auf Modell
cost_per_mtok = self._get_cost_per_token(model)
cost = ((input_tokens + output_tokens) / 1_000_000) * cost_per_mtok
# Statistiken aktualisieren
timestamp = datetime.now().isoformat()
self._record_usage(
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost=cost,
latency_ms=latency_ms
)
# Anomalie-Check nach jedem Request
self._check_anomalies(model)
return result
except requests.exceptions.RequestException as e:
print(f"API-Fehler: {e}")
raise
def _get_cost_per_token(self, model: str) -> float:
"""
Kosten pro Million Tokens (2026 Preise)
"""
prices = {
"deepseek-v3.2": 0.42, # $0.42/MTok - günstigste Option
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50 # $2.50/MTok
}
return prices.get(model, 0.42)
def _record_usage(self, model: str, input_tokens: int,
output_tokens: int, cost: float, latency_ms: float):
"""
Interner Usage-Recorder mit 1-Minute-Granularität
"""
minute_key = datetime.now().strftime("%Y-%m-%d %H:%M")
self.usage_stats[model]["requests"] += 1
self.usage_stats[model]["input_tokens"] += input_tokens
self.usage_stats[model]["output_tokens"] += output_tokens
self.usage_stats[model]["cost"] += cost
self.usage_stats[model]["latencies"].append(latency_ms)
def _check_anomalies(self, model: str):
"""
Prüft auf Anomalien basierend auf konfigurierten Thresholds
"""
stats = self.usage_stats[model]
# Check 1: Requests pro Minute
recent_requests = len([l for l in stats["latencies"]
if time.time() - (stats["latencies"].index(l) * 60) < 60])
if recent_requests > self.anomaly_thresholds["requests_per_minute"]:
self._trigger_alert(
alert_type="HIGH_REQUEST_RATE",
model=model,
message=f"Ungewöhnlich hohe Request-Rate: {recent_requests}/min",
severity="WARNING"
)
# Check 2: Latenz-Anomalie
if stats["latencies"]:
avg_latency = sum(stats["latencies"]) / len(stats["latencies"])
if avg_latency > self.anomaly_thresholds["p99_latency_ms"]:
self._trigger_alert(
alert_type="HIGH_LATENCY",
model=model,
message=f"Durchschnittliche Latenz: {avg_latency:.2f}ms",
severity="WARNING"
)
def _trigger_alert(self, alert_type: str, model: str, message: str, severity: str):
"""
Alert-Handler - erweitern Sie dies für Slack, E-Mail, PagerDuty etc.
"""
timestamp = datetime.now().isoformat()
alert = {
"timestamp": timestamp,
"type": alert_type,
"model": model,
"severity": severity,
"message": message
}
print(f"🚨 ALERT [{severity}] {alert_type}: {message}")
# Hier könnten Sie Webhook-Aufrufe, E-Mail-Sendungen etc. implementieren
return alert
Beispiel-Nutzung
tracker = HolySheepUsageTracker(api_key="YOUR_HOLYSHEEP_API_KEY")
Test-Request
response = tracker.call_chat_completion(
messages=[{"role": "user", "content": "Erkläre RAG-Systeme"}],
model="deepseek-v3.2"
)
print(f"Antwort: {response['choices'][0]['message']['content'][:100]}...")2. Statistische Anomalie-Erkennung mit Z-Score
In der Praxis habe ich festgestellt, dass einfache Schwellenwerte nicht ausreichen. Deshalb implementierte ich eine statistische Anomalie-Erkennung basierend auf dem Z-Score-Algorithmus, der Abweichungen vom Normalverhalten automatisch erkennt:
import numpy as np
from typing import List, Tuple, Dict
from dataclasses import dataclass
from datetime import datetime, timedelta
import threading
import time
@dataclass
class AnomalyResult:
"""Struktur für Anomalie-Erkennungs-Ergebnisse"""
is_anomaly: bool
z_score: float
threshold_exceeded: bool
deviation_percentage: float
current_value: float
expected_value: float
severity: str # INFO, WARNING, CRITICAL
class StatisticalAnomalyDetector:
"""
Erkennt Anomalien in API-Nutzungsmustern mit Z-Score-Analyse
Kalibriert sich automatisch auf Ihre Nutzungsmuster
"""
def __init__(self, window_size: int = 100, z_threshold: float = 2.5):
"""
Args:
window_size: Anzahl der historischen Datenpunkte für Baseline
z_threshold: Z-Score-Schwelle für Anomalie-Erkennung (default: 2.5)
"""
self.window_size = window_size
self.z_threshold = z_threshold
self.baseline_data: Dict[str, List[float]] = {
"requests_per_minute": [],
"tokens_per_request": [],
"cost_per_hour": [],
"latency_ms": []
}
self.lock = threading.Lock()
# Seasonal baselines (z.B. für Geschäftszeiten)
self.seasonal_patterns = {}
def add_data_point(self, metric_name: str, value: float):
"""
Fügt neuen Datenpunkt zur Baseline hinzu
"""
with self.lock:
if metric_name not in self.baseline_data:
self.baseline_data[metric_name] = []
self.baseline_data[metric_name].append(value)
# Window-Größe begrenzen für Performance
if len(self.baseline_data[metric_name]) > self.window_size:
self.baseline_data[metric_name].pop(0)
def detect(self, metric_name: str, current_value: float) -> AnomalyResult:
"""
Führt Anomalie-Erkennung für eine Metrik durch
Returns:
AnomalyResult mit Erkennungsdetails
"""
with self.lock:
if metric_name not in self.baseline_data:
self.baseline_data[metric_name] = []
historical = self.baseline_data[metric_name]
if len(historical) < 10:
# Nicht genug Daten für zuverlässige Erkennung
return AnomalyResult(
is_anomaly=False,
z_score=0.0,
threshold_exceeded=False,
deviation_percentage=0.0,
current_value=current_value,
expected_value=current_value,
severity="INFO"
)
# Berechne Statistiken
mean = np.mean(historical)
std = np.std(historical)
if std == 0:
std = 1 # Prevent division by zero
# Z-Score berechnen
z_score = (current_value - mean) / std
# deviation percentage
deviation_pct = abs((current_value - mean) / mean * 100) if mean > 0 else 0
# Anomalie-Schwellen
is_anomaly = abs(z_score) > self.z_threshold
# Severity bestimmen
severity = self._determine_severity(abs(z_score), deviation_pct)
return AnomalyResult(
is_anomaly=is_anomaly,
z_score=z_score,
threshold_exceeded=is_anomaly,
deviation_percentage=deviation_pct,
current_value=current_value,
expected_value=mean,
severity=severity
)
def _determine_severity(self, abs_z_score: float, deviation_pct: float) -> str:
"""
Bestimmt Schweregrad basierend auf Z-Score und Abweichung
"""
if abs_z_score > 4.0 or deviation_pct > 500:
return "CRITICAL"
elif abs_z_score > 3.0 or deviation_pct > 200:
return "WARNING"
elif abs_z_score > 2.0 or deviation_pct > 100:
return "INFO"
return "NORMAL"
def get_baseline_stats(self, metric_name: str) -> Dict:
"""
Gibt Statistiken für die Baseline einer Metrik zurück
"""
with self.lock:
data = self.baseline_data.get(metric_name, [])
if len(data) < 2:
return {"status": "insufficient_data"}
return {
"count": len(data),
"mean": round(np.mean(data), 4),
"std": round(np.std(data), 4),
"min": round(np.min(data), 4),
"max": round(np.max(data), 4),
"p50": round(np.percentile(data, 50), 4),
"p95": round(np.percentile(data, 95), 4),
"p99": round(np.percentile(data, 99), 4)
}
class CostAnomalyMonitor:
"""
Spezialisierter Monitor für API-Kosten mit Budget-Alerting
"""
def __init__(self, hourly_budget: float = 50.0, daily_budget: float = 500.0):
"""
Args:
hourly_budget: Budget-Limit pro Stunde in Dollar (default: $50)
daily_budget: Budget-Limit pro Tag in Dollar (default: $500)
"""
self.hourly_budget = hourly_budget
self.daily_budget = daily_budget
self.hourly_costs: List[Tuple[datetime, float]] = []
self.daily_costs: List[Tuple[datetime, float]] = []
self.detector = StatisticalAnomalyDetector()
# Kosten-Projektion
self.projected_daily_cost = 0.0
def record_cost(self, cost: float, timestamp: datetime = None):
"""
Zeichnet einen Kostenpunkt auf und prüft auf Budget-Überschreitung
"""
if timestamp is None:
timestamp = datetime.now()
self.hourly_costs.append((timestamp, cost))
self.daily_costs.append((timestamp, cost))
# Alte Daten bereinigen (älter als 25 Stunden)
cutoff_hourly = timestamp - timedelta(hours=25)
self.hourly_costs = [(t, c) for t, c in self.hourly_costs if t > cutoff_hourly]
# Projektion berechnen
self._calculate_projection()
# Anomalie-Check
return self._check_cost_anomaly()
def _calculate_projection(self):
"""
Projiziert die täglichen Kosten basierend auf aktuellen Verbrauch
"""
now = datetime.now()
hour_start = now.replace(minute=0, second=0, microsecond=0)
# Kosten seit Mitternacht
midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
today_costs = sum(c for t, c in self.daily_costs if t >= midnight)
# Stunden seit Mitternacht (mindestens 1 für Division)
hours_passed = max((now - midnight).seconds / 3600, 1)
# Projektion
self.projected_daily_cost = (today_costs / hours_passed) * 24
def _check_cost_anomaly(self) -> Dict:
"""
Prüft auf Budget-Überschreitungen und Kostenanomalien
"""
now = datetime.now()
hour_start = now.replace(minute=0, second=0, microsecond=0)
# Aktuelle Stunde Kosten
current_hour_costs = sum(c for t, c in self.hourly_costs
if t >= hour_start)
alerts = []
# Check 1: Stündliches Budget
if current_hour_costs > self.hourly_budget:
alerts.append({
"type": "HOURLY_BUDGET_EXCEEDED",
"current": current_hour_costs,
"limit": self.hourly_budget,
"severity": "CRITICAL" if current_hour_costs > self.hourly_budget * 2 else "WARNING"
})
# Check 2: Tägliche Projektion
if self.projected_daily_cost > self.daily_budget:
alerts.append({
"type": "DAILY_PROJECTION_EXCEEDED",
"projected": self.projected_daily_cost,
"limit": self.daily_budget,
"severity": "CRITICAL"
})
# Check 3: Statistische Anomalie
self.detector.add_data_point("cost_per_hour", current_hour_costs)
anomaly_result = self.detector.detect("cost_per_hour", current_hour_costs)
if anomaly_result.is_anomaly:
alerts.append({
"type": "STATISTICAL_ANOMALY",
"z_score": anomaly_result.z_score,
"expected": anomaly_result.expected_value,
"current": current_hour_costs,
"severity": anomaly_result.severity
})
return {
"current_hour_cost": round(current_hour_costs, 4),
"projected_daily_cost": round(self.projected_daily_cost, 2),
"hourly_budget": self.hourly_budget,
"daily_budget": self.daily_budget,
"alerts": alerts
}
def get_cost_breakdown(self) -> Dict:
"""
Gibt detaillierte Kostenaufstellung zurück
"""
stats = self.detector.get_baseline_stats("cost_per_hour")
now = datetime.now()
midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
today_total = sum(c for t, c in self.daily_costs if t >= midnight)
return {
"today_total_usd": round(today_total, 4),
"hourly_average_usd": round(today_total / max((now - midnight).seconds / 3600, 1), 4),
"projected_daily_usd": round(self.projected_daily_cost, 2),
"hourly_baseline_stats": stats,
"hourly_budget_remaining": round(self.hourly_budget - sum(c for t, c in self.hourly_costs
if t >= now.replace(minute=0, second=0, microsecond=0)), 2)
}
Praktisches Beispiel: E-Commerce KI-Kundenservice
def demo_ecommerce_cost_monitoring():
"""
Demonstrates cost monitoring for an E-commerce AI customer service
simulating peak hours and anomaly detection
"""
monitor = CostAnomalyMonitor(hourly_budget=50.0, daily_budget=500.0)
print("=== E-Commerce KI-Kundenservice Monitoring Demo ===\n")
# Simuliere normale Nutzung während Geschäftszeiten
normal_costs = [0.45, 0.52, 0.48, 0.55, 0.42, 0.51, 0.49, 0.53]
print("Phase 1: Normale Nutzung (Geschäftszeiten)")
for i, cost in enumerate(normal_costs):
result = monitor.record_cost(cost)
print(f" Request {i+1}: ${cost:.2f} - Alerts: {len(result['alerts'])}")
# Simuliere Kostenexplosion (wie in meinem Projekt)
print("\nPhase 2: Unerwartete Kostenexplosion (47.000 Requests/min)")
explosion_costs = [2.50, 8.30, 15.20, 32.80, 68.40]
for i, cost in enumerate(explosion_costs):
result = monitor.record_cost(cost)
print(f" Burst {i+1}: ${cost:.2f}")
print(f" → Projektion: ${result['projected_daily_cost']:.2f}/Tag")
if result['alerts']:
for alert in result['alerts']:
print(f" 🚨 {alert['type']}: Severity={alert['severity']}")
# Zeige finale Statistiken
breakdown = monitor.get_cost_breakdown()
print("\n=== Kostenübersicht ===")
print(f" Heute gesamt: ${breakdown['today_total_usd']:.2f}")
print(f" Projektion: ${breakdown['projected_daily_usd']:.2f}/Tag")
print(f" Baselines: {breakdown['hourly_baseline_stats']}")
demo_ecommerce_cost_monitoring()3. Multi-Channel Alerting-System
Basierend auf meiner Erfahrung aus dem eingangs erwähnten Projekt habe ich ein Alerting-System entwickelt, das verschiedene Kanäle unterstützt und unterschiedliche Eskalationsstufen hat:
import smtplib
import json
import logging
from enum import Enum
from typing import List, Dict, Callable, Optional
from dataclasses import dataclass, field
from datetime import datetime
import threading
import queue
class AlertSeverity(Enum):
"""Alert-Schweregrade nach ITU-Standard"""
INFO = 1
WARNING = 2
ERROR = 3
CRITICAL = 4
class AlertChannel(Enum):
"""Unterstützte Alert-Kanäle"""
CONSOLE = "console"
EMAIL = "email"
SLACK = "slack"
WEBHOOK = "webhook"
PAGERDUTY = "pagerduty"
@dataclass
class Alert:
"""Standardisierte Alert-Struktur"""
timestamp: datetime
severity: AlertSeverity
channel: AlertChannel
title: str
message: str
metadata: Dict = field(default_factory=dict)
acknowledged: bool = False
resolved: bool = False
class AlertManager:
"""
Zentrales Alert-Management-System für API-Überwachung
Unterstützt Multi-Channel-Alerting und automatische Eskalation
"""
def __init__(self):
self.alerts: List[Alert] = []
self.handlers: Dict[AlertChannel, Callable] = {}
self.alert_queue: queue.Queue = queue.Queue()
self.escalation_rules: Dict[AlertSeverity, int] = {
AlertSeverity.INFO: 300, # 5 Minuten bis Eskalation
AlertSeverity.WARNING: 120, # 2 Minuten
AlertSeverity.ERROR: 30, # 30 Sekunden
AlertSeverity.CRITICAL: 0 # Sofort eskalieren
}
self.running = False
self._lock = threading.Lock()
# Logging konfigurieren
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
# Standard-Handler registrieren
self._register_default_handlers()
def _register_default_handlers(self):
"""Registriert Standard-Alert-Handler"""
self.register_handler(AlertChannel.CONSOLE, self._console_handler)
def register_handler(self, channel: AlertChannel, handler: Callable):
"""Registriert einen Handler für einen Alert-Kanal"""
self.handlers[channel] = handler
self.logger.info(f"Handler für {channel.value} registriert")
def send_alert(self, severity: AlertSeverity, title: str, message: str,
channel: AlertChannel = AlertChannel.CONSOLE,
metadata: Optional[Dict] = None):
"""
Sendet einen Alert über den konfigurierten Kanal
Args:
severity: Schweregrad des Alerts
title: Kurztitel
message: Detaillierte Nachricht
channel: Alert-Kanal
metadata: Zusätzliche Metadaten
"""
alert = Alert(
timestamp=datetime.now(),
severity=severity,
channel=channel,
title=title,
message=message,
metadata=metadata or {}
)
with self._lock:
self.alerts.append(alert)
# Alert verarbeiten
self._process_alert(alert)
return alert
def _process_alert(self, alert: Alert):
"""Verarbeitet einen Alert durch alle relevanten Handler"""
handler = self.handlers.get(alert.channel)
if handler:
try:
handler(alert)
except Exception as e:
self.logger.error(f"Handler-Fehler für {alert.channel.value}: {e}")
else:
self.logger.warning(f"Kein Handler für Kanal {alert.channel.value}")
# Prüfe auf Eskalationsbedarf
if alert.severity == AlertSeverity.CRITICAL:
self._trigger_escalation(alert)
def _console_handler(self, alert: Alert):
"""Konsolen-Handler für Alert-Ausgabe"""
severity_emoji = {
AlertSeverity.INFO: "ℹ️",
AlertSeverity.WARNING: "⚠️",
AlertSeverity.ERROR: "❌",
AlertSeverity.CRITICAL: "🚨"
}
emoji = severity_emoji.get(alert.severity, "📢")
timestamp = alert.timestamp.strftime("%Y-%m-%d %H:%M:%S")
print(f"\n{emoji} [{timestamp}] {alert.severity.name}")
print(f" {alert.title}")
print(f" {alert.message}")
if alert.metadata:
print(f" Metadaten: {json.dumps(alert.metadata, indent=2)}")
def _trigger_escalation(self, alert: Alert):
"""Eskaliert kritische Alerts automatisch"""
escalation_channels = [
AlertChannel.EMAIL,
AlertChannel.SLACK,
AlertChannel.PAGERDUTY
]
for channel in escalation_channels:
if channel in self.handlers and channel != alert.channel:
# Kopie des Alerts für jeden Kanal
escalated = Alert(
timestamp=datetime.now(),
severity=AlertSeverity.CRITICAL,
channel=channel,
title=f"ESCALATED: {alert.title}",
message=f"Original-Kanal: {alert.channel.value}\n{alert.message}",
metadata={**alert.metadata, "escalated_from": alert.channel.value}
)
self._process_alert(escalated)
def configure_slack(self, webhook_url: str, channel: str = "#api-alerts"):
"""
Konfiguriert Slack-Alerting
Args:
webhook_url: Slack Webhook URL
channel: Slack-Kanal
"""
def slack_handler(alert: Alert):
payload = {
"channel": channel,
"username": "HolySheep AI Monitor",
"icon_emoji": self._get_severity_emoji(alert.severity),
"attachments": [{
"color": self._get_severity_color(alert.severity),
"title": f"{self._get_severity_emoji(alert.severity)} {alert.title}",
"text": alert.message,
"fields": [
{"title": "Severity", "value": alert.severity.name, "short": True},
{"title": "Time", "value": alert.timestamp.isoformat(), "short": True}
],
"footer": "HolySheep AI Usage Monitor"
}]
}
try:
response = requests.post(
webhook_url,
data=json.dumps(payload),
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
self.logger.info(f"Slack-Alert gesendet: {alert.title}")
except requests.exceptions.RequestException as e:
self.logger.error(f"Slack-Fehler: {e}")
self.register_handler(AlertChannel.SLACK, slack_handler)
def configure_email(self, smtp_host: str, smtp_port: int,
sender: str, recipients: List[str],
username: str, password: str):
"""
Konfiguriert E-Mail-Alerting
"""
def email_handler(alert: Alert):
subject = f"[{alert.severity.name}] {alert.title}"
body = f"""
API Usage Alert - HolySheep AI Monitor
=====================================
Zeitpunkt: {alert.timestamp.isoformat()}
Schweregrad: {alert.severity.name}
Nachricht:
{alert.message}
Metadaten:
{json.dumps(alert.metadata, indent=2)}
---
Diese Nachricht wurde automatisch generiert.
"""
try:
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.starttls()
server.login(username, password)
for recipient in recipients:
server.sendmail(sender, recipient,
f"Subject: {subject}\n\n{body}")
self.logger.info(f"E-Mail-Alert gesendet an {recipients}")
except Exception as e:
self.logger.error(f"E-Mail-Fehler: {e}")
self.register_handler(AlertChannel.EMAIL, email_handler)
def configure_webhook(self, webhook_url: str, secret: str = None):
"""
Konfiguriert generisches Webhook-Alerting
"""
def webhook_handler(alert: Alert):
payload = {
"event": "api_usage_alert",
"timestamp": alert.timestamp.isoformat(),
"severity": alert.severity.name,
"title": alert.title,
"message": alert.message,
"metadata": alert.metadata
}
headers = {"Content-Type": "application/json"}
if secret:
import hmac
import hashlib
signature = hmac.new(
secret.encode(),
json.dumps(payload).encode(),
hashlib.sha256
).hexdigest()
headers["X-Signature"] = signature
try:
response = requests.post(
webhook_url,
json=payload,
headers=headers
)
response.raise_for_status()
self.logger.info(f"Webhook-Alert gesendet: {alert.title}")
except requests.exceptions.RequestException as e:
self.logger.error(f"Webhook-Fehler: {e}")
self.register_handler(AlertChannel.WEBHOOK, webhook_handler)
@staticmethod
def _get_severity_emoji(severity: AlertSeverity) -> str:
"""Gibt Emoji basierend auf Schweregrad zurück"""
return {
AlertSeverity.INFO: "ℹ️",
AlertSeverity.WARNING: "⚠️",
AlertSeverity.ERROR: "❌",
AlertSeverity.CRITICAL: "🚨"
}.get(severity, "📢")
@staticmethod
def _get_severity_color(severity: AlertSeverity) -> str:
"""Gibt Farbe basierend auf Schweregrad zurück"""
return {
AlertSeverity.INFO: "#36a64f",
AlertSeverity.WARNING: "#ff9800",
AlertSeverity.ERROR: "#f44336",
AlertSeverity.CRITICAL: "#b71c1c"
}.get(severity, "#808080")
def get_alert_summary(self, hours: int = 24) -> Dict:
"""Gibt Zusammenfassung der Alerts der letzten Stunden zurück"""
cutoff = datetime.now() - timedelta(hours=hours)
recent_alerts = [a for a in self.alerts if a.timestamp >= cutoff]
return {
"total_alerts": len(recent_alerts),
"by_severity": {
s.name: len([a for a in recent_alerts if a.severity == s])
for s in AlertSeverity
},
"unacknowledged": len([a for a in recent_alerts if not a.acknowledged]),
"critical_count": len([a for a in recent_alerts
if a.severity == AlertSeverity.CRITICAL])
}
Praktisches Beispiel
def demo_multi_channel_alerting():
"""Demonstriert Multi-Channel Alerting"""
manager = AlertManager()
# Slack konfigurieren (Beispiel)
# manager.configure_slack("https://hooks.slack.com/services/XXX/YYY/ZZZ")
# Webhook konfigurieren (Beispiel)
# manager.configure_webhook("https://your-webhook-endpoint.com/alerts")
print("=== Multi-Channel Alerting Demo ===\n")
# Simuliere verschiedene Alert-Szenarien
scenarios = [
(AlertSeverity.INFO, "Hohe Nutzung erkannt",
"Request-Rate überstieg 500/min -可能是正常峰值"),
(AlertSeverity.WARNING, "Budget zu 75% ausgeschöpft",
"Tagesbudget $500 bereits bei $375 — Projektion: $600"),
(AlertSeverity.ERROR, "Latenz-Anomalie",
"Durchschnittliche Latenz: 2340ms (Normal: 45ms)"),
(AlertSeverity.CRITICAL, "KOSTENEXPLOSION ERKANNT",
"47.000 Requests/min! Sofortige Intervention erforderlich!")
]
for severity, title, message in scenarios:
manager.send_alert(
severity=severity,
title=title,
message=message,
channel=AlertChannel.CONSOLE,
metadata={
"api_endpoint": "https://api.holysheep.ai/v1/chat/completions",
"model": "deepseek-v3.2",
"estimated_cost_impact": 1200.00
}
)
# Alert-Zusammenfassung
summary = manager.get_alert_summary()
print(f"\n=== Alert-Zusammenfassung (24h) ===")
print(f" Gesamt: {summary['total_alerts']}")
print(f" Nach Schweregrad: {summary['by_severity']}")
print(f" Unbestätigt: {summary['unacknowledged']}")
print(f" Kritisch: {summary['critical_count']}")
demo_multi_channel_alerting()