Enterprise MLOps & Deployment

// Enterprise MLOps Pipeline Implementation
import kubeflow
import mlflow
import kubernetes
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import yaml

@dataclass
class MLOpsConfig:
    cluster_name: str
    namespace: str
    registry_url: str
    monitoring_enabled: bool = True
    auto_scaling: bool = True
    security_enabled: bool = True

class EnterpriseMLOpsPlatform:
    def __init__(self, config: MLOpsConfig):
        self.config = config

        # Initialize Kubernetes client
        self.k8s_client = kubernetes.client.ApiClient()
        self.apps_v1 = kubernetes.client.AppsV1Api()
        self.core_v1 = kubernetes.client.CoreV1Api()

        # MLflow client voor model management
        self.mlflow_client = mlflow.tracking.MlflowClient()

        # Kubeflow pipelines client
        self.kfp_client = kubeflow.pipelines.Client()

        # Model registry en deployment tracker
        self.model_registry = ModelRegistry()
        self.deployment_tracker = DeploymentTracker()

    def create_training_pipeline(self, pipeline_config: Dict) -> str:
        """Create en deploy een Kubeflow training pipeline"""

        @kubeflow.dsl.pipeline(
            name=pipeline_config['name'],
            description=pipeline_config['description']
        )
        def ml_training_pipeline(
            data_path: str,
            model_type: str,
            hyperparameters: Dict
        ):
            # Data validation step
            data_validation = self.create_data_validation_op(data_path)

            # Data preprocessing
            preprocessing = self.create_preprocessing_op(
                data_validation.outputs['validated_data']
            )

            # Feature engineering
            feature_engineering = self.create_feature_engineering_op(
                preprocessing.outputs['processed_data']
            )

            # Model training met distributed computing
            training = self.create_distributed_training_op(
                data=feature_engineering.outputs['features'],
                model_type=model_type,
                hyperparameters=hyperparameters,
                gpu_count=pipeline_config.get('gpu_count', 4)
            )

            # Model validation
            validation = self.create_model_validation_op(
                training.outputs['model'],
                feature_engineering.outputs['test_data']
            )

            # Model registration
            registration = self.create_model_registration_op(
                training.outputs['model'],
                validation.outputs['metrics']
            )

            return registration.outputs['model_version']

        # Compile en upload pipeline
        pipeline_package = kubeflow.dsl.compiler.Compiler().compile(
            ml_training_pipeline,
            f"{pipeline_config['name']}.yaml"
        )

        # Create pipeline run
        experiment = self.kfp_client.create_experiment(
            name=pipeline_config['experiment_name']
        )

        run = self.kfp_client.run_pipeline(
            experiment_id=experiment.id,
            job_name=f"{pipeline_config['name']}-{int(time.time())}",
            pipeline_package_path=pipeline_package,
            params=pipeline_config.get('parameters', {})
        )

        return run.id

    def deploy_model(self, model_version: str, deployment_config: Dict) -> Dict:
        """Deploy model naar Kubernetes met auto-scaling en monitoring"""

        # Get model artifacts from registry
        model_info = self.model_registry.get_model(model_version)

        # Create deployment manifest
        deployment_manifest = self.create_deployment_manifest(
            model_info, deployment_config
        )

        # Apply deployment
        deployment = self.apps_v1.create_namespaced_deployment(
            namespace=self.config.namespace,
            body=deployment_manifest
        )

        # Create service
        service_manifest = self.create_service_manifest(deployment_config)
        service = self.core_v1.create_namespaced_service(
            namespace=self.config.namespace,
            body=service_manifest
        )

        # Setup horizontal pod autoscaler
        if self.config.auto_scaling:
            hpa_manifest = self.create_hpa_manifest(deployment_config)
            self.create_hpa(hpa_manifest)

        # Setup monitoring
        if self.config.monitoring_enabled:
            self.setup_model_monitoring(model_version, deployment_config)

        # Register deployment
        deployment_info = {
            'model_version': model_version,
            'deployment_name': deployment.metadata.name,
            'service_name': service.metadata.name,
            'namespace': self.config.namespace,
            'endpoints': self.get_service_endpoints(service),
            'status': 'deployed',
            'deployment_time': datetime.utcnow().isoformat()
        }

        self.deployment_tracker.register_deployment(deployment_info)

        return deployment_info

    def create_deployment_manifest(self, model_info: Dict, config: Dict) -> Dict:
        """Create Kubernetes deployment manifest voor model serving"""

        return {
            'apiVersion': 'apps/v1',
            'kind': 'Deployment',
            'metadata': {
                'name': f"model-{model_info['name']}-{model_info['version']}",
                'labels': {
                    'app': 'ml-model',
                    'model': model_info['name'],
                    'version': model_info['version']
                }
            },
            'spec': {
                'replicas': config.get('replicas', 3),
                'selector': {
                    'matchLabels': {
                        'app': 'ml-model',
                        'model': model_info['name']
                    }
                },
                'template': {
                    'metadata': {
                        'labels': {
                            'app': 'ml-model',
                            'model': model_info['name'],
                            'version': model_info['version']
                        }
                    },
                    'spec': {
                        'containers': [{
                            'name': 'model-server',
                            'image': f"{self.config.registry_url}/model-server:{model_info['version']}",
                            'ports': [{
                                'containerPort': 8080,
                                'name': 'http'
                            }],
                            'env': [
                                {'name': 'MODEL_NAME', 'value': model_info['name']},
                                {'name': 'MODEL_VERSION', 'value': model_info['version']},
                                {'name': 'MODEL_PATH', 'value': model_info['artifact_path']}
                            ],
                            'resources': {
                                'requests': {
                                    'cpu': config.get('cpu_request', '500m'),
                                    'memory': config.get('memory_request', '1Gi')
                                },
                                'limits': {
                                    'cpu': config.get('cpu_limit', '2'),
                                    'memory': config.get('memory_limit', '4Gi'),
                                    'nvidia.com/gpu': config.get('gpu_limit', 0)
                                }
                            },
                            'livenessProbe': {
                                'httpGet': {
                                    'path': '/health',
                                    'port': 8080
                                },
                                'initialDelaySeconds': 30,
                                'periodSeconds': 10
                            },
                            'readinessProbe': {
                                'httpGet': {
                                    'path': '/ready',
                                    'port': 8080
                                },
                                'initialDelaySeconds': 5,
                                'periodSeconds': 5
                            }
                        }]
                    }
                }
            }
        }

