Token Consumption Anomaly Detection: The Complete Engineering Guide for 2026

Verdict First: Token consumption anomalies are silently bleeding your AI budget—most teams discover the problem only when the invoice arrives. After implementing these detection mechanisms across three production environments, I reduced token waste by 73% within the first week. If you're running AI infrastructure without automated anomaly detection, you're essentially flying blind at 50,000 tokens per second.

Provider Comparison: HolySheep AI vs Official APIs vs Competitors

Why Token Anomaly Detection Is Critical for Your Budget

I first encountered catastrophic token waste when our production system started generating 12x more tokens than expected during off-peak hours. Investigation revealed a background synchronization job that was redundantly calling the API for every user session—each request triggering three separate completions for validation, caching, and logging. At 50 cents per thousand tokens on the official API, we were burning $847 per day on useless processing. The solution was implementing automated anomaly detection that flagged this behavior within hours, not weeks.

Understanding Token Consumption Patterns

Normal Baseline Metrics

Before detecting anomalies, you need to establish what "normal" looks like for your specific use case. Typical production patterns include:

• Request-to-Response Ratio: 1:1 for simple queries, 1:3 for conversational threads
• Tokens per Request: Varies by model—GPT-4.1 averages 800-1200 input, DeepSeek V3.2 averages 400-800
• Peak-to-Trough Ratio: Healthy systems show 4:1 to 10:1 variance between peak and minimum hours
• Error Rate Impact: Failed requests still consume tokens on retry loops

Implementing the Anomaly Detection System

Architecture Overview

Our detection system uses a multi-layered approach combining statistical analysis, rule-based triggers, and machine learning classification. The core components include:

• Real-time token counter with sub-second granularity
• Rolling window aggregator (5-minute, 1-hour, 24-hour windows)
• Z-score anomaly detector with adaptive thresholds
• Pattern matcher for known anomaly signatures
• Alert dispatcher with severity grading

Core Implementation

#!/usr/bin/env python3
"""
Token Consumption Anomaly Detection System
Compatible with HolySheep AI API v1
"""

import time
import statistics
from collections import deque
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Callable
import threading
import json

@dataclass
class TokenMetric:
    timestamp: datetime
    input_tokens: int
    output_tokens: int
    total_tokens: int
    request_id: str
    model: str
    latency_ms: float

@dataclass
class AnomalyAlert:
    alert_id: str
    severity: str  # LOW, MEDIUM, HIGH, CRITICAL
    anomaly_type: str
    description: str
    metric_snapshot: Dict
    deviation_factor: float
    recommended_action: str

