Machine Learning with AI: Automated Model Generation for Rapid Development

# Core Python dependencies
pip install pandas numpy scikit-learn matplotlib seaborn

# AutoML libraries
pip install auto-sklearn h2o automl pycaret

# Cloud ML clients
pip install google-cloud-automl boto3 azure-ai-ml

# Model tracking and deployment
pip install mlflow wandb

# Development tools
pip install jupyter notebook plotly

# config.py - Centralized configuration management
import os
from pathlib import Path

class MLConfig:
    def __init__(self):
        self.project_root = Path(__file__).parent
        self.data_dir = self.project_root / "data"
        self.models_dir = self.project_root / "models"
        self.logs_dir = self.project_root / "logs"
        
        # Create directories if they don't exist
        for directory in [self.data_dir, self.models_dir, self.logs_dir]:
            directory.mkdir(exist_ok=True)
    
    # Cloud service credentials
    GOOGLE_CLOUD_PROJECT = os.getenv('GOOGLE_CLOUD_PROJECT')
    AWS_REGION = os.getenv('AWS_DEFAULT_REGION', 'us-east-1')
    AZURE_SUBSCRIPTION_ID = os.getenv('AZURE_SUBSCRIPTION_ID')
    
    # Model configuration
    MODEL_TIMEOUT = 3600  # 1 hour timeout for AutoML
    VALIDATION_SPLIT = 0.2
    RANDOM_STATE = 42

config = MLConfig()

# data_handler.py - Automated data preprocessing
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import train_test_split
from typing import Tuple, Dict, Any

class AutoDataPreprocessor:
    def __init__(self, target_column: str):
        self.target_column = target_column
        self.scalers = {}
        self.encoders = {}
        self.feature_info = {}
        
    def analyze_dataset(self, df: pd.DataFrame) -> Dict[str, Any]:
        """Automatically analyze dataset characteristics"""
        analysis = {
            'shape': df.shape,
            'missing_values': df.isnull().sum().to_dict(),
            'numeric_features': df.select_dtypes(include=[np.number]).columns.tolist(),
            'categorical_features': df.select_dtypes(include=['object']).columns.tolist(),
            'target_type': 'classification' if df[self.target_column].dtype == 'object' else 'regression'
        }
        
        # Detect high cardinality categorical features
        high_cardinality = []
        for col in analysis['categorical_features']:
            if df[col].nunique() > 50:
                high_cardinality.append(col)
        analysis['high_cardinality_features'] = high_cardinality
        
        self.feature_info = analysis
        return analysis
    
    def preprocess_features(self, df: pd.DataFrame, fit: bool = True) -> pd.DataFrame:
        """Automatically preprocess features based on data types"""
        df_processed = df.copy()
        
        # Handle missing values
        for col in df_processed.columns:
            if col == self.target_column:
                continue
                
            if df_processed[col].dtype in ['int64', 'float64']:
                # Numeric features: fill with median
                if fit:
                    median_val = df_processed[col].median()
                    self.scalers[f'{col}_median'] = median_val
                df_processed[col].fillna(self.scalers[f'{col}_median'], inplace=True)
                
                # Scale numeric features
                if fit:
                    scaler = StandardScaler()
                    df_processed[col] = scaler.fit_transform(df_processed[[col]])
                    self.scalers[col] = scaler
                else:
                    df_processed[col] = self.scalers[col].transform(df_processed[[col]])
            else:
                # Categorical features: fill with mode
                if fit:
                    mode_val = df_processed[col].mode()[0] if len(df_processed[col].mode()) > 0 else 'unknown'
                    self.encoders[f'{col}_mode'] = mode_val
                df_processed[col].fillna(self.encoders[f'{col}_mode'], inplace=True)
                
                # Encode categorical features
                if col not in self.feature_info.get('high_cardinality_features', []):
                    if fit:
                        encoder = LabelEncoder()
                        df_processed[col] = encoder.fit_transform(df_processed[col])
                        self.encoders[col] = encoder
                    else:
                        # Handle unseen categories
                        try:
                            df_processed[col] = self.encoders[col].transform(df_processed[col])
                        except ValueError:
                            # Replace unseen categories with most frequent
                            df_processed[col] = df_processed[col].map(
                                lambda x: x if x in self.encoders[col].classes_ else self.encoders[col].classes_[0]
                            )
                            df_processed[col] = self.encoders[col].transform(df_processed[col])
                else:
                    # Drop high cardinality features or use target encoding
                    df_processed.drop(col, axis=1, inplace=True)
        
        return df_processed
    
    def prepare_train_test_split(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
        """Prepare train-test split with preprocessing"""
        # Separate features and target
        X = df.drop(self.target_column, axis=1)
        y = df[self.target_column]
        
        # Encode target if classification
        if self.feature_info['target_type'] == 'classification':
            target_encoder = LabelEncoder()
            y = target_encoder.fit_transform(y)
            self.encoders['target'] = target_encoder
        
        # Split the data
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=config.VALIDATION_SPLIT, 
            random_state=config.RANDOM_STATE, 
            stratify=y if self.feature_info['target_type'] == 'classification' else None
        )
        
        # Preprocess features
        X_train_processed = self.preprocess_features(X_train, fit=True)
        X_test_processed = self.preprocess_features(X_test, fit=False)
        
        return X_train_processed, X_test_processed, y_train, y_test