# A/B Testing Framework voor model deployment
class ABTestingFramework:
    def __init__(self, mlops_platform: EnterpriseMLOpsPlatform):
        self.mlops = mlops_platform
        self.test_tracker = ABTestTracker()

    def create_ab_test(self, test_config: Dict) -> str:
        """Setup A/B test tussen model versies"""

        # Deploy control model (baseline)
        control_deployment = self.mlops.deploy_model(
            test_config['control_model_version'],
            {
                **test_config['deployment_config'],
                'replicas': test_config['control_traffic_percentage'] // 10
            }
        )

        # Deploy treatment model (challenger)
        treatment_deployment = self.mlops.deploy_model(
            test_config['treatment_model_version'],
            {
                **test_config['deployment_config'],
                'replicas': test_config['treatment_traffic_percentage'] // 10
            }
        )

        # Setup traffic splitting met Istio
        virtual_service = self.create_traffic_splitting_config(
            control_deployment, treatment_deployment, test_config
        )

        # Register A/B test
        ab_test = {
            'test_id': f"ab-test-{int(time.time())}",
            'control_model': test_config['control_model_version'],
            'treatment_model': test_config['treatment_model_version'],
            'traffic_split': {
                'control': test_config['control_traffic_percentage'],
                'treatment': test_config['treatment_traffic_percentage']
            },
            'success_metrics': test_config['success_metrics'],
            'duration_days': test_config['duration_days'],
            'start_time': datetime.utcnow().isoformat(),
            'status': 'running'
        }

        self.test_tracker.register_test(ab_test)

        # Setup monitoring voor A/B test metrics
        self.setup_ab_test_monitoring(ab_test)

        return ab_test['test_id']

    def analyze_ab_test_results(self, test_id: str) -> Dict:
        """Analyze A/B test results en statistical significance"""

        test_info = self.test_tracker.get_test(test_id)

        # Collect metrics for both variants
        control_metrics = self.collect_variant_metrics(
            test_info['control_model'], test_info['start_time']
        )
        treatment_metrics = self.collect_variant_metrics(
            test_info['treatment_model'], test_info['start_time']
        )

        # Statistical analysis
        results = {}
        for metric in test_info['success_metrics']:
            control_values = control_metrics[metric]
            treatment_values = treatment_metrics[metric]

            # T-test voor statistical significance
            from scipy import stats
            t_stat, p_value = stats.ttest_ind(control_values, treatment_values)

            # Effect size (Cohen's d)
            pooled_std = np.sqrt(((len(control_values) - 1) * np.var(control_values) +
                                 (len(treatment_values) - 1) * np.var(treatment_values)) /
                                (len(control_values) + len(treatment_values) - 2))
            cohens_d = (np.mean(treatment_values) - np.mean(control_values)) / pooled_std

            results[metric] = {
                'control_mean': np.mean(control_values),
                'treatment_mean': np.mean(treatment_values),
                'lift': (np.mean(treatment_values) - np.mean(control_values)) / np.mean(control_values),
                'p_value': p_value,
                'significant': p_value < 0.05,
                'effect_size': cohens_d,
                'confidence_interval': self.calculate_confidence_interval(
                    control_values, treatment_values
                )
            }

        # Overall recommendation
        recommendation = self.generate_recommendation(results, test_info)

        return {
            'test_id': test_id,
            'results': results,
            'recommendation': recommendation,
            'analysis_time': datetime.utcnow().isoformat()
        }

