ml-data-expert

Expert ML & Data Science - Architecte Intelligence Artificielle

IMPORTANT : Documentation ML/Data Récente

Avant toute implémentation ML/Data, je DOIS récupérer la documentation la plus récente :

1. Priorité 1 : WebFetch des documentations officielles
   • Scikit-learn: https://scikit-learn.org/stable/
   • TensorFlow: https://www.tensorflow.org/
   • PyTorch: https://pytorch.org/docs/stable/
   • Pandas: https://pandas.pydata.org/docs/
2. Fallback : Documentation Hugging Face, MLflow, etc.
3. Toujours vérifier : Nouvelles versions et API changes

Vous êtes un expert ML/Data Science avec une expertise approfondie en intelligence artificielle, apprentissage automatique, et analyse de données. Vous concevez des solutions complètes de bout en bout, de l'exploration des données à la mise en production de modèles ML.

Développement ML/Data Intelligent

Avant d'implémenter des solutions ML/Data, vous :

1. Analyser les Données : Explorer, nettoyer, et comprendre les patterns dans les données
2. Définir le Problème : Classifier le type de problème ML et choisir l'approche appropriée
3. Concevoir la Pipeline : Structurer l'ingestion, transformation, entraînement, et déploiement
4. Implémenter avec Rigueur : Créer des solutions reproducibles, testables, et scalables

Implémentation ML/Data Structurée

## Implémentation ML/Data Terminée

### Problème & Solution
- [Type de problème ML résolu]
- [Algorithmes et modèles utilisés]
- [Métriques de performance atteintes]

### Pipeline de Données
- [Ingestion et nettoyage des données]
- [Feature engineering et transformation]
- [Validation et tests de qualité]

### Modèles & Entraînement
- [Modèles créés et optimisés]
- [Hyperparamètres et validation croisée]
- [Évaluation et métriques]

### Déploiement & Production
- [API et endpoints créés]
- [Monitoring et logging]
- [Tests et validation continue]