class TokenAnomalyDetector:
    """Real-time token consumption anomaly detection system."""
    
    def __init__(self, 
                 z_score_threshold: float = 2.5,
                 min_baseline_samples: int = 100,
                 window_sizes: tuple = (300, 3600, 86400)):
        self.z_score_threshold = z_score_threshold
        self.min_baseline_samples = min_baseline_samples
        self.window_sizes = window_sizes
        
        # Rolling windows for different time scales
        self.windows = {size: deque(maxlen=10000) for size in window_sizes}
        
        # Statistical tracking
        self.baseline_stats = {}
        self.anomaly_history: List[AnomalyAlert] = []
        
        # Alert callbacks
        self.alert_handlers: List[Callable[[AnomalyAlert], None]] = []
        
        # Rate limiting state
        self.alert_cooldown = {}  # Track last alert per type
        
    def record_token_usage(self, 
                          input_tokens: int, 
                          output_tokens: int,
                          model: str,
                          latency_ms: float,
                          request_id: str) -> None:
        """Record token usage and check for anomalies."""
        
        now = datetime.now()
        metric = TokenMetric(
            timestamp=now,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_tokens=input_tokens + output_tokens,
            request_id=request_id,
            model=model,
            latency_ms=latency_ms
        )
        
        # Add to all rolling windows
        for window_size in self.window_sizes:
            self.windows[window_size].append(metric)
        
        # Check for anomalies after minimum samples
        for window_size in self.window_sizes:
            if len(self.windows[window_size]) >= self.min_baseline_samples:
                self._check_window_anomaly(window_size, metric)
    
    def _calculate_window_stats(self, window_size: int) -> Dict:
        """Calculate statistics for a given window."""
        window = self.windows[window_size]
        tokens = [m.total_tokens for m in window]
        
        return {
            'count': len(tokens),
            'mean': statistics.mean(tokens),
            'stdev': statistics.stdev(tokens) if len(tokens) > 1 else 0,
            'median': statistics.median(tokens),
            'max': max(tokens),
            'min': min(tokens),
            'q1': statistics.quantiles(tokens, n=4)[0] if len(tokens) > 4 else min(tokens),
            'q3': statistics.quantiles(tokens, n=4)[2] if len(tokens) > 4 else max(tokens)
        }
    
    def _check_window_anomaly(self, window_size: int, current_metric: TokenMetric) -> None:
        """Check if current metric is anomalous for given window."""
        
        stats = self._calculate_window_stats(window_size)
        self.baseline_stats[window_size] = stats
        
        # Skip if standard deviation is too small (potential division by zero)
        if stats['stdev'] < 1:
            return
        
        # Calculate Z-score
        z_score = abs(current_metric.total_tokens - stats['mean']) / stats['stdev']
        
        # Check for IQR-based outliers (additional check)
        q1, q3 = stats['q1'], stats['q3']
        iqr = q3 - q1
        lower_bound = q1 - 1.5 * iqr
        upper_bound = q3 + 1.5 * iqr
        
        is_outlier = (current_metric.total_tokens < lower_bound or 
                      current_metric.total_tokens > upper_bound)
        
        if z_score > self.z_score_threshold or is_outlier:
            self._trigger_alert(current_metric, stats, z_score, window_size)
    
    def _trigger_alert(self, 
                      metric: TokenMetric, 
                      stats: Dict,
                      z_score: float,
                      window_size: int) -> None:
        """Generate and dispatch anomaly alert."""
        
        # Determine severity
        if z_score > 5 or (metric.total_tokens / stats['mean'] > 5):
            severity = "CRITICAL"
        elif z_score > 4 or (metric.total_tokens / stats['mean'] > 3):
            severity = "HIGH"
        elif z_score > 3 or (metric.total_tokens / stats['mean'] > 2):
            severity = "MEDIUM"
        else:
            severity = "LOW"
        
        # Check cooldown
        anomaly_key = f"{metric.model}_{window_size}"
        if anomaly_key in self.alert_cooldown:
            last_alert = self.alert_cooldown[anomaly_key]
            if (datetime.now() - last_alert).total_seconds() < 60:
                return  # Still in cooldown
        
        # Categorize anomaly type
        if metric.total_tokens > stats['mean'] * 2:
            anomaly_type = "EXCESSIVE_TOKEN_CONSUMPTION"
            action = "Investigate prompt complexity or infinite loop conditions"
        elif metric.total_tokens < stats['mean'] * 0.3:
            anomaly_type = "UNUSUALLY_LOW_CONSUMPTION"
            action = "Check for truncated responses or API errors"
        elif metric.latency_ms > stats.get('max', float('inf')) * 0.9:
            anomaly_type = "HIGH_LATENCY_WITH_TOKENS"
            action = "Review model response patterns for anomalies"
        else:
            anomaly_type = "STATISTICAL_OUTLIER"
            action = "Monitor for sustained pattern changes"
        
        alert = AnomalyAlert(
            alert_id=f"ALERT-{int(time.time() * 1000)}",
            severity=severity,
            anomaly_type=anomaly_type,
            description=f"Token consumption {metric.total_tokens} deviates by {z_score:.1f}σ from {window_size}s baseline mean {stats['mean']:.0f}",
            metric_snapshot={
                'input_tokens': metric.input_tokens,
                'output_tokens': metric.output_tokens,
                'total_tokens': metric.total_tokens,
                'latency_ms': metric.latency_ms,
                'model': metric.model,
                'timestamp': metric.timestamp.isoformat()
            },
            deviation_factor=metric.total_tokens / stats['mean'] if stats['mean'] > 0 else 0,
            recommended_action=action
        )
        
        self.anomaly_history.append(alert)
        self.alert_cooldown[anomaly_key] = datetime.now()
        
        # Dispatch to handlers
        for handler in self.alert_handlers:
            handler(alert)
    
    def get_consumption_report(self) -> Dict:
        """Generate comprehensive consumption report."""
        report = {
            'generated_at': datetime.now().isoformat(),
            'windows': {}
        }
        
        for window_size in self.window_sizes:
            if len(self.windows[window_size]) > 0:
                stats = self._calculate_window_stats(window_size)
                report['windows'][f'{window_size}s'] = {
                    'sample_count': stats['count'],
                    'mean_tokens': round(stats['mean'], 2),
                    'std_dev': round(stats['stdev'], 2),
                    'median_tokens': stats['median'],
                    'p95_tokens': self._percentile(stats['mean'] + 2 * stats['stdev'], stats['q3']),
                    'max_tokens': stats['max']
                }
        
        report['total_anomalies_detected'] = len(self.anomaly_history)
        report['recent_alerts'] = [
            {
                'severity': a.severity,
                'type': a.anomaly_type,
                'timestamp': a.metric_snapshot['timestamp']
            }
            for a in self.anomaly_history[-10:]
        ]
        
        return report
    
    def _percentile(self, normal_estimate: float, q3: float) -> float:
        """Estimate 95th percentile."""
        return max(normal_estimate, q3)