# Model Monitoring en Drift Detection
class ModelMonitoringSystem:
    def __init__(self):
        self.drift_detectors = {}
        self.performance_trackers = {}
        self.alert_manager = AlertManager()

    def setup_monitoring(self, model_deployment: Dict):
        """Setup comprehensive monitoring voor deployed model"""

        model_id = f"{model_deployment['model_version']}"

        # Data drift monitoring
        self.drift_detectors[model_id] = DataDriftDetector(
            reference_data=model_deployment['training_data_sample']
        )

        # Performance tracking
        self.performance_trackers[model_id] = PerformanceTracker(
            expected_metrics=model_deployment['baseline_metrics']
        )

        # Setup alerts
        self.setup_alerting_rules(model_id, model_deployment)

    def monitor_inference_data(self, model_id: str, inference_data: np.ndarray,
                              predictions: np.ndarray, ground_truth: Optional[np.ndarray] = None):
        """Monitor incoming inference data voor drift en performance"""

        monitoring_results = {
            'timestamp': datetime.utcnow().isoformat(),
            'model_id': model_id,
            'data_drift': {},
            'performance_metrics': {},
            'alerts': []
        }

        # Data drift detection
        if model_id in self.drift_detectors:
            drift_results = self.drift_detectors[model_id].detect_drift(inference_data)
            monitoring_results['data_drift'] = drift_results

            if drift_results['drift_detected']:
                alert = {
                    'type': 'data_drift',
                    'severity': 'warning',
                    'message': f"Data drift detected for model {model_id}",
                    'details': drift_results
                }
                monitoring_results['alerts'].append(alert)
                self.alert_manager.send_alert(alert)

        # Performance monitoring
        if ground_truth is not None and model_id in self.performance_trackers:
            performance = self.performance_trackers[model_id].track_performance(
                predictions, ground_truth
            )
            monitoring_results['performance_metrics'] = performance

            if performance['degradation_detected']:
                alert = {
                    'type': 'performance_degradation',
                    'severity': 'critical',
                    'message': f"Performance degradation detected for model {model_id}",
                    'details': performance
                }
                monitoring_results['alerts'].append(alert)
                self.alert_manager.send_alert(alert)

        return monitoring_results