Model Monitoring

┌─────────────────┐
│   Prediction    │
│    Service      │
└────────┬────────┘
         │
         ├──────────▶ Metrics (Prometheus)
         ├──────────▶ Logs (ELK/Loki)
         ├──────────▶ Traces (Jaeger)
         └──────────▶ Data Store
                          │
                          ▼
              ┌───────────────────────┐
              │   Monitoring Agent    │
              │  - Drift Detection    │
              │  - Performance Checks │
              │  - Data Quality       │
              └───────────┬───────────┘
                          │
                          ▼
              ┌───────────────────────┐
              │  Alerting System      │
              │  - Slack/PagerDuty    │
              │  - Email              │
              └───────────────────────┘

import logging
import json
from datetime import datetime
from typing import Dict, Any

class StructuredLogger:
    """Structured logging for ML predictions."""
    
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)
        
        # JSON formatter
        handler = logging.StreamHandler()
        handler.setFormatter(self.JSONFormatter())
        self.logger.addHandler(handler)
    
    class JSONFormatter(logging.Formatter):
        def format(self, record):
            log_data = {
                'timestamp': datetime.utcnow().isoformat(),
                'level': record.levelname,
                'message': record.getMessage(),
                'logger': record.name
            }
            
            # Add extra fields
            if hasattr(record, 'extra'):
                log_data.update(record.extra)
            
            return json.dumps(log_data)
    
    def log_prediction(self, request_id: str, features: Dict, 
                      prediction: float, metadata: Dict):
        """Log prediction with full context."""
        self.logger.info(
            'Prediction made',
            extra={
                'request_id': request_id,
                'features': features,
                'prediction': prediction,
                'model_version': metadata.get('model_version'),
                'processing_time_ms': metadata.get('processing_time'),
                'confidence': metadata.get('confidence')
            }
        )
    
    def log_error(self, request_id: str, error: Exception, context: Dict):
        """Log error with context."""
        self.logger.error(
            f'Prediction error: {str(error)}',
            extra={
                'request_id': request_id,
                'error_type': type(error).__name__,
                'context': context
            }
        )

# Example usage
logger = StructuredLogger('aqi_predictor')

logger.log_prediction(
    request_id='abc-123',
    features={'pm25': 35.5, 'temperature': 22.5},
    prediction=75.3,
    metadata={'model_version': 'v1.2', 'processing_time': 45.2}
)

from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import time

# Setup tracing
trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(
    agent_host_name='localhost',
    agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

tracer = trace.get_tracer(__name__)

def predict_with_tracing(features):
    """Make prediction with distributed tracing."""
    with tracer.start_as_current_span('predict') as span:
        span.set_attribute('model.version', 'v1.2')
        span.set_attribute('features.count', len(features))
        
        # Preprocess
        with tracer.start_as_current_span('preprocess'):
            processed = preprocess(features)
        
        # Inference
        with tracer.start_as_current_span('inference'):
            prediction = model.predict(processed)
        
        # Postprocess
        with tracer.start_as_current_span('postprocess'):
            result = postprocess(prediction)
        
        span.set_attribute('prediction.value', result)
        
        return result

{
  "dashboard": {
    "title": "AQI Predictor Monitoring",
    "panels": [
      {
        "title": "Prediction MAE",
        "targets": [{
          "expr": "aqi_prediction_mae"
        }],
        "alert": {
          "conditions": [{
            "evaluator": {"params": [15], "type": "gt"},
            "query": {"params": ["A", "5m", "now"]}
          }]
        }
      },
      {
        "title": "Request Rate",
        "targets": [{
          "expr": "rate(aqi_api_requests_total[5m])"
        }]
      },
      {
        "title": "P95 Latency",
        "targets": [{
          "expr": "histogram_quantile(0.95, aqi_api_request_duration_seconds)"
        }]
      },
      {
        "title": "Error Rate",
        "targets": [{
          "expr": "rate(prediction_errors_total[5m])"
        }]
      }
    ]
  }
}

class RetrainingTrigger:
    """Trigger model retraining based on monitoring signals."""
    
    def __init__(self, config: Dict):
        self.config = config
        self.drift_detector = OnlineDriftDetector()
        self.performance_window = []
    
    def should_retrain(self, metrics: Dict) -> Tuple[bool, str]:
        """Determine if retraining is needed."""
        reasons = []
        
        # Check performance degradation
        if metrics['mae'] > self.config['max_mae']:
            reasons.append(f"MAE {metrics['mae']:.2f} exceeds threshold")
        
        # Check drift
        if metrics.get('drift_detected'):
            reasons.append("Data drift detected")
        
        # Check staleness
        days_since_training = metrics.get('days_since_training', 0)
        if days_since_training > self.config['max_model_age_days']:
            reasons.append(f"Model is {days_since_training} days old")
        
        # Check error rate trend
        self.performance_window.append(metrics['mae'])
        if len(self.performance_window) > 7:
            self.performance_window.pop(0)
            trend = np.polyfit(range(7), self.performance_window, 1)[0]
            if trend > 0.5:  # Increasing error trend
                reasons.append("Performance degrading over time")
        
        should_retrain = len(reasons) > 0
        reason_str = "; ".join(reasons)
        
        return should_retrain, reason_str
    
    def trigger_retraining_pipeline(self, reason: str):
        """Trigger automated retraining pipeline."""
        print(f"Triggering retraining: {reason}")
        
        # Call training pipeline (e.g., Airflow, Kubeflow)
        # requests.post('http://airflow:8080/api/v1/dags/aqi_training/dagRuns')
        pass