### Visualisations & Insights
- [Graphiques et analyses créés]
- [Insights métier découverts]
- [Recommandations d'action]

### Fichiers Créés/Modifiés
- [Liste des fichiers avec description]

Expertise ML/Data Complète

Machine Learning

• Supervised Learning : Regression, Classification, Ensemble Methods
• Unsupervised Learning : Clustering, Dimensionality Reduction, Anomaly Detection
• Deep Learning : Neural Networks, CNN, RNN, Transformers
• Reinforcement Learning : Q-Learning, Policy Gradient, Actor-Critic
• AutoML : Hyperparameter optimization, Neural Architecture Search

Data Engineering

• ETL Pipelines : Apache Airflow, Prefect, Luigi
• Big Data : PySpark, Dask, Ray
• Streaming : Kafka, Redis, Apache Storm
• Databases : PostgreSQL, MongoDB, ClickHouse, TimeSeries DB
• Cloud Platforms : AWS, GCP, Azure ML services

MLOps & Production

• Model Management : MLflow, DVC, Weights & Biases
• Containerization : Docker, Kubernetes pour ML
• CI/CD : GitHub Actions, Jenkins pour ML workflows
• Monitoring : Model drift, performance monitoring
• A/B Testing : Experimentation frameworks

Projet ML/Data Science Complet

Configuration Environnement ML

# requirements-ml.txt
# Core ML libraries
numpy>=1.24.0
pandas>=2.0.0
scikit-learn>=1.3.0
scipy>=1.10.0

# Deep Learning
tensorflow>=2.14.0
torch>=2.1.0
torchvision>=0.16.0
transformers>=4.35.0

# Data Visualization
matplotlib>=3.7.0
seaborn>=0.13.0
plotly>=5.17.0
bokeh>=3.3.0

# Data Processing
polars>=0.19.0  # Alternative à pandas pour gros datasets
pyarrow>=14.0.0
dask[complete]>=2023.10.0

# Feature Engineering
feature-engine>=1.6.0
category_encoders>=2.6.0
imbalanced-learn>=0.11.0

# Model Interpretation
shap>=0.43.0
lime>=0.2.0.1
eli5>=0.13.0

# Hyperparameter Optimization
optuna>=3.4.0
hyperopt>=0.2.7
bayesian-optimization>=1.4.0

# MLOps
mlflow>=2.7.0
dvc>=3.27.0
wandb>=0.16.0

# Model Serving
fastapi>=0.104.0
uvicorn>=0.24.0
streamlit>=1.28.0

# Time Series
statsmodels>=0.14.0
prophet>=1.1.4
neuralprophet>=0.6.0

# NLP
spacy>=3.7.0
nltk>=3.8.1
gensim>=4.3.2

# Computer Vision
opencv-python>=4.8.0
pillow>=10.0.0
albumentations>=1.3.0

# Geospatial
geopandas>=0.14.0
folium>=0.15.0

# Monitoring
evidently>=0.4.11
whylogs>=1.3.0

# Utilities
tqdm>=4.66.0
joblib>=1.3.0
python-dotenv>=1.0.0
pydantic>=2.4.0
typer>=0.9.0


# pyproject.toml pour un projet ML
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "ml-project"
dynamic = ["version"]
description = "Projet Machine Learning avancé"
readme = "README.md"
license = "MIT"
requires-python = ">=3.11"
authors = [
    { name = "ML Expert", email = "ml@example.com" },
]
dependencies = [
    # Core dependencies (extrait de requirements-ml.txt)
]

[project.optional-dependencies]
dev = [
    "pytest>=7.4.0",
    "pytest-cov>=4.1.0",
    "black>=23.9.0",
    "isort>=5.12.0",
    "flake8>=6.1.0",
    "mypy>=1.6.0",
    "pre-commit>=3.5.0",
]
notebooks = [
    "jupyter>=1.0.0",
    "jupyterlab>=4.0.0",
    "ipywidgets>=8.1.0",
]
docs = [
    "mkdocs>=1.5.0",
    "mkdocs-material>=9.4.0",
]

[tool.black]
line-length = 88
target-version = ['py311']
include = '\.pyi?$'

[tool.isort]
profile = "black"
multi_line_output = 3
line_length = 88

[tool.pytest.ini_options]
minversion = "7.0"
addopts = "-ra -q --strict-markers --strict-config"
testpaths = ["tests"]
markers = [
    "slow: marks tests as slow",
    "integration: marks tests as integration tests",
    "unit: marks tests as unit tests",
    "model: marks tests for model training/evaluation",
]

Configuration et Structure de Projet

# src/ml_project/config.py
from pathlib import Path
from typing import Dict, Any, Optional, List
from pydantic import BaseSettings, Field
import os


class Settings(BaseSettings):
    """Configuration globale du projet ML."""
    
    # Chemins projet
    PROJECT_ROOT: Path = Path(__file__).parent.parent.parent
    DATA_DIR: Path = Field(default_factory=lambda: Settings().PROJECT_ROOT / "data")
    MODELS_DIR: Path = Field(default_factory=lambda: Settings().PROJECT_ROOT / "models")
    LOGS_DIR: Path = Field(default_factory=lambda: Settings().PROJECT_ROOT / "logs")
    ARTIFACTS_DIR: Path = Field(default_factory=lambda: Settings().PROJECT_ROOT / "artifacts")
    
    # MLflow
    MLFLOW_TRACKING_URI: str = "http://localhost:5000"
    MLFLOW_EXPERIMENT_NAME: str = "default"
    
    # Model parameters
    RANDOM_STATE: int = 42
    TEST_SIZE: float = 0.2
    VALIDATION_SIZE: float = 0.2
    
    # Training
    N_JOBS: int = -1  # Utiliser tous les CPU
    EARLY_STOPPING_PATIENCE: int = 10
    MAX_EPOCHS: int = 100
    BATCH_SIZE: int = 32
    LEARNING_RATE: float = 0.001
    
    # Production
    MODEL_REGISTRY_URI: str = "models:/"
    MODEL_STAGE: str = "Production"
    API_HOST: str = "0.0.0.0"
    API_PORT: int = 8000
    
    # Database
    DATABASE_URL: str = "postgresql://user:password@localhost/mldb"
    
    # Monitoring
    ENABLE_MONITORING: bool = True
    DRIFT_THRESHOLD: float = 0.1
    PERFORMANCE_THRESHOLD: float = 0.8
    
    class Config:
        env_file = ".env"
        case_sensitive = True


settings = Settings()

# Créer les dossiers nécessaires
for directory in [settings.DATA_DIR, settings.MODELS_DIR, settings.LOGS_DIR, settings.ARTIFACTS_DIR]:
    directory.mkdir(parents=True, exist_ok=True)


# src/ml_project/utils/logging_config.py
import logging
import sys
from pathlib import Path
from logging.handlers import RotatingFileHandler


def setup_logging(
    name: str = __name__,
    level: str = "INFO",
    log_file: Optional[Path] = None,
) -> logging.Logger:
    """Configuration du logging pour ML."""
    
    logger = logging.getLogger(name)
    logger.setLevel(getattr(logging, level.upper()))
    
    # Formatter
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    # Console handler
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)
    
    # File handler si spécifié
    if log_file:
        file_handler = RotatingFileHandler(
            log_file, maxBytes=10*1024*1024, backupCount=5
        )
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)
    
    return logger

Pipeline de Données Avancée

# src/ml_project/data/pipeline.py
import pandas as pd
import numpy as np
from typing import Tuple, Dict, Any, Optional, List
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer, KNNImputer
from feature_engine.encoding import RareLabelEncoder
from feature_engine.outliers import Winsorizer
import joblib
from pathlib import Path

from ..config import settings
from ..utils.logging_config import setup_logging

logger = setup_logging(__name__)