# Example usage
preprocessor = AutoDataPreprocessor(target_column='target')

# automl_models.py - Multi-provider AutoML implementation
from abc import ABC, abstractmethod
import mlflow
import mlflow.sklearn
from sklearn.metrics import accuracy_score, mean_squared_error, classification_report
import joblib

class AutoMLProvider(ABC):
    """Abstract base class for AutoML providers"""
    
    def __init__(self, experiment_name: str):
        self.experiment_name = experiment_name
        self.model = None
        self.metrics = {}
        
        # Initialize MLflow tracking
        mlflow.set_experiment(experiment_name)
    
    @abstractmethod
    def train(self, X_train, y_train, **kwargs):
        pass
    
    @abstractmethod
    def predict(self, X_test):
        pass
    
    def evaluate(self, X_test, y_test, problem_type: str):
        """Evaluate model performance"""
        predictions = self.predict(X_test)
        
        if problem_type == 'classification':
            accuracy = accuracy_score(y_test, predictions)
            self.metrics['accuracy'] = accuracy
            self.metrics['classification_report'] = classification_report(y_test, predictions)
        else:
            mse = mean_squared_error(y_test, predictions)
            rmse = np.sqrt(mse)
            self.metrics['mse'] = mse
            self.metrics['rmse'] = rmse
        
        return self.metrics
    
    def save_model(self, model_path: str):
        """Save model with MLflow tracking"""
        with mlflow.start_run():
            # Log metrics
            for metric_name, metric_value in self.metrics.items():
                if isinstance(metric_value, (int, float)):
                    mlflow.log_metric(metric_name, metric_value)
            
            # Log model
            mlflow.sklearn.log_model(self.model, "model")
            joblib.dump(self.model, model_path)
            mlflow.log_artifact(model_path)

class AutoSklearnProvider(AutoMLProvider):
    """Auto-sklearn implementation"""
    
    def train(self, X_train, y_train, time_budget: int = 300, **kwargs):
        """Train model using auto-sklearn"""
        try:
            import autosklearn.classification
            import autosklearn.regression
        except ImportError:
            raise ImportError("auto-sklearn not installed. Run: pip install auto-sklearn")
        
        problem_type = kwargs.get('problem_type', 'classification')
        
        if problem_type == 'classification':
            self.model = autosklearn.classification.AutoSklearnClassifier(
                time_left_for_this_task=time_budget,
                per_run_time_limit=30,
                seed=config.RANDOM_STATE
            )
        else:
            self.model = autosklearn.regression.AutoSklearnRegressor(
                time_left_for_this_task=time_budget,
                per_run_time_limit=30,
                seed=config.RANDOM_STATE
            )
        
        self.model.fit(X_train, y_train)
        return self.model
    
    def predict(self, X_test):
        return self.model.predict(X_test)
    
    def get_model_info(self):
        """Get information about the best model found"""
        return {
            'best_models': str(self.model.show_models()),
            'statistics': str(self.model.sprint_statistics())
        }

