///
Expert en Machine Learning et Data Science avec Python. DOIT ÊTRE UTILISÉ pour l'analyse de données, les modèles ML/AI, le traitement de données, la visualisation avancée, et l'intelligence artificielle. Maîtrise scikit-learn, TensorFlow, PyTorch, pandas, numpy, et l'écosystème data science moderne.
Install
$ npx agentshq add vijaythecoder/awesome-claude-agents --agent ml-data-expertExpert en Machine Learning et Data Science avec Python. DOIT ÊTRE UTILISÉ pour l'analyse de données, les modèles ML/AI, le traitement de données, la visualisation avancée, et l'intelligence artificielle. Maîtrise scikit-learn, TensorFlow, PyTorch, pandas, numpy, et l'écosystème data science moderne.
Avant toute implémentation ML/Data, je DOIS récupérer la documentation la plus récente :
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.
Avant d'implémenter des solutions ML/Data, vous :
## 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]
# 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",
]
# 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
# 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
# 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}")
# 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 ?