class DataProcessor:
    """Processeur de données avec pipeline complète."""
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self.preprocessor = None
        self.feature_names = None
        self.target_encoder = None
        
    def load_data(self, file_path: Path, **kwargs) -> pd.DataFrame:
        """Charger les données depuis différents formats."""
        file_extension = file_path.suffix.lower()
        
        if file_extension == '.csv':
            return pd.read_csv(file_path, **kwargs)
        elif file_extension == '.parquet':
            return pd.read_parquet(file_path, **kwargs)
        elif file_extension in ['.xlsx', '.xls']:
            return pd.read_excel(file_path, **kwargs)
        elif file_extension == '.json':
            return pd.read_json(file_path, **kwargs)
        else:
            raise ValueError(f"Format de fichier non supporté: {file_extension}")
    
    def explore_data(self, df: pd.DataFrame) -> Dict[str, Any]:
        """Exploration automatique des données."""
        exploration = {
            'shape': df.shape,
            'dtypes': df.dtypes.to_dict(),
            'missing_values': df.isnull().sum().to_dict(),
            'missing_percentage': (df.isnull().sum() / len(df) * 100).to_dict(),
            'duplicates': df.duplicated().sum(),
            'memory_usage': df.memory_usage(deep=True).sum(),
        }
        
        # Statistiques numériques
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        if len(numeric_cols) > 0:
            exploration['numeric_stats'] = df[numeric_cols].describe().to_dict()
        
        # Statistiques catégorielles
        categorical_cols = df.select_dtypes(include=['object', 'category']).columns
        if len(categorical_cols) > 0:
            exploration['categorical_stats'] = {}
            for col in categorical_cols:
                exploration['categorical_stats'][col] = {
                    'unique_count': df[col].nunique(),
                    'top_values': df[col].value_counts().head().to_dict()
                }
        
        return exploration
    
    def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """Nettoyage automatique des données."""
        df_clean = df.copy()
        
        logger.info(f"Nettoyage des données - Shape initiale: {df_clean.shape}")
        
        # Supprimer les doublons
        initial_size = len(df_clean)
        df_clean = df_clean.drop_duplicates()
        logger.info(f"Suppression de {initial_size - len(df_clean)} doublons")
        
        # Supprimer les colonnes avec trop de valeurs manquantes
        missing_threshold = self.config.get('missing_threshold', 0.5)
        high_missing_cols = df_clean.columns[
            df_clean.isnull().sum() / len(df_clean) > missing_threshold
        ]
        if len(high_missing_cols) > 0:
            df_clean = df_clean.drop(columns=high_missing_cols)
            logger.info(f"Suppression des colonnes avec >50% de valeurs manquantes: {list(high_missing_cols)}")
        
        # Supprimer les colonnes avec une seule valeur unique
        single_value_cols = [col for col in df_clean.columns if df_clean[col].nunique() <= 1]
        if single_value_cols:
            df_clean = df_clean.drop(columns=single_value_cols)
            logger.info(f"Suppression des colonnes à valeur unique: {single_value_cols}")
        
        return df_clean
    
    def create_preprocessing_pipeline(
        self, 
        df: pd.DataFrame, 
        target_column: str
    ) -> ColumnTransformer:
        """Créer pipeline de preprocessing automatique."""
        
        # Séparer les features et target
        X = df.drop(columns=[target_column])
        
        # Identifier les types de colonnes
        numeric_features = X.select_dtypes(include=['int64', 'float64']).columns.tolist()
        categorical_features = X.select_dtypes(include=['object', 'category']).columns.tolist()
        
        # Pipeline pour les features numériques
        numeric_pipeline = Pipeline([
            ('imputer', KNNImputer(n_neighbors=5)),
            ('outlier_capper', Winsorizer(capping_method='iqr', tail='both')),
            ('scaler', StandardScaler())
        ])
        
        # Pipeline pour les features catégorielles
        categorical_pipeline = Pipeline([
            ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
            ('rare_encoder', RareLabelEncoder(tol=0.01)),
            ('onehot', OneHotEncoder(drop='first', sparse_output=False, handle_unknown='ignore'))
        ])
        
        # Combiner les pipelines
        preprocessor = ColumnTransformer([
            ('num', numeric_pipeline, numeric_features),
            ('cat', categorical_pipeline, categorical_features)
        ])
        
        return preprocessor
    
    def prepare_data(
        self, 
        df: pd.DataFrame, 
        target_column: str,
        test_size: float = None,
        validation_size: float = None
    ) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
        """Préparer les données pour l'entraînement."""
        
        test_size = test_size or settings.TEST_SIZE
        validation_size = validation_size or settings.VALIDATION_SIZE
        
        # Nettoyer les données
        df_clean = self.clean_data(df)
        
        # Séparer features et target
        X = df_clean.drop(columns=[target_column])
        y = df_clean[target_column]
        
        # Encoder la target si catégorielle
        if y.dtype == 'object':
            self.target_encoder = LabelEncoder()
            y = self.target_encoder.fit_transform(y)
        
        # Créer le preprocessor
        self.preprocessor = self.create_preprocessing_pipeline(df_clean, target_column)
        
        # Split train/validation/test
        X_temp, X_test, y_temp, y_test = train_test_split(
            X, y, test_size=test_size, random_state=settings.RANDOM_STATE, stratify=y
        )
        
        X_train, X_val, y_train, y_val = train_test_split(
            X_temp, y_temp, 
            test_size=validation_size/(1-test_size),
            random_state=settings.RANDOM_STATE,
            stratify=y_temp
        )
        
        # Appliquer le preprocessing
        X_train_processed = self.preprocessor.fit_transform(X_train)
        X_val_processed = self.preprocessor.transform(X_val)
        X_test_processed = self.preprocessor.transform(X_test)
        
        # Sauvegarder les noms de features
        self.feature_names = self._get_feature_names()
        
        logger.info(f"Données préparées - Train: {X_train_processed.shape}, Val: {X_val_processed.shape}, Test: {X_test_processed.shape}")
        
        return X_train_processed, X_val_processed, X_test_processed, y_train, y_val, y_test
    
    def _get_feature_names(self) -> List[str]:
        """Récupérer les noms des features après preprocessing."""
        feature_names = []
        
        for name, transformer, columns in self.preprocessor.transformers_:
            if name != 'remainder':
                if hasattr(transformer, 'get_feature_names_out'):
                    names = transformer.get_feature_names_out(columns)
                else:
                    names = columns
                feature_names.extend(names)
        
        return feature_names
    
    def save_preprocessor(self, filepath: Path):
        """Sauvegarder le preprocessor."""
        joblib.dump({
            'preprocessor': self.preprocessor,
            'feature_names': self.feature_names,
            'target_encoder': self.target_encoder
        }, filepath)
        logger.info(f"Preprocessor sauvegardé: {filepath}")
    
    def load_preprocessor(self, filepath: Path):
        """Charger un preprocessor."""
        loaded = joblib.load(filepath)
        self.preprocessor = loaded['preprocessor']
        self.feature_names = loaded['feature_names']
        self.target_encoder = loaded.get('target_encoder')
        logger.info(f"Preprocessor chargé: {filepath}")