class H2OAutoMLProvider(AutoMLProvider):
    """H2O AutoML implementation"""
    
    def __init__(self, experiment_name: str):
        super().__init__(experiment_name)
        self._initialize_h2o()
    
    def _initialize_h2o(self):
        """Initialize H2O cluster"""
        try:
            import h2o
            from h2o.automl import H2OAutoML
            h2o.init()
            self.h2o = h2o
            self.H2OAutoML = H2OAutoML
        except ImportError:
            raise ImportError("h2o not installed. Run: pip install h2o")
    
    def train(self, X_train, y_train, time_budget: int = 300, **kwargs):
        """Train model using H2O AutoML"""
        # Convert to H2O frame
        train_data = pd.concat([X_train, pd.Series(y_train, name='target')], axis=1)
        h2o_train = self.h2o.H2OFrame(train_data)
        
        # Define target and features
        target = 'target'
        features = h2o_train.columns[:-1]
        
        # Configure AutoML
        self.model = self.H2OAutoML(
            max_runtime_secs=time_budget,
            seed=config.RANDOM_STATE,
            project_name=self.experiment_name
        )
        
        # Train the model
        self.model.train(x=features, y=target, training_frame=h2o_train)
        return self.model
    
    def predict(self, X_test):
        """Make predictions using H2O model"""
        h2o_test = self.h2o.H2OFrame(X_test)
        predictions = self.model.leader.predict(h2o_test)
        return predictions.as_data_frame().values.flatten()
    
    def get_leaderboard(self):
        """Get AutoML leaderboard"""
        return self.model.leaderboard.as_data_frame()

class PyCaretProvider(AutoMLProvider):
    """PyCaret AutoML implementation"""
    
    def train(self, X_train, y_train, time_budget: int = 300, **kwargs):
        """Train model using PyCaret"""
        try:
            import pycaret.classification as pycl
            import pycaret.regression as pycr
        except ImportError:
            raise ImportError("pycaret not installed. Run: pip install pycaret")
        
        problem_type = kwargs.get('problem_type', 'classification')
        
        # Prepare data
        train_data = pd.concat([X_train, pd.Series(y_train, name='target')], axis=1)
        
        if problem_type == 'classification':
            # Setup classification environment
            clf = pycl.setup(
                data=train_data,
                target='target',
                session_id=config.RANDOM_STATE,
                train_size=0.8,
                silent=True
            )
            
            # Compare models and select best
            best_models = pycl.compare_models(
                include=['rf', 'et', 'xgboost', 'lightgbm', 'catboost'],
                sort='Accuracy',
                n_select=3
            )
            
            # Create ensemble
            self.model = pycl.ensemble_model(best_models)
            self.model = pycl.finalize_model(self.model)
            
        else:
            # Setup regression environment
            reg = pycr.setup(
                data=train_data,
                target='target',
                session_id=config.RANDOM_STATE,
                train_size=0.8,
                silent=True
            )
            
            # Compare models and select best
            best_models = pycr.compare_models(
                include=['rf', 'et', 'xgboost', 'lightgbm', 'catboost'],
                sort='RMSE',
                n_select=3
            )
            
            # Create ensemble
            self.model = pycr.ensemble_model(best_models)
            self.model = pycr.finalize_model(self.model)
        
        return self.model
    
    def predict(self, X_test):
        return self.model.predict(X_test)