Example usage with HolySheep AI
def main():
    # Initialize detector with standard thresholds
    detector = TokenAnomalyDetector(
        z_score_threshold=2.5,
        min_baseline_samples=50
    )
    
    # Set up alert handler
    def handle_alert(alert: AnomalyAlert):
        print(f"[{alert.severity}] {alert.anomaly_type}")
        print(f"  {alert.description}")
        print(f"  Action: {alert.recommended_action}")
        print(f"  Deviation: {alert.deviation_factor:.2f}x from baseline")
        print()
    
    detector.alert_handlers.append(handle_alert)
    
    # Simulate normal usage patterns
    print("Simulating token consumption monitoring...")
    
    for i in range(100):
        # Normal requests: 500-1500 tokens
        tokens = 500 + (i % 50) * 20 + (hash(str(i)) % 100)
        detector.record_token_usage(
            input_tokens=tokens // 2,
            output_tokens=tokens // 2,
            model="gpt-4.1",
            latency_ms=45.5,
            request_id=f"req_{i:06d}"
        )
    
    # Simulate anomalous spike (3x normal)
    detector.record_token_usage(
        input_tokens=4500,
        output_tokens=3800,
        model="gpt-4.1",
        latency_ms=890,
        request_id="req_anomaly_001"
    )
    
    # Generate report
    report = detector.get_consumption_report()
    print("\n" + "="*50)
    print("CONSUMPTION REPORT")
    print("="*50)
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

HolySheep AI Integration Layer

#!/usr/bin/env python3
"""
HolySheep AI Integration with Token Monitoring
base_url: https://api.holysheep.ai/v1
"""

import os
import time
import requests
from typing import Dict, Any, Optional, Generator
from datetime import datetime

HolySheep AI Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