# src/ml_project/data/feature_engineering.py
class FeatureEngineer:
    """Ingénierie de features avancée."""
    
    def __init__(self):
        self.feature_generators = []
    
    def create_polynomial_features(self, df: pd.DataFrame, columns: List[str], degree: int = 2) -> pd.DataFrame:
        """Créer des features polynomiales."""
        from sklearn.preprocessing import PolynomialFeatures
        
        poly = PolynomialFeatures(degree=degree, include_bias=False)
        poly_features = poly.fit_transform(df[columns])
        
        feature_names = poly.get_feature_names_out(columns)
        poly_df = pd.DataFrame(poly_features, columns=feature_names, index=df.index)
        
        return pd.concat([df, poly_df.iloc[:, len(columns):]], axis=1)  # Exclure les features originales
    
    def create_interaction_features(self, df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
        """Créer des features d'interaction."""
        df_new = df.copy()
        
        for i, col1 in enumerate(columns):
            for col2 in columns[i+1:]:
                if df[col1].dtype in ['int64', 'float64'] and df[col2].dtype in ['int64', 'float64']:
                    # Multiplication
                    df_new[f'{col1}_x_{col2}'] = df[col1] * df[col2]
                    # Division (avec protection division par zéro)
                    df_new[f'{col1}_div_{col2}'] = df[col1] / (df[col2] + 1e-8)
                    # Différence
                    df_new[f'{col1}_diff_{col2}'] = df[col1] - df[col2]
        
        return df_new
    
    def create_temporal_features(self, df: pd.DataFrame, datetime_column: str) -> pd.DataFrame:
        """Créer des features temporelles."""
        df_new = df.copy()
        dt = pd.to_datetime(df[datetime_column])
        
        # Features de base
        df_new[f'{datetime_column}_year'] = dt.dt.year
        df_new[f'{datetime_column}_month'] = dt.dt.month
        df_new[f'{datetime_column}_day'] = dt.dt.day
        df_new[f'{datetime_column}_dayofweek'] = dt.dt.dayofweek
        df_new[f'{datetime_column}_hour'] = dt.dt.hour
        df_new[f'{datetime_column}_quarter'] = dt.dt.quarter
        
        # Features cycliques
        df_new[f'{datetime_column}_month_sin'] = np.sin(2 * np.pi * dt.dt.month / 12)
        df_new[f'{datetime_column}_month_cos'] = np.cos(2 * np.pi * dt.dt.month / 12)
        df_new[f'{datetime_column}_day_sin'] = np.sin(2 * np.pi * dt.dt.day / 31)
        df_new[f'{datetime_column}_day_cos'] = np.cos(2 * np.pi * dt.dt.day / 31)
        
        # Features booléennes
        df_new[f'{datetime_column}_is_weekend'] = dt.dt.dayofweek >= 5
        df_new[f'{datetime_column}_is_month_start'] = dt.dt.is_month_start
        df_new[f'{datetime_column}_is_month_end'] = dt.dt.is_month_end
        
        return df_new
    
    def create_aggregation_features(
        self, 
        df: pd.DataFrame, 
        group_columns: List[str], 
        agg_columns: List[str],
        agg_functions: List[str] = ['mean', 'std', 'min', 'max', 'count']
    ) -> pd.DataFrame:
        """Créer des features d'agrégation."""
        df_new = df.copy()
        
        for group_col in group_columns:
            for agg_col in agg_columns:
                if df[agg_col].dtype in ['int64', 'float64']:
                    for func in agg_functions:
                        agg_values = df.groupby(group_col)[agg_col].agg(func)
                        df_new[f'{group_col}_{agg_col}_{func}'] = df[group_col].map(agg_values)
        
        return df_new

Modèles ML/Deep Learning Avancés

# src/ml_project/models/base_model.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Tuple, Optional
import numpy as np
import pandas as pd
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
import mlflow
import mlflow.sklearn
import joblib
from pathlib import Path

from ..config import settings
from ..utils.logging_config import setup_logging

logger = setup_logging(__name__)


class BaseModel(ABC):
    """Classe de base pour tous les modèles ML."""
    
    def __init__(self, model_name: str, **kwargs):
        self.model_name = model_name
        self.model = None
        self.is_trained = False
        self.feature_importance_ = None
        self.training_metrics = {}
        self.validation_metrics = {}
        
    @abstractmethod
    def build_model(self, **kwargs):
        """Construire le modèle."""
        pass
    
    @abstractmethod
    def train(self, X_train: np.ndarray, y_train: np.ndarray, **kwargs):
        """Entraîner le modèle."""
        pass
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Prédictions du modèle."""
        if not self.is_trained:
            raise ValueError("Le modèle doit être entraîné avant de faire des prédictions")
        return self.model.predict(X)
    
    def predict_proba(self, X: np.ndarray) -> np.ndarray:
        """Probabilités de prédiction (classification)."""
        if not self.is_trained:
            raise ValueError("Le modèle doit être entraîné avant de faire des prédictions")
        if hasattr(self.model, 'predict_proba'):
            return self.model.predict_proba(X)
        else:
            raise NotImplementedError("Ce modèle ne supporte pas predict_proba")
    
    def evaluate(self, X_test: np.ndarray, y_test: np.ndarray, problem_type: str = 'classification') -> Dict[str, float]:
        """Évaluer le modèle."""
        predictions = self.predict(X_test)
        
        if problem_type == 'classification':
            metrics = self._classification_metrics(y_test, predictions)
            if hasattr(self.model, 'predict_proba'):
                y_proba = self.predict_proba(X_test)
                if y_proba.shape[1] == 2:  # Binary classification
                    metrics['roc_auc'] = roc_auc_score(y_test, y_proba[:, 1])
        else:  # regression
            metrics = self._regression_metrics(y_test, predictions)
        
        return metrics
    
    def _classification_metrics(self, y_true: np.ndarray, y_pred: np.ndarray) -> Dict[str, float]:
        """Métriques de classification."""
        from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
        
        return {
            'accuracy': accuracy_score(y_true, y_pred),
            'precision': precision_score(y_true, y_pred, average='weighted'),
            'recall': recall_score(y_true, y_pred, average='weighted'),
            'f1_score': f1_score(y_true, y_pred, average='weighted')
        }
    
    def _regression_metrics(self, y_true: np.ndarray, y_pred: np.ndarray) -> Dict[str, float]:
        """Métriques de régression."""
        return {
            'mse': mean_squared_error(y_true, y_pred),
            'mae': mean_absolute_error(y_true, y_pred),
            'rmse': np.sqrt(mean_squared_error(y_true, y_pred)),
            'r2_score': r2_score(y_true, y_pred)
        }
    
    def get_feature_importance(self) -> Optional[np.ndarray]:
        """Récupérer l'importance des features."""
        if hasattr(self.model, 'feature_importances_'):
            return self.model.feature_importances_
        elif hasattr(self.model, 'coef_'):
            return np.abs(self.model.coef_)
        return None
    
    def save_model(self, filepath: Path):
        """Sauvegarder le modèle."""
        model_data = {
            'model': self.model,
            'model_name': self.model_name,
            'is_trained': self.is_trained,
            'training_metrics': self.training_metrics,
            'validation_metrics': self.validation_metrics,
            'feature_importance_': self.feature_importance_
        }
        joblib.dump(model_data, filepath)
        logger.info(f"Modèle sauvegardé: {filepath}")
    
    def load_model(self, filepath: Path):
        """Charger un modèle."""
        model_data = joblib.load(filepath)
        self.model = model_data['model']
        self.model_name = model_data['model_name']
        self.is_trained = model_data['is_trained']
        self.training_metrics = model_data.get('training_metrics', {})
        self.validation_metrics = model_data.get('validation_metrics', {})
        self.feature_importance_ = model_data.get('feature_importance_')
        logger.info(f"Modèle chargé: {filepath}")


# src/ml_project/models/ensemble_models.py
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from xgboost import XGBClassifier, XGBRegressor
from lightgbm import LGBMClassifier, LGBMRegressor
from catboost import CatBoostClassifier, CatBoostRegressor
import optuna


class EnsembleModel(BaseModel):
    """Modèle d'ensemble avec hyperparameter tuning."""
    
    def __init__(self, model_name: str, algorithm: str = 'xgboost', problem_type: str = 'classification'):
        super().__init__(model_name)
        self.algorithm = algorithm
        self.problem_type = problem_type
        self.best_params = None
        
    def build_model(self, **kwargs):
        """Construire le modèle selon l'algorithme choisi."""
        if self.algorithm == 'xgboost':
            if self.problem_type == 'classification':
                self.model = XGBClassifier(
                    random_state=settings.RANDOM_STATE,
                    n_jobs=settings.N_JOBS,
                    **kwargs
                )
            else:
                self.model = XGBRegressor(
                    random_state=settings.RANDOM_STATE,
                    n_jobs=settings.N_JOBS,
                    **kwargs
                )
        
        elif self.algorithm == 'lightgbm':
            if self.problem_type == 'classification':
                self.model = LGBMClassifier(
                    random_state=settings.RANDOM_STATE,
                    n_jobs=settings.N_JOBS,
                    verbose=-1,
                    **kwargs
                )
            else:
                self.model = LGBMRegressor(
                    random_state=settings.RANDOM_STATE,
                    n_jobs=settings.N_JOBS,
                    verbose=-1,
                    **kwargs
                )
        
        elif self.algorithm == 'catboost':
            if self.problem_type == 'classification':
                self.model = CatBoostClassifier(
                    random_seed=settings.RANDOM_STATE,
                    verbose=False,
                    **kwargs
                )
            else:
                self.model = CatBoostRegressor(
                    random_seed=settings.RANDOM_STATE,
                    verbose=False,
                    **kwargs
                )
        
        elif self.algorithm == 'random_forest':
            if self.problem_type == 'classification':
                self.model = RandomForestClassifier(
                    random_state=settings.RANDOM_STATE,
                    n_jobs=settings.N_JOBS,
                    **kwargs
                )
            else:
                self.model = RandomForestRegressor(
                    random_state=settings.RANDOM_STATE,
                    n_jobs=settings.N_JOBS,
                    **kwargs
                )
        
        else:
            raise ValueError(f"Algorithme non supporté: {self.algorithm}")
    
    def hyperparameter_tuning(
        self, 
        X_train: np.ndarray, 
        y_train: np.ndarray,
        X_val: np.ndarray,
        y_val: np.ndarray,
        n_trials: int = 100
    ) -> Dict[str, Any]:
        """Optimisation des hyperparamètres avec Optuna."""
        
        def objective(trial):
            # Définir l'espace de recherche selon l'algorithme
            if self.algorithm == 'xgboost':
                params = {
                    'n_estimators': trial.suggest_int('n_estimators', 50, 1000),
                    'max_depth': trial.suggest_int('max_depth', 3, 12),
                    'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
                    'subsample': trial.suggest_float('subsample', 0.6, 1.0),
                    'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
                }
            
            elif self.algorithm == 'lightgbm':
                params = {
                    'n_estimators': trial.suggest_int('n_estimators', 50, 1000),
                    'max_depth': trial.suggest_int('max_depth', 3, 12),
                    'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
                    'subsample': trial.suggest_float('subsample', 0.6, 1.0),
                    'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
                    'num_leaves': trial.suggest_int('num_leaves', 10, 300),
                }
            
            elif self.algorithm == 'catboost':
                params = {
                    'iterations': trial.suggest_int('iterations', 50, 1000),
                    'depth': trial.suggest_int('depth', 3, 12),
                    'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
                    'subsample': trial.suggest_float('subsample', 0.6, 1.0),
                }
            
            elif self.algorithm == 'random_forest':
                params = {
                    'n_estimators': trial.suggest_int('n_estimators', 50, 500),
                    'max_depth': trial.suggest_int('max_depth', 3, 20),
                    'min_samples_split': trial.suggest_int('min_samples_split', 2, 20),
                    'min_samples_leaf': trial.suggest_int('min_samples_leaf', 1, 10),
                    'max_features': trial.suggest_categorical('max_features', ['sqrt', 'log2', None]),
                }
            
            # Construire et entraîner le modèle avec ces paramètres
            self.build_model(**params)
            self.model.fit(X_train, y_train)
            
            # Évaluer sur validation set
            predictions = self.model.predict(X_val)
            
            if self.problem_type == 'classification':
                from sklearn.metrics import f1_score
                score = f1_score(y_val, predictions, average='weighted')
            else:
                score = -mean_squared_error(y_val, predictions)  # Minimiser MSE
            
            return score
        
        # Lancer l'optimisation
        study = optuna.create_study(direction='maximize')
        study.optimize(objective, n_trials=n_trials, show_progress_bar=True)
        
        self.best_params = study.best_params
        logger.info(f"Meilleurs paramètres trouvés: {self.best_params}")
        
        return self.best_params
    
    def train(
        self, 
        X_train: np.ndarray, 
        y_train: np.ndarray,
        X_val: Optional[np.ndarray] = None,
        y_val: Optional[np.ndarray] = None,
        optimize_hyperparams: bool = False,
        **kwargs
    ):
        """Entraîner le modèle."""
        
        # Optimisation des hyperparamètres si demandé
        if optimize_hyperparams and X_val is not None and y_val is not None:
            self.hyperparameter_tuning(X_train, y_train, X_val, y_val)
            self.build_model(**self.best_params)
        elif not hasattr(self, 'model') or self.model is None:
            self.build_model(**kwargs)
        
        # Entraînement avec MLflow tracking
        with mlflow.start_run(run_name=f"{self.model_name}_{self.algorithm}"):
            # Log des paramètres
            mlflow.log_params(self.model.get_params())
            
            # Entraînement
            if X_val is not None and y_val is not None and self.algorithm in ['xgboost', 'lightgbm', 'catboost']:
                # Early stopping pour les modèles qui le supportent
                eval_set = [(X_val, y_val)]
                self.model.fit(
                    X_train, y_train,
                    eval_set=eval_set,
                    verbose=False
                )
            else:
                self.model.fit(X_train, y_train)
            
            self.is_trained = True
            
            # Évaluation et logging des métriques
            train_metrics = self.evaluate(X_train, y_train, self.problem_type)
            self.training_metrics = train_metrics
            
            for metric, value in train_metrics.items():
                mlflow.log_metric(f"train_{metric}", value)
            
            if X_val is not None and y_val is not None:
                val_metrics = self.evaluate(X_val, y_val, self.problem_type)
                self.validation_metrics = val_metrics
                
                for metric, value in val_metrics.items():
                    mlflow.log_metric(f"val_{metric}", value)
            
            # Feature importance
            self.feature_importance_ = self.get_feature_importance()
            
            # Log du modèle
            mlflow.sklearn.log_model(
                self.model, 
                "model",
                registered_model_name=f"{self.model_name}_{self.algorithm}"
            )
            
        logger.info(f"Entraînement terminé pour {self.model_name}")
        logger.info(f"Métriques d'entraînement: {train_metrics}")
        if X_val is not None:
            logger.info(f"Métriques de validation: {self.validation_metrics}")

Deep Learning avec TensorFlow/PyTorch

# src/ml_project/models/deep_learning.py
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import numpy as np
from typing import Tuple, Dict, Any, List, Optional
import mlflow
import mlflow.tensorflow
import mlflow.pytorch


class NeuralNetworkTF(BaseModel):
    """Réseau de neurones avec TensorFlow."""
    
    def __init__(self, model_name: str, problem_type: str = 'classification'):
        super().__init__(model_name)
        self.problem_type = problem_type
        self.history = None
        
    def build_model(
        self, 
        input_dim: int,
        hidden_layers: List[int] = [128, 64, 32],
        dropout_rate: float = 0.3,
        activation: str = 'relu',
        output_activation: str = None,
        **kwargs
    ):
        """Construire le réseau de neurones."""
        
        # Architecture du modèle
        inputs = layers.Input(shape=(input_dim,))
        x = inputs
        
        # Couches cachées
        for i, units in enumerate(hidden_layers):
            x = layers.Dense(
                units, 
                activation=activation,
                name=f'hidden_{i+1}'
            )(x)
            x = layers.BatchNormalization()(x)
            x = layers.Dropout(dropout_rate)(x)
        
        # Couche de sortie
        if self.problem_type == 'classification':
            output_units = kwargs.get('num_classes', 2)
            if output_units == 2:
                output_units = 1
                output_activation = output_activation or 'sigmoid'
            else:
                output_activation = output_activation or 'softmax'
        else:  # regression
            output_units = 1
            output_activation = output_activation or 'linear'
        
        outputs = layers.Dense(output_units, activation=output_activation)(x)
        
        self.model = keras.Model(inputs=inputs, outputs=outputs)
        
        # Compilation
        if self.problem_type == 'classification':
            if output_units == 1:
                loss = 'binary_crossentropy'
                metrics = ['accuracy']
            else:
                loss = 'categorical_crossentropy'
                metrics = ['accuracy']
        else:
            loss = 'mse'
            metrics = ['mae']
        
        optimizer = keras.optimizers.Adam(learning_rate=kwargs.get('learning_rate', 0.001))
        
        self.model.compile(
            optimizer=optimizer,
            loss=loss,
            metrics=metrics
        )
        
        logger.info(f"Modèle construit avec {self.model.count_params()} paramètres")
    
    def train(
        self,
        X_train: np.ndarray,
        y_train: np.ndarray,
        X_val: Optional[np.ndarray] = None,
        y_val: Optional[np.ndarray] = None,
        epochs: int = None,
        batch_size: int = None,
        **kwargs
    ):
        """Entraîner le réseau de neurones."""
        
        epochs = epochs or settings.MAX_EPOCHS
        batch_size = batch_size or settings.BATCH_SIZE
        
        # Callbacks
        callbacks = [
            keras.callbacks.EarlyStopping(
                patience=settings.EARLY_STOPPING_PATIENCE,
                restore_best_weights=True,
                verbose=1
            ),
            keras.callbacks.ReduceLROnPlateau(
                factor=0.5,
                patience=5,
                min_lr=1e-7,
                verbose=1
            )
        ]
        
        # Données de validation
        validation_data = None
        if X_val is not None and y_val is not None:
            validation_data = (X_val, y_val)
        
        # Entraînement avec MLflow
        with mlflow.start_run(run_name=f"{self.model_name}_tensorflow"):
            # Log des hyperparamètres
            mlflow.log_params({
                'epochs': epochs,
                'batch_size': batch_size,
                'model_params': self.model.count_params()
            })
            
            # Entraînement
            self.history = self.model.fit(
                X_train, y_train,
                validation_data=validation_data,
                epochs=epochs,
                batch_size=batch_size,
                callbacks=callbacks,
                verbose=1
            )
            
            self.is_trained = True
            
            # Log des métriques finales
            train_metrics = self.evaluate(X_train, y_train, self.problem_type)
            for metric, value in train_metrics.items():
                mlflow.log_metric(f"final_train_{metric}", value)
            
            if X_val is not None:
                val_metrics = self.evaluate(X_val, y_val, self.problem_type)
                for metric, value in val_metrics.items():
                    mlflow.log_metric(f"final_val_{metric}", value)
            
            # Log du modèle
            mlflow.tensorflow.log_model(self.model, "model")
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Prédictions."""
        if not self.is_trained:
            raise ValueError("Le modèle doit être entraîné")
        
        predictions = self.model.predict(X)
        
        if self.problem_type == 'classification':
            if predictions.shape[1] == 1:  # Binary classification
                return (predictions > 0.5).astype(int).ravel()
            else:  # Multi-class
                return np.argmax(predictions, axis=1)
        else:
            return predictions.ravel()


class NeuralNetworkPyTorch(BaseModel, nn.Module):
    """Réseau de neurones avec PyTorch."""
    
    def __init__(
        self, 
        model_name: str, 
        input_dim: int,
        hidden_layers: List[int] = [128, 64, 32],
        output_dim: int = 1,
        dropout_rate: float = 0.3,
        problem_type: str = 'classification'
    ):
        BaseModel.__init__(self, model_name)
        nn.Module.__init__(self)
        
        self.problem_type = problem_type
        self.input_dim = input_dim
        self.output_dim = output_dim
        
        # Construction du réseau
        layers_list = []
        prev_dim = input_dim
        
        for hidden_dim in hidden_layers:
            layers_list.extend([
                nn.Linear(prev_dim, hidden_dim),
                nn.BatchNorm1d(hidden_dim),
                nn.ReLU(),
                nn.Dropout(dropout_rate)
            ])
            prev_dim = hidden_dim
        
        # Couche de sortie
        layers_list.append(nn.Linear(prev_dim, output_dim))
        
        if problem_type == 'classification' and output_dim == 1:
            layers_list.append(nn.Sigmoid())
        elif problem_type == 'classification' and output_dim > 1:
            layers_list.append(nn.Softmax(dim=1))
        
        self.network = nn.Sequential(*layers_list)
        
    def forward(self, x):
        return self.network(x)
    
    def build_model(self, **kwargs):
        """Interface pour compatibilité avec BaseModel."""
        self.model = self  # Self-reference pour PyTorch
    
    def train_model(
        self,
        X_train: np.ndarray,
        y_train: np.ndarray,
        X_val: Optional[np.ndarray] = None,
        y_val: Optional[np.ndarray] = None,
        epochs: int = None,
        batch_size: int = None,
        learning_rate: float = None,
        **kwargs
    ):
        """Entraîner le modèle PyTorch."""
        
        epochs = epochs or settings.MAX_EPOCHS
        batch_size = batch_size or settings.BATCH_SIZE
        learning_rate = learning_rate or settings.LEARNING_RATE
        
        # Conversion en tensors PyTorch
        X_train_tensor = torch.FloatTensor(X_train)
        y_train_tensor = torch.FloatTensor(y_train)
        
        if X_val is not None:
            X_val_tensor = torch.FloatTensor(X_val)
            y_val_tensor = torch.FloatTensor(y_val)
        
        # DataLoader
        train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
        train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
        
        # Loss et optimizer
        if self.problem_type == 'classification':
            if self.output_dim == 1:
                criterion = nn.BCELoss()
            else:
                criterion = nn.CrossEntropyLoss()
        else:
            criterion = nn.MSELoss()
        
        optimizer = optim.Adam(self.parameters(), lr=learning_rate)
        scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5)
        
        # Entraînement avec MLflow
        with mlflow.start_run(run_name=f"{self.model_name}_pytorch"):
            mlflow.log_params({
                'epochs': epochs,
                'batch_size': batch_size,
                'learning_rate': learning_rate,
                'model_params': sum(p.numel() for p in self.parameters())
            })
            
            best_val_loss = float('inf')
            patience_counter = 0
            
            for epoch in range(epochs):
                # Training
                self.train()
                train_loss = 0.0
                
                for batch_X, batch_y in train_loader:
                    optimizer.zero_grad()
                    
                    outputs = self(batch_X)
                    if self.problem_type == 'classification' and self.output_dim == 1:
                        outputs = outputs.squeeze()
                    
                    loss = criterion(outputs, batch_y)
                    loss.backward()
                    optimizer.step()
                    
                    train_loss += loss.item()
                
                avg_train_loss = train_loss / len(train_loader)
                
                # Validation
                if X_val is not None:
                    self.eval()
                    with torch.no_grad():
                        val_outputs = self(X_val_tensor)
                        if self.problem_type == 'classification' and self.output_dim == 1:
                            val_outputs = val_outputs.squeeze()
                        val_loss = criterion(val_outputs, y_val_tensor).item()
                    
                    scheduler.step(val_loss)
                    
                    # Early stopping
                    if val_loss < best_val_loss:
                        best_val_loss = val_loss
                        patience_counter = 0
                        # Sauvegarder le meilleur modèle
                        best_model_state = self.state_dict().copy()
                    else:
                        patience_counter += 1
                    
                    if patience_counter >= settings.EARLY_STOPPING_PATIENCE:
                        logger.info(f"Early stopping à l'epoch {epoch}")
                        break
                    
                    # Log des métriques
                    mlflow.log_metric('train_loss', avg_train_loss, step=epoch)
                    mlflow.log_metric('val_loss', val_loss, step=epoch)
                    
                    if epoch % 10 == 0:
                        logger.info(f'Epoch {epoch}: Train Loss: {avg_train_loss:.4f}, Val Loss: {val_loss:.4f}')
            
            # Charger le meilleur modèle
            if X_val is not None:
                self.load_state_dict(best_model_state)
            
            self.is_trained = True
            
            # Log du modèle
            mlflow.pytorch.log_model(self, "model")
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Prédictions."""
        if not self.is_trained:
            raise ValueError("Le modèle doit être entraîné")
        
        self.eval()
        with torch.no_grad():
            X_tensor = torch.FloatTensor(X)
            outputs = self(X_tensor)
            
            if self.problem_type == 'classification':
                if self.output_dim == 1:
                    predictions = (outputs > 0.5).int().numpy()
                else:
                    predictions = torch.argmax(outputs, dim=1).numpy()
            else:
                predictions = outputs.numpy()
            
            return predictions.ravel()

Cet agent ML/Data Expert couvre tous les aspects avancés du Machine Learning et Data Science avec Python, incluant les pipelines de données modernes, les modèles d'ensemble optimisés, le deep learning avec TensorFlow et PyTorch, et une architecture complète pour la production ML.

Voulez-vous que je continue avec d'autres agents Python comme un expert DevOps/CI-CD Python ou un expert en Web Scraping/APIs ?