""" Module d'interprétation des modèles de machine learning. Ce module contient des fonctions pour : - récupérer les importances natives des features d'un modèle tree-based - regrouper les importances par variable métier - calculer la permutation importance sur le jeu de test - analyser les erreurs de prédiction (résidus) en log et en valeurs réelles - sauvegarder tous les artefacts du modèle (CSV, métriques, graphiques) de manière organisée """ import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns import json from sklearn.inspection import permutation_importance from sklearn.metrics import mean_absolute_error, root_mean_squared_error, r2_score def get_feature_importance(pipeline, top_n=20, plot=True): """ Récupère les importances natives des features d'un modèle tree-based. args: pipeline : sklearn.pipeline.Pipeline, modèle entraîné avec un préprocesseur intégré top_n : int, nombre de features à afficher plot : bool, afficher le graphique ou pas returns: feature_importance_df : pd.DataFrame, importance de chaque feature """ preprocessor = pipeline.named_steps["preprocessor"] model = pipeline.named_steps["model"] feature_names = preprocessor.get_feature_names_out() importances = model.feature_importances_ feature_importance_df = pd.DataFrame({ "feature": feature_names, "importance": importances }).sort_values(by="importance", ascending=False) top_features = feature_importance_df.head(top_n) if plot: plt.figure(figsize=(10, 6)) sns.barplot( data=top_features, x="importance", y="feature" ) plt.title(f"Top {top_n} - Feature importance native") plt.xlabel("Importance") plt.ylabel("Variable") plt.tight_layout() plt.show() return feature_importance_df def simplify_feature_name(feature_name): """ Regroupe les variables transformées vers leur variable métier d'origine. Par exemple : - item_1, item_2, item_3 -> item - area_1, area_2, area_3 -> area args: feature_name : str, nom de la feature transformée returns: str, nom de la variable métier d'origine """ feature_name = feature_name.replace("num__", "").replace("cat__", "") if feature_name.startswith("item_"): return "item" if feature_name.startswith("area_"): return "area" return feature_name def get_grouped_feature_importance(feature_importance_df, top_n=20, plot=True): """ Regroupe les importances par variable métier. Par exemple : - item_1, item_2, item_3 -> item - area_1, area_2, area_3 -> area Calcule l'importance cumulée pour chaque variable métier. Trie par importance décroissante. Affiche un graphique des variables métier les plus importantes. args: feature_importance_df : pd.DataFrame, importance de chaque feature transformée top_n : int, nombre de variables métier à afficher plot : bool, afficher le graphique ou pas returns: grouped_importance_df : pd.DataFrame, importance cumulée par variable métier """ grouped_df = feature_importance_df.copy() grouped_df["feature_group"] = grouped_df["feature"].apply(simplify_feature_name) grouped_df = ( grouped_df .groupby("feature_group", as_index=False)["importance"] .sum() .sort_values(by="importance", ascending=False) ) top_grouped = grouped_df.head(top_n) if plot: plt.figure(figsize=(10, 6)) sns.barplot( data=top_grouped, x="importance", y="feature_group" ) plt.title(f"Top {top_n} - Variables métier les plus importantes") plt.xlabel("Importance cumulée") plt.ylabel("Variable métier") plt.tight_layout() plt.show() return grouped_df def get_permutation_importance( pipeline, X_test, y_test, n_repeats=10, random_state=42, top_n=20, scoring="neg_root_mean_squared_error", plot=True ): """ Calcule la permutation importance sur le jeu de test. Mesure la perte de performance quand une variable est mélangée. Affiche un graphique des variables les plus importantes par permutation. args: pipeline : sklearn.Pipeline, modèle entraîné X_test : pd.DataFrame, données de test y_test : pd.Series, valeurs cibles de test n_repeats : int, nombre de répétitions pour la permutation random_state : int, graine pour la reproductibilité top_n : int, nombre de variables à afficher scoring : str, métrique de performance plot : bool, afficher le graphique ou pas returns: permutation_df : pd.DataFrame, importance par permutation """ result = permutation_importance( pipeline, X_test, y_test, n_repeats=n_repeats, random_state=random_state, scoring=scoring, n_jobs=-1 ) permutation_df = pd.DataFrame({ "feature": X_test.columns, "importance_mean": result.importances_mean, "importance_std": result.importances_std }).sort_values(by="importance_mean", ascending=False) top_features = permutation_df.head(top_n).copy() if plot: top_features_plot = top_features.sort_values( by="importance_mean", ascending=True ) plt.figure(figsize=(10, 6)) plt.barh( y=top_features_plot["feature"], width=top_features_plot["importance_mean"], xerr=top_features_plot["importance_std"] ) plt.title(f"Top {top_n} - Permutation importance") plt.xlabel("Impact moyen sur la performance") plt.ylabel("Variable") plt.tight_layout() plt.show() return permutation_df def analyze_residuals(pipeline, X_test, y_test, plot=True): """ Analyse les erreurs de prédiction : - en log, cohérent avec l'entraînement - en valeurs réelles, utile pour l'interprétation métier Affiche des graphiques d'analyse des résidus. args: pipeline : sklearn.Pipeline, modèle entraîné X_test : pd.DataFrame, données de test y_test : pd.Series, valeurs cibles de test plot : bool, afficher les graphiques ou pas returns: residuals_df : pd.DataFrame, résidus et valeurs associées metrics : dict, métriques d'évaluation des résidus """ # Prédictions en log y_pred_log = pipeline.predict(X_test) y_test_log = pd.Series(y_test).reset_index(drop=True) y_pred_log = pd.Series(y_pred_log).reset_index(drop=True) residuals_log = y_test_log - y_pred_log # Conversion en valeurs réelles y_test_real = np.expm1(y_test_log) y_pred_real = np.expm1(y_pred_log) residuals_real = y_test_real - y_pred_real residuals_df = pd.DataFrame({ "y_true_log": y_test_log, "y_pred_log": y_pred_log, "residual_log": residuals_log, "abs_residual_log": np.abs(residuals_log), "y_true_real": y_test_real, "y_pred_real": y_pred_real, "residual_real": residuals_real, "abs_residual_real": np.abs(residuals_real), }) metrics = { "mae_log": mean_absolute_error(y_test_log, y_pred_log), "rmse_log": root_mean_squared_error(y_test_log, y_pred_log), "r2_log": r2_score(y_test_log, y_pred_log), "mae_real": mean_absolute_error(y_test_real, y_pred_real), "rmse_real": root_mean_squared_error(y_test_real, y_pred_real), "r2_real": r2_score(y_test_real, y_pred_real), } if plot: # Résidus en log plt.figure(figsize=(8, 5)) sns.scatterplot( data=residuals_df, x="y_pred_log", y="residual_log" ) plt.axhline(0, color="black", linestyle="--") plt.title("Analyse des résidus - échelle log") plt.xlabel("Valeurs prédites (log)") plt.ylabel("Erreur de prédiction (log)") plt.tight_layout() plt.show() plt.figure(figsize=(8, 5)) sns.histplot(residuals_df["residual_log"], kde=True) plt.title("Distribution des erreurs - échelle log") plt.xlabel("Erreur (log)") plt.tight_layout() plt.show() # Résidus en réel plt.figure(figsize=(8, 5)) sns.scatterplot( data=residuals_df, x="y_pred_real", y="residual_real" ) plt.axhline(0, color="black", linestyle="--") plt.title("Analyse des résidus - valeurs réelles") plt.xlabel("Valeurs prédites") plt.ylabel("Erreur de prédiction") plt.tight_layout() plt.show() plt.figure(figsize=(8, 5)) sns.histplot(residuals_df["residual_real"], kde=True) plt.title("Distribution des erreurs - valeurs réelles") plt.xlabel("Erreur") plt.tight_layout() plt.show() return residuals_df, metrics def save_model_artifacts( base_path, feature_importance_df, grouped_importance_df, permutation_df, residuals_df, residual_metrics, final_model_df=None, top_n=20 ): """ Sauvegarde tous les artefacts du modèle : - CSV - métriques - graphiques Les artefacts sont organisés dans des dossiers "tables" et "figures" sous le dossier racine. Les graphiques sont enregistrés au format PNG avec une résolution de 300 dpi. Les CSV contiennent les données d'importance des features, les résidus et la comparaison des modèles finaux. Les métriques sont sauvegardées dans un fichier JSON pour une utilisation ultérieure. Args: base_path (Path): dossier racine des artefacts feature_importance_df (pd.DataFrame): importance native des features grouped_importance_df (pd.DataFrame): importance regroupée par variable métier permutation_df (pd.DataFrame): importance par permutation residuals_df (pd.DataFrame): résidus et valeurs associées residual_metrics (dict): métriques d'évaluation des résidus final_model_df (pd.DataFrame, optional): comparaison des modèles finaux top_n (int, optional): nombre de features à afficher dans les graphiques (par défaut 20) """ tables_dir = base_path / "tables" figures_dir = base_path / "figures" tables_dir.mkdir(parents=True, exist_ok=True) figures_dir.mkdir(parents=True, exist_ok=True) # ===================== # CSV # ===================== feature_importance_df.to_csv( tables_dir / "feature_importance_native.csv", index=False ) grouped_importance_df.to_csv( tables_dir / "feature_importance_grouped.csv", index=False ) permutation_df.to_csv( tables_dir / "permutation_importance.csv", index=False ) residuals_df.to_csv( tables_dir / "residuals.csv", index=False ) if final_model_df is not None: final_model_df.drop(columns=["trained_pipeline"], errors="ignore").to_csv( tables_dir / "final_model_comparison.csv", index=False ) # Metrics with open(tables_dir / "metrics.json", "w", encoding="utf-8") as f: json.dump( {k: float(v) for k, v in residual_metrics.items()}, f, indent=4 ) # ===================== # GRAPHIQUES # ===================== # Feature importance top_features = feature_importance_df.head(top_n) plt.figure(figsize=(10, 6)) sns.barplot( data=top_features, x="importance", y="feature" ) plt.title(f"Top {top_n} - Feature importance native") plt.tight_layout() plt.savefig(figures_dir / "feature_importance_native.png", dpi=300) plt.close() # Feature importance groupée top_grouped = grouped_importance_df.head(10) plt.figure(figsize=(10, 6)) sns.barplot( data=top_grouped, x="importance", y="feature_group" ) plt.title("Feature importance métier") plt.tight_layout() plt.savefig(figures_dir / "feature_importance_grouped.png", dpi=300) plt.close() # Permutation importance top_perm = permutation_df.head(10).sort_values( by="importance_mean", ascending=True ) plt.figure(figsize=(10, 6)) plt.barh( y=top_perm["feature"], width=top_perm["importance_mean"], xerr=top_perm["importance_std"] ) plt.title("Permutation importance") plt.tight_layout() plt.savefig(figures_dir / "permutation_importance.png", dpi=300) plt.close() # Résidus log if "y_pred_log" in residuals_df.columns: plt.figure(figsize=(8, 5)) sns.scatterplot( data=residuals_df, x="y_pred_log", y="residual_log" ) plt.axhline(0, linestyle="--") plt.title("Résidus - log") plt.tight_layout() plt.savefig(figures_dir / "residuals_log.png", dpi=300) plt.close() # Résidus réel if "y_pred_real" in residuals_df.columns: plt.figure(figsize=(8, 5)) sns.scatterplot( data=residuals_df, x="y_pred_real", y="residual_real" ) plt.axhline(0, linestyle="--") plt.title("Résidus - réel") plt.tight_layout() plt.savefig(figures_dir / "residuals_real.png", dpi=300) plt.close() print(f"Artefacts sauvegardés dans : {base_path}")