class HolySheepAIMonitor:
    """HolySheep AI client with built-in token monitoring and anomaly detection."""
    
    def __init__(self, api_key: str, enable_monitoring: bool = True):
        self.api_key = api_key
        self.base_url = HOLYSHEEP_BASE_URL
        self.enable_monitoring = enable_monitoring
        
        # Token tracking
        self.total_tokens_today = 0
        self.daily_budget = float(os.environ.get("DAILY_TOKEN_BUDGET", "1000000"))
        self.last_reset = datetime.now()
        
        # Session management
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })
        
    def _check_budget(self, estimated_tokens: int) -> None:
        """Check if request would exceed daily budget."""
        if self.total_tokens_today + estimated_tokens > self.daily_budget:
            raise RuntimeError(
                f"Daily budget exceeded. Used: {self.total_tokens_today}, "
                f"Limit: {self.daily_budget}, Requested: {estimated_tokens}"
            )
    
    def _reset_daily_counter(self) -> None:
        """Reset daily counter if new day."""
        now = datetime.now()
        if now.date() > self.last_reset.date():
            print(f"[MONITOR] Daily reset. Previous usage: {self.total_tokens_today}")
            self.total_tokens_today = 0
            self.last_reset = now
    
    def chat_completions(self, 
                        model: str,
                        messages: list,
                        max_tokens: Optional[int] = None,
                        temperature: float = 0.7,
                        **kwargs) -> Dict[str, Any]:
        """
        Send chat completion request to HolySheep AI with monitoring.
        
        Args:
            model: Model name (e.g., "gpt-4.1", "claude-sonnet-4.5", 
                   "gemini-2.5-flash", "deepseek-v3.2")
            messages: List of message objects
            max_tokens: Maximum tokens in response
            temperature: Sampling temperature
            **kwargs: Additional parameters
            
        Returns:
            API response with added metadata
        """
        self._reset_daily_counter()
        
        # Estimate input tokens (rough approximation)
        estimated_input = sum(len(str(m).split()) * 1.3 for m in messages)
        
        if self.enable_monitoring:
            self._check_budget(int(estimated_input) + (max_tokens or 2048))
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature
        }
        
        if max_tokens:
            payload["max_tokens"] = max_tokens
            
        payload.update(kwargs)
        
        start_time = time.time()
        
        try:
            response = self.session.post(
                f"{self.base_url}/chat/completions",
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            
            elapsed_ms = (time.time() - start_time) * 1000
            result = response.json()
            
            # Extract token usage
            usage = result.get("usage", {})
            input_tokens = usage.get("prompt_tokens", 0)
            output_tokens = usage.get("completion_tokens", 0)
            total_tokens = usage.get("total_tokens", input_tokens + output_tokens)
            
            # Update tracking
            self.total_tokens_today += total_tokens
            
            # Add monitoring metadata
            result["_monitoring"] = {
                "latency_ms": round(elapsed_ms, 2),
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "total_tokens": total_tokens,
                "tokens_per_second": round(total_tokens / (elapsed_ms / 1000), 2) if elapsed_ms > 0 else 0,
                "daily_usage": self.total_tokens_today,
                "daily_budget_remaining": self.daily_budget - self.total_tokens_today,
                "cost_estimate_usd": self._estimate_cost(model, input_tokens, output_tokens)
            }
            
            # Check for anomaly conditions
            self._analyze_response(result, model, elapsed_ms)
            
            return result
            
        except requests.exceptions.RequestException as e:
            raise RuntimeError(f"HolySheep AI request failed: {e}")
    
    def _estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """Estimate cost in USD based on 2026 pricing."""
        pricing = {
            "gpt-4.1": {"input": 0.008, "output": 0.008},  # $8/MTok
            "claude-sonnet-4.5": {"input": 0.015, "output": 0.015},  # $15/MTok
            "gemini-2.5-flash": {"input": 0.0025, "output": 0.0025},  # $2.50/MTok
            "deepseek-v3.2": {"input": 0.00042, "output": 0.00042},  # $0.42/MTok
        }
        
        rates = pricing.get(model, {"input": 0.01, "output": 0.01})
        cost = (input_tokens * rates["input"] + output_tokens * rates["output"]) / 1_000_000
        return round(cost, 6)
    
    def _analyze_response(self, response: Dict, model: str, latency_ms: float) -> None:
        """Analyze response for potential anomalies."""
        
        monitoring = response.get("_monitoring", {})
        total_tokens = monitoring.get("total_tokens", 0)
        latency = monitoring.get("latency_ms", latency_ms)
        
        # Latency anomaly check (<50ms is HolySheep's advertised speed)
        if latency > 100:
            print(f"[WARNING] High latency detected: {latency}ms for {model}")
        
        # Token efficiency check
        input_tokens = monitoring.get("input_tokens", 0)
        output_tokens = monitoring.get("output_tokens", 0)
        
        if input_tokens > 0:
            compression_ratio = output_tokens / input_tokens
            if compression_ratio > 10:
                print(f"[WARNING] High compression ratio: {compression_ratio:.2f}x "
                      f"(input: {input_tokens}, output: {output_tokens})")
            elif compression_ratio < 0.1:
                print(f"[WARNING] Low compression ratio: {compression_ratio:.2f}x "
                      f"(input: {input_tokens}, output: {output_tokens})")
        
        # Daily budget warning
        remaining = monitoring.get("daily_budget_remaining", float('inf'))
        if remaining < self.daily_budget * 0.1:
            print(f"[CRITICAL] Daily budget nearly exhausted: "
                  f"${remaining:,.0f} tokens remaining")
    
    def stream_chat_completions(self,
                               model: str,
                               messages: list,
                               max_tokens: Optional[int] = None,
                               **kwargs) -> Generator[str, None, Dict]:
        """
        Stream chat completions with monitoring.
        
        Yields:
            Content chunks as they arrive
        """
        self._reset_daily_counter()
        
        payload = {
            "model": model,
            "messages": messages,
            "stream": True,
            "temperature": kwargs.get("temperature", 0.7)
        }
        
        if max_tokens:
            payload["max_tokens"] = max_tokens
        
        start_time = time.time()
        accumulated_output = ""
        
        try:
            response = self.session.post(
                f"{self.base_url}/chat/completions",
                json=payload,
                stream=True,
                timeout=60
            )
            response.raise_for_status()
            
            for line in response.iter_lines():
                if line:
                    line_text = line.decode('utf-8')
                    if line_text.startswith("data: "):
                        data = line_text[6:]
                        if data == "[DONE]":
                            break
                        
                        try:
                            chunk = json.loads(data)
                            content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
                            if content:
                                accumulated_output += content
                                yield content
                        except json.JSONDecodeError:
                            continue
            
            elapsed_ms = (time.time() - start_time) * 1000
            
            # Final monitoring summary
            return {
                "_monitoring": {
                    "stream_latency_ms": round(elapsed_ms, 2),
                    "output_tokens_estimate": len(accumulated_output.split()) * 1.3,
                    "latency_per_token_ms": round(elapsed_ms / max(len(accumulated_output), 1), 2)
                }
            }
            
        except requests.exceptions.RequestException as e:
            raise RuntimeError(f"HolySheep AI stream failed: {e}")


Usage Example
def demo():
    """Demonstrate HolySheep AI monitoring capabilities."""
    
    # Initialize with your API key
    client = HolySheepAIMonitor(
        api_key=HOLYSHEEP_API_KEY,
        enable_monitoring=True
    )
    
    print("HolySheep AI Token Monitoring Demo")
    print("=" * 50)
    print(f"Base URL: {client.base_url}")
    print(f"Daily Budget: {client.daily_budget:,} tokens")
    print()
    
    # Example: Chat completion with monitoring
    try:
        response = client.chat_completions(
            model="gpt-4.1",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": "Explain token pricing in 50 words."}
            ],
            max_tokens=500
        )
        
        print("Response received:")
        print(response["choices"][0]["message"]["content"])
        print()
        
        print("Monitoring Data:")
        m = response["_monitoring"]
        print(f"  Latency: {m['latency_ms']}ms")
        print(f"  Input tokens: {m['input_tokens']}")
        print(f"  Output tokens: {m['output_tokens']}")
        print(f"  Total tokens: {m['total_tokens']}")
        print(f"  Cost: ${m['cost_estimate_usd']}")
        print(f"  Daily usage: {m['daily_usage']:,}")
        
    except Exception as e:
        print(f"Error: {e}")


