Monitoring & Maintenance

4.6.1 Drift Detection: Data Drift & Concept Drift

Model performance degradation in production environments typically results from two fundamental types of drift that must be monitored continuously to maintain system reliability and fairness.

Statistical Detection Methods

Drift Detection Framework

from scipy import stats
import numpy as np
from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Tuple

class DriftSeverity(Enum):
    NONE = "none"
    LOW = "low"
    MODERATE = "moderate"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class DriftResult:
    feature: str
    metric: str
    value: float
    threshold: float
    severity: DriftSeverity
    p_value: float = None

class DriftDetector:
    """Enterprise drift detection with multiple statistical methods."""
    
    PSI_THRESHOLDS = {"low": 0.1, "moderate": 0.2, "high": 0.25}
    KS_THRESHOLDS = {"low": 0.05, "moderate": 0.1, "high": 0.15}
    JSD_THRESHOLDS = {"low": 0.05, "moderate": 0.1, "high": 0.15}
    
    def calculate_psi(self, reference: np.ndarray, 
                      current: np.ndarray, 
                      bins: int = 10) -> float:
        """Calculate Population Stability Index."""
        # Create bins from reference distribution
        _, bin_edges = np.histogram(reference, bins=bins)
        
        # Calculate proportions
        ref_counts, _ = np.histogram(reference, bins=bin_edges)
        cur_counts, _ = np.histogram(current, bins=bin_edges)
        
        # Avoid division by zero
        ref_pct = (ref_counts + 1) / (len(reference) + bins)
        cur_pct = (cur_counts + 1) / (len(current) + bins)
        
        psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
        return psi
    
    def calculate_ks(self, reference: np.ndarray, 
                     current: np.ndarray) -> Tuple[float, float]:
        """Calculate Kolmogorov-Smirnov statistic and p-value."""
        statistic, p_value = stats.ks_2samp(reference, current)
        return statistic, p_value
    
    def calculate_jsd(self, reference: np.ndarray, 
                      current: np.ndarray, 
                      bins: int = 50) -> float:
        """Calculate Jensen-Shannon Divergence."""
        # Create common bins
        all_data = np.concatenate([reference, current])
        _, bin_edges = np.histogram(all_data, bins=bins)
        
        # Calculate distributions
        ref_hist, _ = np.histogram(reference, bins=bin_edges, density=True)
        cur_hist, _ = np.histogram(current, bins=bin_edges, density=True)
        
        # Normalize and avoid zeros
        ref_dist = (ref_hist + 1e-10) / (ref_hist + 1e-10).sum()
        cur_dist = (cur_hist + 1e-10) / (cur_hist + 1e-10).sum()
        
        # Calculate JSD
        m = 0.5 * (ref_dist + cur_dist)
        jsd = 0.5 * (stats.entropy(ref_dist, m) + stats.entropy(cur_dist, m))
        return jsd
    
    def get_severity(self, value: float, 
                     thresholds: Dict[str, float]) -> DriftSeverity:
        """Determine drift severity based on thresholds."""
        if value >= thresholds["high"]:
            return DriftSeverity.CRITICAL
        elif value >= thresholds["moderate"]:
            return DriftSeverity.HIGH
        elif value >= thresholds["low"]:
            return DriftSeverity.MODERATE
        elif value > 0:
            return DriftSeverity.LOW
        return DriftSeverity.NONE
    
    def detect_drift(self, reference: np.ndarray, 
                     current: np.ndarray, 
                     feature_name: str) -> List[DriftResult]:
        """Run comprehensive drift detection."""
        results = []
        
        # PSI
        psi = self.calculate_psi(reference, current)
        results.append(DriftResult(
            feature=feature_name,
            metric="PSI",
            value=psi,
            threshold=self.PSI_THRESHOLDS["moderate"],
            severity=self.get_severity(psi, self.PSI_THRESHOLDS)
        ))
        
        # KS Test
        ks_stat, ks_p = self.calculate_ks(reference, current)
        results.append(DriftResult(
            feature=feature_name,
            metric="KS",
            value=ks_stat,
            threshold=self.KS_THRESHOLDS["moderate"],
            severity=self.get_severity(ks_stat, self.KS_THRESHOLDS),
            p_value=ks_p
        ))
        
        # JSD
        jsd = self.calculate_jsd(reference, current)
        results.append(DriftResult(
            feature=feature_name,
            metric="JSD",
            value=jsd,
            threshold=self.JSD_THRESHOLDS["moderate"],
            severity=self.get_severity(jsd, self.JSD_THRESHOLDS)
        ))
        
        return results
                