# ai_code_generator.py - Generate ML code using AI
import openai
import json
from typing import Dict, List, Any

class MLCodeGenerator:
    """Generate ML code using AI language models"""
    
    def __init__(self, api_key: str = None):
        self.client = openai.OpenAI(api_key=api_key) if api_key else None
        self.templates = self._load_templates()
    
    def _load_templates(self) -> Dict[str, str]:
        """Load code templates for common ML tasks"""
        return {
            'data_analysis': '''
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

def analyze_dataset(df, target_column):
    analysis = {{}}
    
    # Basic statistics
    analysis['shape'] = df.shape
    analysis['columns'] = df.columns.tolist()
    analysis['dtypes'] = df.dtypes.to_dict()
    analysis['missing_values'] = df.isnull().sum().to_dict()
    
    # Target analysis
    if df[target_column].dtype == 'object':
        analysis['target_distribution'] = df[target_column].value_counts().to_dict()
    else:
        analysis['target_stats'] = df[target_column].describe().to_dict()
    
    return analysis
''',
            'feature_engineering': '''
from sklearn.preprocessing import StandardScaler, LabelEncoder, OneHotEncoder
from sklearn.feature_selection import SelectKBest, f_classif

def engineer_features(X_train, X_test, feature_config):
    X_train_processed = X_train.copy()
    X_test_processed = X_test.copy()
    
    # Apply transformations based on config
    for column, transformation in feature_config.items():
        if transformation == 'standard_scale':
            scaler = StandardScaler()
            X_train_processed[column] = scaler.fit_transform(X_train_processed[[column]])
            X_test_processed[column] = scaler.transform(X_test_processed[[column]])
        elif transformation == 'label_encode':
            encoder = LabelEncoder()
            X_train_processed[column] = encoder.fit_transform(X_train_processed[column])
            X_test_processed[column] = encoder.transform(X_test_processed[column])
    
    return X_train_processed, X_test_processed
''',
            'model_training': '''
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC
from sklearn.model_selection import GridSearchCV
from sklearn.metrics import classification_report, accuracy_score

def train_multiple_models(X_train, y_train, X_test, y_test):
    models = {{
        'random_forest': RandomForestClassifier(random_state=42),
        'gradient_boosting': GradientBoostingClassifier(random_state=42),
        'logistic_regression': LogisticRegression(random_state=42),
        'svm': SVC(random_state=42)
    }}
    
    results = {{}}
    
    for name, model in models.items():
        model.fit(X_train, y_train)
        predictions = model.predict(X_test)
        accuracy = accuracy_score(y_test, predictions)
        
        results[name] = {{
            'model': model,
            'accuracy': accuracy,
            'predictions': predictions
        }}
    
    return results
'''
        }
    
    def generate_custom_code(self, task_description: str, data_info: Dict[str, Any]) -> str:
        """Generate custom ML code using AI"""
        if not self.client:
            return self._use_template_fallback(task_description, data_info)
        
        prompt = f"""
Generate Python code for the following machine learning task:

Task: {task_description}

Dataset Information:
- Shape: {data_info.get('shape', 'Unknown')}
- Features: {data_info.get('numeric_features', [])} (numeric), {data_info.get('categorical_features', [])} (categorical)
- Target Type: {data_info.get('target_type', 'Unknown')}
- Missing Values: {data_info.get('missing_values', {})}

Requirements:
1. Use scikit-learn and pandas
2. Include proper error handling
3. Add comments explaining each step
4. Return a complete, runnable function
5. Include basic evaluation metrics

Generate clean, production-ready code:
"""
        
        try:
            response = self.client.chat.completions.create(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": "You are an expert ML engineer. Generate clean, well-documented Python code."},
                    {"role": "user", "content": prompt}
                ],
                max_tokens=2000,
                temperature=0.1
            )
            
            return response.choices[0].message.content
        except Exception as e:
            print(f"AI code generation failed: {e}")
            return self._use_template_fallback(task_description, data_info)
    
    def _use_template_fallback(self, task_description: str, data_info: Dict[str, Any]) -> str:
        """Fallback to templates when AI is unavailable"""
        task_lower = task_description.lower()
        
        if 'analysis' in task_lower or 'explore' in task_lower:
            return self.templates['data_analysis']
        elif 'feature' in task_lower or 'preprocess' in task_lower:
            return self.templates['feature_engineering']
        elif 'train' in task_lower or 'model' in task_lower:
            return self.templates['model_training']
        else:
            return "# Generated template not found. Please specify the task more clearly."
    
    def optimize_hyperparameters(self, model_type: str, search_space: Dict[str, List]) -> str:
        """Generate hyperparameter optimization code"""
        optimization_code = f"""
from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from sklearn.metrics import make_scorer
import numpy as np

def optimize_{model_type}_hyperparameters(X_train, y_train, cv_folds=5):
    # Define the model
    if '{model_type}' == 'random_forest':
        from sklearn.ensemble import RandomForestClassifier
        model = RandomForestClassifier(random_state=42)
    elif '{model_type}' == 'gradient_boosting':
        from sklearn.ensemble import GradientBoostingClassifier
        model = GradientBoostingClassifier(random_state=42)
    elif '{model_type}' == 'svm':
        from sklearn.svm import SVC
        model = SVC(random_state=42)
    else:
        raise ValueError(f"Unsupported model type: {model_type}")
    
    # Define parameter grid
    param_grid = {json.dumps(search_space, indent=4)}
    
    # Perform grid search
    grid_search = GridSearchCV(
        estimator=model,
        param_grid=param_grid,
        cv=cv_folds,
        scoring='accuracy',
        n_jobs=-1,
        verbose=1
    )
    
    # Fit the grid search
    grid_search.fit(X_train, y_train)
    
    return {{
        'best_params': grid_search.best_params_,
        'best_score': grid_search.best_score_,
        'best_model': grid_search.best_estimator_,
        'cv_results': grid_search.cv_results_
    }}
"""
        return optimization_code

