Comprehensive comparison for Observability technology in AI applications

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import ColumnDriftMetric, DatasetDriftMetric
from evidently.test_suite import TestSuite
from evidently.tests import TestColumnDrift, TestShareOfMissingValues
import logging

# Configure logging for production monitoring
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CreditScoringMonitor:
    """Production monitoring for credit scoring ML model using Evidently AI"""
    
    def __init__(self, reference_data: pd.DataFrame):
        self.reference_data = reference_data
        self.column_mapping = ColumnMapping(
            target='default',
            prediction='prediction',
            numerical_features=['income', 'loan_amount', 'credit_history_length', 'debt_to_income'],
            categorical_features=['employment_status', 'loan_purpose', 'home_ownership']
        )
    
    def monitor_production_data(self, current_data: pd.DataFrame) -> dict:
        """Monitor current production data against reference baseline"""
        try:
            # Validate input data
            if current_data.empty:
                raise ValueError("Current data is empty")
            
            # Create drift report
            drift_report = Report(metrics=[
                DataDriftPreset(),
                DataQualityPreset(),
                ColumnDriftMetric(column_name='income'),
                ColumnDriftMetric(column_name='loan_amount'),
                DatasetDriftMetric()
            ])
            
            drift_report.run(
                reference_data=self.reference_data,
                current_data=current_data,
                column_mapping=self.column_mapping
            )
            
            # Create test suite for automated alerts
            test_suite = TestSuite(tests=[
                TestColumnDrift(column_name='income', stattest='ks'),
                TestColumnDrift(column_name='loan_amount', stattest='ks'),
                TestShareOfMissingValues(column_name='income', lt=0.05),
                TestShareOfMissingValues(column_name='debt_to_income', lt=0.05)
            ])
            
            test_suite.run(
                reference_data=self.reference_data,
                current_data=current_data,
                column_mapping=self.column_mapping
            )
            
            # Extract results
            drift_results = drift_report.as_dict()
            test_results = test_suite.as_dict()
            
            # Check for critical issues
            dataset_drift = drift_results['metrics'][2]['result']['drift_detected']
            tests_passed = test_results['summary']['all_passed']
            
            if dataset_drift or not tests_passed:
                logger.warning(f"Data drift detected: {dataset_drift}, Tests passed: {tests_passed}")
                self._trigger_alert(drift_results, test_results)
            
            # Save reports for audit trail
            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
            drift_report.save_html(f'reports/drift_report_{timestamp}.html')
            
            return {
                'timestamp': timestamp,
                'dataset_drift_detected': dataset_drift,
                'tests_passed': tests_passed,
                'drift_score': drift_results['metrics'][2]['result'].get('drift_score', 0),
                'num_drifted_columns': drift_results['metrics'][2]['result'].get('number_of_drifted_columns', 0)
            }
            
        except Exception as e:
            logger.error(f"Monitoring failed: {str(e)}")
            raise
    
    def _trigger_alert(self, drift_results: dict, test_results: dict):
        """Send alerts when drift or test failures detected"""
        logger.critical("ALERT: Model monitoring detected critical issues")
        # Integration point for alerting systems (PagerDuty, Slack, etc.)

# Example usage in production API
if __name__ == '__main__':
    # Load reference data (training data statistics)
    reference_df = pd.DataFrame({
        'income': np.random.normal(50000, 15000, 1000),
        'loan_amount': np.random.normal(20000, 5000, 1000),
        'credit_history_length': np.random.randint(1, 30, 1000),
        'debt_to_income': np.random.uniform(0.1, 0.6, 1000),
        'employment_status': np.random.choice(['employed', 'self_employed', 'unemployed'], 1000),
        'loan_purpose': np.random.choice(['debt_consolidation', 'home', 'car'], 1000),
        'home_ownership': np.random.choice(['rent', 'own', 'mortgage'], 1000),
        'default': np.random.binomial(1, 0.15, 1000),
        'prediction': np.random.binomial(1, 0.15, 1000)
    })
    
    # Simulate current production data with drift
    current_df = pd.DataFrame({
        'income': np.random.normal(55000, 15000, 500),  # Income drift
        'loan_amount': np.random.normal(20000, 5000, 500),
        'credit_history_length': np.random.randint(1, 30, 500),
        'debt_to_income': np.random.uniform(0.1, 0.6, 500),
        'employment_status': np.random.choice(['employed', 'self_employed', 'unemployed'], 500),
        'loan_purpose': np.random.choice(['debt_consolidation', 'home', 'car'], 500),
        'home_ownership': np.random.choice(['rent', 'own', 'mortgage'], 500),
        'default': np.random.binomial(1, 0.15, 500),
        'prediction': np.random.binomial(1, 0.15, 500)
    })
    
    # Initialize monitor and run
    monitor = CreditScoringMonitor(reference_df)
    results = monitor.monitor_production_data(current_df)
    logger.info(f"Monitoring results: {results}")