AI Foundations & Data Strategy

Step 1: Data Preparation

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib

# Load network metrics data
def prepare_network_data():
    """
    Simulate network metrics data
    In production, this would connect to your monitoring system
    """
    data = {
        'packet_loss': np.random.uniform(0, 0.05, 1000),
        'latency_ms': np.random.normal(20, 5, 1000),
        'bandwidth_utilization': np.random.uniform(0.1, 0.9, 1000),
        'connection_count': np.random.poisson(100, 1000),
        'error_rate': np.random.exponential(0.01, 1000)
    }
    
    # Add some anomalies
    data['packet_loss'][::50] = np.random.uniform(0.1, 0.3, 20)
    data['latency_ms'][::100] = np.random.uniform(100, 200, 10)
    
    return pd.DataFrame(data)

df = prepare_network_data()
print(f"Prepared {len(df)} network samples")
                

Step 2: Train Anomaly Detection Model

class NetworkAnomalyDetector:
    def __init__(self, contamination=0.05):
        """
        Initialize anomaly detector
        contamination: expected proportion of anomalies
        """
        self.scaler = StandardScaler()
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42
        )
        
    def train(self, data):
        """Train the anomaly detection model"""
        # Scale features
        X_scaled = self.scaler.fit_transform(data)
        
        # Train Isolation Forest
        self.model.fit(X_scaled)
        
        # Get anomaly scores
        scores = self.model.score_samples(X_scaled)
        predictions = self.model.predict(X_scaled)
        
        return {
            'n_anomalies': sum(predictions == -1),
            'anomaly_rate': sum(predictions == -1) / len(predictions),
            'min_score': scores.min(),
            'max_score': scores.max()
        }
    
    def predict(self, data):
        """Predict anomalies in new data"""
        X_scaled = self.scaler.transform(data)
        prediction = self.model.predict(X_scaled)
        score = self.model.score_samples(X_scaled)
        
        return {
            'is_anomaly': prediction[0] == -1,
            'anomaly_score': float(score[0]),
            'risk_level': self._get_risk_level(score[0])
        }
    
    def _get_risk_level(self, score):
        """Convert anomaly score to risk level"""
        if score > -0.1:
            return "low"
        elif score > -0.3:
            return "medium"
        else:
            return "high"
    
    def save(self, path):
        """Save model to disk"""
        joblib.dump({
            'model': self.model,
            'scaler': self.scaler
        }, path)
    
    def load(self, path):
        """Load model from disk"""
        data = joblib.load(path)
        self.model = data['model']
        self.scaler = data['scaler']

# Train the model
detector = NetworkAnomalyDetector()
results = detector.train(df)
print(f"Training complete: {results}")

# Save the model
detector.save('network_anomaly_model.pkl')
                

Step 3: Create FastAPI Service

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any
import uvicorn

app = FastAPI(title="Network Anomaly Detection API")

# Load trained model
detector = NetworkAnomalyDetector()
detector.load('network_anomaly_model.pkl')

class NetworkMetrics(BaseModel):
    packet_loss: float
    latency_ms: float
    bandwidth_utilization: float
    connection_count: int
    error_rate: float

class AnomalyResponse(BaseModel):
    is_anomaly: bool
    anomaly_score: float
    risk_level: str
    recommendations: list

@app.get("/")
def read_root():
    return {
        "service": "Network Anomaly Detection",
        "version": "1.0.0",
        "status": "operational"
    }

@app.post("/analyze", response_model=AnomalyResponse)
def analyze_metrics(metrics: NetworkMetrics):
    """
    Analyze network metrics for anomalies
    """
    # Prepare data for model
    data = pd.DataFrame([metrics.dict()])
    
    # Get prediction
    result = detector.predict(data)
    
    # Generate recommendations based on risk level
    recommendations = []
    if result['risk_level'] == "high":
        recommendations = [
            "Immediate investigation required",
            "Check for DDoS attacks",
            "Review firewall logs",
            "Verify upstream connectivity"
        ]
    elif result['risk_level'] == "medium":
        recommendations = [
            "Monitor closely for next 30 minutes",
            "Review recent configuration changes",
            "Check bandwidth utilization trends"
        ]
    else:
        recommendations = [
            "Continue normal monitoring",
            "No immediate action required"
        ]
    
    return AnomalyResponse(
        is_anomaly=result['is_anomaly'],
        anomaly_score=result['anomaly_score'],
        risk_level=result['risk_level'],
        recommendations=recommendations
    )

@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": True}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
                

Step 4: Dockerize and Deploy

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
                

# docker-compose.yml
version: '3.8'

services:
  anomaly-detector:
    build: .
    ports:
      - "8000:8000"
    environment:
      - MODEL_PATH=/models/network_anomaly_model.pkl
    volumes:
      - ./models:/models
    restart: unless-stopped