# Example usage
code_generator = MLCodeGenerator()

# model_optimization.py - Advanced optimization strategies
class ModelOptimizer:
    """Advanced model optimization and selection"""
    
    def __init__(self, optimization_strategy: str = 'bayesian'):
        self.strategy = optimization_strategy
        self.optimization_history = []
    
    def optimize_ensemble(self, models: List, X_val, y_val) -> Dict[str, float]:
        """Optimize ensemble weights using validation data"""
        from scipy.optimize import minimize
        
        def ensemble_objective(weights):
            weights = weights / np.sum(weights)  # Normalize weights
            ensemble_pred = np.zeros(len(y_val))
            
            for i, model in enumerate(models):
                pred = model.predict(X_val)
                ensemble_pred += weights[i] * pred
            
            return mean_squared_error(y_val, ensemble_pred)
        
        # Initial equal weights
        initial_weights = np.ones(len(models)) / len(models)
        
        # Optimize weights
        result = minimize(
            ensemble_objective,
            initial_weights,
            method='SLSQP',
            bounds=[(0, 1) for _ in range(len(models))],
            constraints={'type': 'eq', 'fun': lambda w: np.sum(w) - 1}
        )
        
        return {
            'optimal_weights': result.x,
            'ensemble_score': result.fun,
            'optimization_success': result.success
        }
    
    def adaptive_time_budget(self, dataset_size: int, complexity_score: float) -> int:
        """Calculate optimal time budget based on dataset characteristics"""
        base_time = 300  # 5 minutes base
        
        # Adjust for dataset size
        size_factor = min(dataset_size / 10000, 5.0)  # Cap at 5x
        
        # Adjust for complexity (number of features, missing values, etc.)
        complexity_factor = min(complexity_score, 3.0)  # Cap at 3x
        
        optimal_time = int(base_time * size_factor * complexity_factor)
        return max(optimal_time, 180)  # Minimum 3 minutes

