🔍 AI Monitoring & Detection

1. AI Security Monitoring Architecture

A comprehensive monitoring system that tracks AI model behavior, detects anomalies, and identifies security threats in real-time.

Key Metrics to Monitor

• Prediction confidence: Unusually high/low confidence scores
• Input distribution: Changes in data patterns
• Model performance: Accuracy and latency metrics
• Resource utilization: CPU, memory, and GPU usage
• Error rates: Failed predictions and exceptions

2. Adversarial Attack Detection

Techniques to identify when AI models are under adversarial attack or receiving malicious inputs.

Statistical Detection Methods

import numpy as np
from scipy import stats

class AdversarialDetector:
    def __init__(self, baseline_stats=None):
        self.baseline_stats = baseline_stats or {}
        
    def detect_anomaly(self, input_data, model_output):
        """
        Detect potential adversarial inputs using statistical methods
        """
        # Feature-based detection
        feature_stats = self._compute_feature_stats(input_data)
        
        # Output-based detection
        output_stats = self._compute_output_stats(model_output)
        
        # Combined anomaly score
        anomaly_score = self._compute_anomaly_score(feature_stats, output_stats)
        
        return {
            'is_anomaly': anomaly_score > 0.8,
            'anomaly_score': anomaly_score,
            'feature_stats': feature_stats,
            'output_stats': output_stats
        }
    
    def _compute_feature_stats(self, input_data):
        """Compute statistical features of input data"""
        return {
            'mean': np.mean(input_data),
            'std': np.std(input_data),
            'skewness': stats.skew(input_data.flatten()),
            'kurtosis': stats.kurtosis(input_data.flatten()),
            'entropy': self._compute_entropy(input_data)
        }
    
    def _compute_entropy(self, data):
        """Compute entropy of data distribution"""
        hist, _ = np.histogram(data.flatten(), bins=50)
        prob = hist / np.sum(hist)
        prob = prob[prob > 0]  # Remove zeros
        return -np.sum(prob * np.log2(prob))

Ensemble Detection

class EnsembleDetector:
    def __init__(self):
        self.detectors = [
            StatisticalDetector(),
            GradientBasedDetector(),
            UncertaintyDetector(),
            ReconstructionDetector()
        ]
        
    def detect_adversarial(self, input_data, model_output):
        """
        Ensemble detection using multiple methods
        """
        scores = []
        for detector in self.detectors:
            score = detector.detect(input_data, model_output)
            scores.append(score)
        
        # Weighted ensemble decision
        weights = [0.3, 0.25, 0.25, 0.2]  # Learned weights
        final_score = sum(w * s for w, s in zip(weights, scores))
        
        return {
            'is_adversarial': final_score > 0.7,
            'confidence': final_score,
            'individual_scores': scores
        }

3. Model Drift Detection

Monitor for changes in data distribution and model performance over time.

Data Drift Detection

from scipy.stats import ks_2samp
import pandas as pd

class DataDriftDetector:
    def __init__(self, reference_data):
        self.reference_data = reference_data
        
    def detect_drift(self, current_data, threshold=0.05):
        """
        Detect data drift using statistical tests
        """
        drift_results = {}
        
        for column in self.reference_data.columns:
            ref_values = self.reference_data[column].values
            curr_values = current_data[column].values
            
            # Kolmogorov-Smirnov test
            ks_stat, ks_pvalue = ks_2samp(ref_values, curr_values)
            
            # Population Stability Index (PSI)
            psi = self._compute_psi(ref_values, curr_values)
            
            drift_results[column] = {
                'ks_statistic': ks_stat,
                'ks_pvalue': ks_pvalue,
                'psi': psi,
                'has_drift': ks_pvalue < threshold or psi > 0.2
            }
        
        return drift_results
    
    def _compute_psi(self, reference, current, bins=10):
        """Compute Population Stability Index"""
        # Create bins based on reference data
        ref_hist, bin_edges = np.histogram(reference, bins=bins)
        curr_hist, _ = np.histogram(current, bins=bin_edges)
        
        # Normalize to probabilities
        ref_prob = ref_hist / np.sum(ref_hist)
        curr_prob = curr_hist / np.sum(curr_hist)
        
        # Compute PSI
        psi = 0
        for i in range(len(ref_prob)):
            if ref_prob[i] > 0 and curr_prob[i] > 0:
                psi += (curr_prob[i] - ref_prob[i]) * np.log(curr_prob[i] / ref_prob[i])
        
        return psi

Concept Drift Detection

class ConceptDriftDetector:
    def __init__(self, window_size=1000):
        self.window_size = window_size
        self.performance_history = []
        
    def update_performance(self, accuracy, timestamp=None):
        """Update performance metrics"""
        self.performance_history.append({
            'accuracy': accuracy,
            'timestamp': timestamp or time.time()
        })
        
        # Keep only recent history
        if len(self.performance_history) > self.window_size:
            self.performance_history = self.performance_history[-self.window_size:]
    
    def detect_concept_drift(self, threshold=0.05):
        """Detect concept drift based on performance degradation"""
        if len(self.performance_history) < 100:
            return {'has_drift': False, 'confidence': 0.0}
        
        # Split into two windows
        mid_point = len(self.performance_history) // 2
        recent_performance = [p['accuracy'] for p in self.performance_history[mid_point:]]
        historical_performance = [p['accuracy'] for p in self.performance_history[:mid_point]]
        
        # Statistical test for performance difference
        from scipy.stats import ttest_ind
        t_stat, p_value = ttest_ind(historical_performance, recent_performance)
        
        # Calculate drift severity
        mean_diff = np.mean(recent_performance) - np.mean(historical_performance)
        
        return {
            'has_drift': p_value < threshold and mean_diff < -0.02,
            'p_value': p_value,
            'performance_degradation': abs(mean_diff),
            'confidence': 1 - p_value
        }