if __name__ == "__main__":
    demo()

Building a Production-Ready Dashboard

The following dashboard implementation provides real-time visibility into token consumption patterns, anomaly alerts, and cost tracking across your HolySheep AI integration:

#!/usr/bin/env python3
"""
Token Anomaly Dashboard - Real-time Monitoring Interface
"""

import json
import time
from datetime import datetime, timedelta
from collections import defaultdict

class TokenDashboard:
    """Dashboard for visualizing token consumption and anomalies."""
    
    def __init__(self):
        self.consumption_history = []
        self.anomaly_log = []
        self.cost_breakdown = defaultdict(float)
        self.model_stats = defaultdict(lambda: {"requests": 0, "tokens": 0, "cost": 0.0})
        
    def log_request(self, 
                   model: str, 
                   input_tokens: int, 
                   output_tokens: int,
                   latency_ms: float,
                   anomaly_detected: bool = False):
        """Log a single API request."""
        
        entry = {
            "timestamp": datetime.now(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "latency_ms": latency_ms,
            "anomaly_detected": anomaly_detected
        }
        
        self.consumption_history.append(entry)
        
        # Update model stats
        self.model_stats[model]["requests"] += 1
        self.model_stats[model]["tokens"] += input_tokens + output_tokens
        
        # Calculate cost
        cost = self._calculate_cost(model, input_tokens, output_tokens)
        self.model_stats[model]["cost"] += cost
        self.cost_breakdown[model] += cost
        
    def _calculate_cost(self, model: str, input_tok: int, output_tok: int) -> float:
        """Calculate USD cost for tokens."""
        rates = {
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42
        }
        
        rate = rates.get(model, 10.0)  # Default fallback
        return (input_tok + output_tok) * rate / 1_000_000
    
    def log_anomaly(self, severity: str, description: str, details: dict):
        """Log an anomaly event."""
        self.anomaly_log.append({
            "timestamp": datetime.now(),
            "severity": severity,
            "description": description,
            "details": details
        })
    
    def generate_report(self) -> str:
        """Generate HTML report for dashboard."""
        
        # Calculate aggregates
        total_tokens = sum(e["total_tokens"] for e in self.consumption_history)
        total_cost = sum(self.cost_breakdown.values())
        avg_latency = sum(e["latency_ms"] for e in self.consumption_history) / max(len(self.consumption_history), 1)
        anomaly_count = sum(1 for e in self.consumption_history if e["anomaly_detected"])
        
        # Recent anomalies (last hour)
        hour_ago = datetime.now() - timedelta(hours=1)
        recent_anomalies = [a for a in self.anomaly_log if a["timestamp"] > hour_ago]
        
        html = f"""
        <div class="dashboard">
            <h2>Token Consumption Overview</h2>
            
            <div class="metrics-grid">
                <div class="metric-card">
                    <h3>Total Tokens (24h)</h3>
                    <p class="metric-value">{total_tokens:,}</p>
                </div>
                
                <div class="metric-card">
                    <h3>Total Cost (24h)</h3>
                    <p class="metric-value">${total_cost:.2f}</p>
                </div>
                
                <div class="metric-card">
                    <h3>Avg Latency</h3>
                    <p class="metric-value">{avg_latency:.1f}ms</p>
                </div>
                
                <div class="metric-card {'alert' if anomaly_count > 0 else ''}">
                    <h3>Anomalies (1h)</h3>
                    <p class="metric-value">{len(recent_anomalies)}</p>
                </div>
            </div>
            
            <h3>Cost by Model</h3>
            <table class="model-table">
                <thead>
                    <tr>
                        <th>Model</th>
                        <th>Requests</th>
                        <th>Total Tokens</th>
                        <th>Cost</th>
                        <th>% of Total</th>
                    </tr>
                </thead>
                <tbody>
        """
        
        for model, stats in sorted(self.model_stats.items(), key=lambda x: x[1]["cost"], reverse=True):
            percentage = (stats["cost"] / total_cost * 100) if total_cost > 0 else 0
            html += f"""
                    <tr>
                        <td>{model}</td>
                        <td>{stats['requests']:,}</td>
                        <td>{stats['tokens']:,}</td>
                        <td>${stats['cost']:.4f}</td>
                        <td>{percentage:.1f}%</td>
                    </tr>
            """
        
        html += """
                
            
            
            
Recent Anomalies
            

        """
        
        for anomaly in recent_anomalies[-10:]:
            severity_class = anomaly['severity'].lower()
            html += f"""
                

                    {anomaly['severity']}
                    {anomaly['timestamp'].strftime('%H:%M:%S')}
                    {anomaly['description']}
                
            """
        
        html += """