# test_automl_pipeline.py - Complete testing suite
import unittest
import pandas as pd
import numpy as np
from sklearn.datasets import make_classification, make_regression

class TestAutoMLPipeline(unittest.TestCase):
    
    def setUp(self):
        """Set up test datasets"""
        # Classification dataset
        X_clf, y_clf = make_classification(
            n_samples=1000, n_features=20, n_informative=10,
            n_redundant=5, n_classes=3, random_state=42
        )
        self.clf_data = pd.DataFrame(X_clf, columns=[f'feature_{i}' for i in range(20)])
        self.clf_data['target'] = y_clf
        
        # Regression dataset
        X_reg, y_reg = make_regression(
            n_samples=1000, n_features=15, noise=0.1,
            random_state=42
        )
        self.reg_data = pd.DataFrame(X_reg, columns=[f'feature_{i}' for i in range(15)])
        self.reg_data['target'] = y_reg
    
    def test_data_preprocessing(self):
        """Test automated data preprocessing"""
        preprocessor = AutoDataPreprocessor('target')
        
        # Test analysis
        analysis = preprocessor.analyze_dataset(self.clf_data)
        self.assertEqual(analysis['shape'], (1000, 21))
        self.assertEqual(analysis['target_type'], 'classification')
        
        # Test preprocessing
        X_train, X_test, y_train, y_test = preprocessor.prepare_train_test_split(self.clf_data)
        self.assertEqual(X_train.shape[0], 800)  # 80% for training
        self.assertEqual(X_test.shape[0], 200)   # 20% for testing
    
    def test_automl_providers(self):
        """Test different AutoML providers"""
        preprocessor = AutoDataPreprocessor('target')
        X_train, X_test, y_train, y_test = preprocessor.prepare_train_test_split(self.clf_data)
        
        # Test Auto-sklearn (if available)
        try:
            provider = AutoSklearnProvider("test_autosklearn")
            model = provider.train(X_train, y_train, time_budget=60, problem_type='classification')
            predictions = provider.predict(X_test)
            self.assertEqual(len(predictions), len(y_test))
        except ImportError:
            self.skipTest("Auto-sklearn not available")
        
        # Test PyCaret (if available)
        try:
            provider = PyCaretProvider("test_pycaret")
            model = provider.train(X_train, y_train, time_budget=60, problem_type='classification')
            predictions = provider.predict(X_test)
            self.assertEqual(len(predictions), len(y_test))
        except ImportError:
            self.skipTest("PyCaret not available")
    
    def test_code_generation(self):
        """Test AI code generation functionality"""
        code_generator = MLCodeGenerator()
        
        # Test template fallback
        data_info = {
            'shape': (1000, 20),
            'numeric_features': ['feature_1', 'feature_2'],
            'categorical_features': ['category_1'],
            'target_type': 'classification'
        }
        
        code = code_generator.generate_custom_code("data analysis", data_info)
        self.assertIn("import pandas as pd", code)
        self.assertIn("def analyze_dataset", code)
    
    def test_model_evaluation(self):
        """Test model evaluation metrics"""
        from sklearn.ensemble import RandomForestClassifier
        
        preprocessor = AutoDataPreprocessor('target')
        X_train, X_test, y_train, y_test = preprocessor.prepare_train_test_split(self.clf_data)
        
        # Create a simple model for testing
        model = RandomForestClassifier(n_estimators=10, random_state=42)
        model.fit(X_train, y_train)
        
        # Test evaluation
        provider = AutoSklearnProvider("test_evaluation")
        provider.model = model
        metrics = provider.evaluate(X_test, y_test, 'classification')
        
        self.assertIn('accuracy', metrics)
        self.assertIsInstance(metrics['accuracy'], float)
        self.assertGreaterEqual(metrics['accuracy'], 0.0)
        self.assertLessEqual(metrics['accuracy'], 1.0)