4.6.2 Continuous Bias Monitoring

Fairness is not a static property—it must be continuously monitored throughout deployment. Models that were fair at launch can develop discriminatory patterns as data distributions shift, user populations change, or feedback loops amplify historical biases.

Continuous Fairness Metrics Framework

Feedback Loop Detection

AI systems can create self-reinforcing bias through feedback loops where model outputs influence future training data:

Intersectional Monitoring

Single-axis fairness metrics may miss discrimination that emerges at the intersection of multiple protected characteristics:

import pandas as pd
import numpy as np
from itertools import combinations, product
from typing import List, Dict, Tuple

class IntersectionalBiasMonitor:
    """Monitor fairness across intersections of protected attributes."""
    
    def __init__(self, protected_attributes: List[str], 
                 min_group_size: int = 30):
        self.protected_attributes = protected_attributes
        self.min_group_size = min_group_size
    
    def generate_intersections(self, 
                               data: pd.DataFrame) -> Dict[str, pd.Series]:
        """Generate all intersectional groups."""
        intersections = {}
        
        # Single attributes
        for attr in self.protected_attributes:
            for value in data[attr].unique():
                key = f"{attr}={value}"
                intersections[key] = data[attr] == value
        
        # All pairwise intersections
        for attr1, attr2 in combinations(self.protected_attributes, 2):
            for v1, v2 in product(data[attr1].unique(), 
                                  data[attr2].unique()):
                key = f"{attr1}={v1} & {attr2}={v2}"
                mask = (data[attr1] == v1) & (data[attr2] == v2)
                if mask.sum() >= self.min_group_size:
                    intersections[key] = mask
        
        return intersections
    
    def calculate_group_metrics(self, 
                                y_true: np.ndarray, 
                                y_pred: np.ndarray,
                                mask: np.ndarray) -> Dict[str, float]:
        """Calculate fairness metrics for a group."""
        y_true_g = y_true[mask]
        y_pred_g = y_pred[mask]
        
        if len(y_true_g) == 0:
            return {}
        
        # Selection rate
        selection_rate = y_pred_g.mean()
        
        # True positive rate
        pos_mask = y_true_g == 1
        tpr = y_pred_g[pos_mask].mean() if pos_mask.sum() > 0 else np.nan
        
        # False positive rate
        neg_mask = y_true_g == 0
        fpr = y_pred_g[neg_mask].mean() if neg_mask.sum() > 0 else np.nan
        
        # Positive predictive value
        pred_pos_mask = y_pred_g == 1
        ppv = y_true_g[pred_pos_mask].mean() if pred_pos_mask.sum() > 0 else np.nan
        
        return {
            "n": len(y_true_g),
            "selection_rate": selection_rate,
            "tpr": tpr,
            "fpr": fpr,
            "ppv": ppv
        }
    
    def monitor(self, data: pd.DataFrame, 
                y_true: np.ndarray, 
                y_pred: np.ndarray) -> pd.DataFrame:
        """Run intersectional bias monitoring."""
        intersections = self.generate_intersections(data)
        
        results = []
        overall_sr = y_pred.mean()
        
        for group_name, mask in intersections.items():
            metrics = self.calculate_group_metrics(
                y_true, y_pred, mask.values
            )
            if metrics:
                metrics["group"] = group_name
                metrics["selection_rate_ratio"] = (
                    metrics["selection_rate"] / overall_sr
                )
                metrics["disparate_impact"] = (
                    "YES" if metrics["selection_rate_ratio"] < 0.8 
                    or metrics["selection_rate_ratio"] > 1.25 
                    else "NO"
                )
                results.append(metrics)
        
        return pd.DataFrame(results).sort_values(
            "selection_rate_ratio"
        )
                

Alert Thresholds & Escalation

4.6.3 Incident Response Plan for AI Failures

AI systems can fail in ways that differ fundamentally from traditional software failures. Organizations must prepare specific incident response procedures that address the unique characteristics of AI incidents, including emergent behaviors, silent failures, and cascading effects.

Incident Response Framework

Incident Severity Classification

EU AI Act Incident Reporting Requirements

Incident Response Playbooks

Post-Incident Documentation

4.6.4 Model Retirement & Decommissioning

Models have lifecycles and must eventually be retired. Proper decommissioning ensures continued compliance, data protection, and smooth transitions to successor systems.

Implementation Checklist

Key Deliverables