def run_integration_test():
    """Run complete integration test"""
    print("Running AutoML Integration Test...")
    
    # Generate test dataset
    X, y = make_classification(n_samples=500, n_features=10, random_state=42)
    test_data = pd.DataFrame(X, columns=[f'feature_{i}' for i in range(10)])
    test_data['target'] = y
    
    try:
        # Test complete pipeline
        preprocessor = AutoDataPreprocessor('target')
        analysis = preprocessor.analyze_dataset(test_data)
        print(f"Dataset analysis: {analysis['shape']} shape, {analysis['target_type']} problem")
        
        X_train, X_test, y_train, y_test = preprocessor.prepare_train_test_split(test_data)
        print(f"Data split: {X_train.shape[0]} train, {X_test.shape[0]} test samples")
        
        # Try each provider
        providers = [
            ("AutoSklearn", AutoSklearnProvider),
            ("PyCaret", PyCaretProvider),
            ("H2O", H2OAutoMLProvider)
        ]
        
        results = {}
        for name, provider_class in providers:
            try:
                provider = provider_class(f"test_{name.lower()}")
                model = provider.train(X_train, y_train, time_budget=60, problem_type='classification')
                metrics = provider.evaluate(X_test, y_test, 'classification')
                results[name] = metrics['accuracy']
                print(f"{name} accuracy: {metrics['accuracy']:.3f}")
            except ImportError as e:
                print(f"{name} not available: {e}")
            except Exception as e:
                print(f"{name} failed: {e}")
        
        if results:
            best_provider = max(results, key=results.get)
            print(f"Best performing provider: {best_provider} ({results[best_provider]:.3f})")
        
        print("Integration test completed successfully!")
        return True
        
    except Exception as e:
        print(f"Integration test failed: {e}")
        return False

if __name__ == "__main__":
    # Run unit tests
    unittest.main(argv=[''], exit=False, verbosity=2)
    
    # Run integration test
    run_integration_test()

# model_monitoring.py - Production monitoring setup
import mlflow
import pandas as pd
from typing import Dict, Any
import warnings

class ModelMonitor:
    """Monitor model performance in production"""
    
    def __init__(self, model_name: str, model_version: str):
        self.model_name = model_name
        self.model_version = model_version
        self.performance_threshold = 0.8
        self.drift_threshold = 0.1
    
    def log_prediction_batch(self, features: pd.DataFrame, predictions: np.ndarray, 
                           actual: np.ndarray = None) -> Dict[str, Any]:
        """Log batch predictions for monitoring"""
        with mlflow.start_run():
            # Log prediction statistics
            mlflow.log_metric("batch_size", len(predictions))
            mlflow.log_metric("prediction_mean", np.mean(predictions))
            mlflow.log_metric("prediction_std", np.std(predictions))
            
            # If actual values available, compute accuracy
            if actual is not None:
                accuracy = accuracy_score(actual, predictions)
                mlflow.log_metric("batch_accuracy", accuracy)
                
                if accuracy < self.performance_threshold:
                    warnings.warn(f"Model performance below threshold: {accuracy:.3f}")
            
            return {
                "batch_size": len(predictions),
                "timestamp": pd.Timestamp.now(),
                "model_version": self.model_version
            }
    
    def detect_data_drift(self, reference_data: pd.DataFrame, 
                         current_data: pd.DataFrame) -> Dict[str, float]:
        """Detect data drift using statistical tests"""
        from scipy.stats import ks_2samp
        
        drift_scores = {}
        
        for column in reference_data.columns:
            if reference_data[column].dtype in ['int64', 'float64']:
                # Kolmogorov-Smirnov test for numeric features
                statistic, p_value = ks_2samp(reference_data[column], current_data[column])
                drift_scores[column] = {
                    'ks_statistic': statistic,
                    'p_value': p_value,
                    'drift_detected': p_value < 0.05
                }
        
        return drift_scores

# Dockerfile for AutoML service
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]