""" MODULE: FORECASTING ENGINE - EXPERT EDITION AMÉLIORÉE ====================================================== Responsabilité : Entraînement ML, Audit Comparatif (3 Modèles) et Prédiction. Fonctionnalités : - Split Train/Val/Test temporel - Prédiction récursive avec saisonnalité - SHAP systématique et lisible - Métriques explicatives pour non-techniciens - Visualisations avancées Améliorations v2.0 : ✅ Split 3 parties (Train 70% / Val 15% / Test 15%) ✅ Prédiction récursive (utilise ses propres lags) ✅ SHAP toujours affiché avec explication ✅ Tableau comparatif détaillé des modèles ✅ Intervalles de confiance sur prédictions ✅ Alertes automatiques sur anomalies prévues ✅ Export des prédictions """ import pandas as pd import numpy as np import streamlit as st import plotly.graph_objects as go from plotly.subplots import make_subplots from datetime import datetime, timedelta import warnings # ML Models from sklearn.ensemble import RandomForestRegressor from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score import xgboost as xgb import lightgbm as lgb # SHAP pour interprétabilité try: import shap import matplotlib.pyplot as plt SHAP_AVAILABLE = True except ImportError: SHAP_AVAILABLE = False warnings.filterwarnings('ignore') # ===================================================== # ============= MOTEUR DE PRÉDICTION ================== # ===================================================== class HospitalForecaster: """ Moteur de prédiction hospitalière avec comparaison de modèles et interprétabilité SHAP. """ def __init__(self, ml_dataset): self.df = ml_dataset.copy() self.df['date'] = pd.to_datetime(self.df['date']) # Stockage des modèles et résultats self.models = {} self.metrics = {} self.predictions = { 'train': {}, 'val': {}, 'test': {} } self.best_model_name = None # Datasets splits self.X_train = None self.X_val = None self.X_test = None self.y_train = None self.y_val = None self.y_test = None self.dates_train = None self.dates_val = None self.dates_test = None self.feature_names = None def prepare_data(self, target_col='nb_admissions', val_days=30, test_days=30): """ Split temporel en 3 parties : Train / Validation / Test. Args: target_col: Variable cible à prédire val_days: Nombre de jours pour validation test_days: Nombre de jours pour test Returns: Dictionnaire avec les plages de dates """ # Tri chronologique df = self.df.sort_values('date').reset_index(drop=True) # Exclusion des colonnes non-features exclude_cols = [ target_col, 'date', 'stay_id', 'patient_id', 'scenario', 'service_nom', 'periode_conges_label' ] # Sélection STRICTE des colonnes numériques uniquement all_cols = [c for c in df.columns if c not in exclude_cols] feature_cols = [c for c in all_cols if np.issubdtype(df[c].dtype, np.number)] # Vérification de la cible if target_col not in df.columns: raise ValueError(f"❌ Colonne cible '{target_col}' introuvable") # Calcul des indices de split n = len(df) test_start = n - test_days val_start = test_start - val_days # Extraction X et y X = df[feature_cols] y = df[target_col] dates = df['date'] # Split self.X_train = X.iloc[:val_start] self.X_val = X.iloc[val_start:test_start] self.X_test = X.iloc[test_start:] self.y_train = y.iloc[:val_start] self.y_val = y.iloc[val_start:test_start] self.y_test = y.iloc[test_start:] self.dates_train = dates.iloc[:val_start] self.dates_val = dates.iloc[val_start:test_start] self.dates_test = dates.iloc[test_start:] self.feature_names = feature_cols self.target_col = target_col # Informations de split split_info = { 'train': { 'start': self.dates_train.min(), 'end': self.dates_train.max(), 'n_samples': len(self.X_train) }, 'val': { 'start': self.dates_val.min(), 'end': self.dates_val.max(), 'n_samples': len(self.X_val) }, 'test': { 'start': self.dates_test.min(), 'end': self.dates_test.max(), 'n_samples': len(self.X_test) } } return split_info def train_all_models(self): """ Entraînement des 3 modèles avec validation. Returns: Dictionnaire des métriques """ progress_bar = st.progress(0, text=" Entraînement en cours...") # ===== 1. LightGBM ===== progress_bar.progress(0.1, text=" LightGBM...") lgb_model = lgb.LGBMRegressor( n_estimators=500, learning_rate=0.05, num_leaves=31, max_depth=6, min_child_samples=20, subsample=0.8, colsample_bytree=0.8, random_state=42, verbose=-1, n_jobs=-1 ) lgb_model.fit( self.X_train, self.y_train, eval_set=[(self.X_val, self.y_val)], callbacks=[lgb.early_stopping(stopping_rounds=50, verbose=False)] ) self.models['LightGBM'] = lgb_model self._evaluate_model('LightGBM', lgb_model) # ===== 2. XGBoost ===== progress_bar.progress(0.4, text=" XGBoost...") xgb_model = xgb.XGBRegressor( n_estimators=500, learning_rate=0.05, max_depth=6, min_child_weight=3, subsample=0.8, colsample_bytree=0.8, random_state=42, n_jobs=-1, early_stopping_rounds=50, eval_metric='rmse' ) xgb_model.fit( self.X_train, self.y_train, eval_set=[(self.X_val, self.y_val)], verbose=False ) self.models['XGBoost'] = xgb_model self._evaluate_model('XGBoost', xgb_model) # ===== 3. Random Forest ===== progress_bar.progress(0.7, text=" Random Forest...") rf_model = RandomForestRegressor( n_estimators=300, max_depth=15, min_samples_split=10, min_samples_leaf=5, max_features='sqrt', random_state=42, n_jobs=-1, verbose=0 ) rf_model.fit(self.X_train, self.y_train) self.models['Random Forest'] = rf_model self._evaluate_model('Random Forest', rf_model) progress_bar.progress(1.0, text="✅ Entraînement terminé!") progress_bar.empty() # Sélection du meilleur modèle self._select_best_model() return self.metrics def _evaluate_model(self, model_name, model): """ Évalue un modèle sur les 3 sets (train, val, test). Args: model_name: Nom du modèle model: Instance du modèle entraîné """ # Prédictions sur les 3 sets pred_train = model.predict(self.X_train) pred_val = model.predict(self.X_val) pred_test = model.predict(self.X_test) # Stockage des prédictions self.predictions['train'][model_name] = pred_train self.predictions['val'][model_name] = pred_val self.predictions['test'][model_name] = pred_test # Métriques sur le test set (évaluation finale) rmse = np.sqrt(mean_squared_error(self.y_test, pred_test)) mae = mean_absolute_error(self.y_test, pred_test) r2 = r2_score(self.y_test, pred_test) # MAPE (Mean Absolute Percentage Error) mask = self.y_test != 0 if mask.sum() > 0: mape = np.mean(np.abs((self.y_test[mask] - pred_test[mask]) / self.y_test[mask])) * 100 else: mape = np.nan # Métriques sur validation (pour early stopping) rmse_val = np.sqrt(mean_squared_error(self.y_val, pred_val)) mae_val = mean_absolute_error(self.y_val, pred_val) self.metrics[model_name] = { 'RMSE': rmse, 'MAE': mae, 'R²': r2, 'MAPE': mape, 'RMSE_val': rmse_val, 'MAE_val': mae_val } def _select_best_model(self): """ Sélection automatique du meilleur modèle basée sur RMSE + MAE. """ scores = {} for model_name, metrics in self.metrics.items(): # Score composite (moyenne normalisée RMSE + MAE) score = (metrics['RMSE'] + metrics['MAE']) / 2 scores[model_name] = score # Meilleur = score le plus bas self.best_model_name = min(scores, key=scores.get) def generate_future_forecast_recursive(self, months=3): """ Prédiction RÉCURSIVE vers le futur avec saisonnalité. Le modèle utilise ses propres prédictions comme lags pour les prédictions suivantes, conservant ainsi les tendances. Args: months: Nombre de mois à prédire Returns: Tuple (dates_futures, prédictions, intervalles_confiance) """ days = months * 30 model = self.models[self.best_model_name] # Dernière ligne connue (base pour récursion) last_row = self.df.iloc[-1:][self.feature_names].copy() last_date = self.df['date'].max() future_dates = [] future_predictions = [] # Identification des colonnes de lags lag_cols = [c for c in self.feature_names if '_lag_' in c] rolling_cols = [c for c in self.feature_names if any(x in c for x in ['_mean_', '_std_', '_min_', '_max_'])] # Prédiction jour par jour for i in range(1, days + 1): current_date = last_date + timedelta(days=i) future_dates.append(current_date) # Mise à jour des features temporelles row = last_row.copy() # Features temporelles basiques if 'jour_semaine' in row.columns: row['jour_semaine'] = current_date.dayofweek if 'mois' in row.columns: row['mois'] = current_date.month if 'trimestre' in row.columns: row['trimestre'] = current_date.quarter if 'jour_annee' in row.columns: row['jour_annee'] = current_date.dayofyear # Features cycliques if 'mois_sin' in row.columns: row['mois_sin'] = np.sin(2 * np.pi * current_date.month / 12) if 'mois_cos' in row.columns: row['mois_cos'] = np.cos(2 * np.pi * current_date.month / 12) if 'jour_semaine_sin' in row.columns: row['jour_semaine_sin'] = np.sin(2 * np.pi * current_date.dayofweek / 7) if 'jour_semaine_cos' in row.columns: row['jour_semaine_cos'] = np.cos(2 * np.pi * current_date.dayofweek / 7) # Features de périodes hospitalières mois = current_date.month if 'periode_grippe' in row.columns: row['periode_grippe'] = int((mois >= 11) or (mois <= 3)) if 'periode_canicule' in row.columns: row['periode_canicule'] = int((mois >= 6) and (mois <= 8)) if 'periode_gastro' in row.columns: row['periode_gastro'] = int((mois >= 11) or (mois <= 2)) if 'periode_bronchiolite' in row.columns: row['periode_bronchiolite'] = int((mois >= 10) and (mois <= 2)) # Weekend if 'est_weekend' in row.columns: row['est_weekend'] = int(current_date.dayofweek >= 5) # Prédiction pred = model.predict(row)[0] future_predictions.append(pred) # RÉCURSION : Mise à jour des lags avec la nouvelle prédiction # Pour le prochain jour, cette prédiction devient le lag_1 if len(future_predictions) >= 1 and f'{self.target_col}_lag_1' in lag_cols: row[f'{self.target_col}_lag_1'] = pred if len(future_predictions) >= 7 and f'{self.target_col}_lag_7' in lag_cols: row[f'{self.target_col}_lag_7'] = future_predictions[-7] if len(future_predictions) >= 14 and f'{self.target_col}_lag_14' in lag_cols: row[f'{self.target_col}_lag_14'] = future_predictions[-14] # Mise à jour des rolling stats (approximation simple) if len(future_predictions) >= 7: recent_7 = future_predictions[-7:] if f'{self.target_col}_mean_7j' in rolling_cols: row[f'{self.target_col}_mean_7j'] = np.mean(recent_7) if f'{self.target_col}_std_7j' in rolling_cols: row[f'{self.target_col}_std_7j'] = np.std(recent_7) if len(future_predictions) >= 14: recent_14 = future_predictions[-14:] if f'{self.target_col}_mean_14j' in rolling_cols: row[f'{self.target_col}_mean_14j'] = np.mean(recent_14) if f'{self.target_col}_std_14j' in rolling_cols: row[f'{self.target_col}_std_14j'] = np.std(recent_14) # Préparation pour la prochaine itération last_row = row # Calcul des intervalles de confiance (basé sur l'erreur du modèle) mae = self.metrics[self.best_model_name]['MAE'] lower_bound = np.array(future_predictions) - 1.96 * mae upper_bound = np.array(future_predictions) + 1.96 * mae confidence_intervals = { 'lower': lower_bound, 'upper': upper_bound } return future_dates, np.array(future_predictions), confidence_intervals def compute_shap_values(self, n_samples=200): """ Calcule les valeurs SHAP pour interprétabilité. Args: n_samples: Nombre d'échantillons pour le calcul Returns: Tuple (explainer, shap_values, X_sample) ou (None, None, None) """ if not SHAP_AVAILABLE: return None, None, None model = self.models[self.best_model_name] # Échantillonnage pour la vitesse X_sample = self.X_test.sample(n=min(n_samples, len(self.X_test)), random_state=42) try: # TreeExplainer optimisé pour les modèles d'arbres explainer = shap.TreeExplainer(model) shap_values = explainer.shap_values(X_sample) return explainer, shap_values, X_sample except Exception as e: st.warning(f"⚠️ Erreur SHAP : {str(e)}") return None, None, None def detect_future_anomalies(self, future_predictions, threshold_percentile=90): """ Détecte les anomalies prévues dans les prédictions futures. Args: future_predictions: Array des prédictions futures threshold_percentile: Percentile pour définir une anomalie Returns: Liste d'alertes """ # Calcul du seuil basé sur l'historique historical_values = self.df[self.target_col].values threshold = np.percentile(historical_values, threshold_percentile) # Détection des pics alerts = [] for i, pred in enumerate(future_predictions): if pred > threshold: alerts.append({ 'day': i + 1, 'value': pred, 'threshold': threshold, 'severity': 'HIGH' if pred > threshold * 1.2 else 'MEDIUM' }) return alerts # ===================================================== # ============= INTERFACE STREAMLIT =================== # ===================================================== def show_forecasting_module(client, sheet_name): """ Interface Streamlit pour le module de forecasting. """ st.markdown("## 🔮 ORACLE : FORECASTING & AUDIT") st.caption("Intelligence Artificielle de prédiction des flux hospitaliers avec interprétabilité complète.") # ===== VÉRIFICATION DONNÉES ===== if 'ml_features' not in st.session_state: st.error("❌ **Données ML absentes.** Veuillez d'abord exécuter le module **Feature Engineering**.") if st.button(" Aller au Feature Engineering"): st.switch_page("pages/03_Features.py") return df = st.session_state['ml_features'] st.success(f"✅ Dataset ML chargé : **{len(df):,}** échantillons") # ===== CONFIGURATION ===== st.divider() with st.expander(" Configuration de l'Entraînement", expanded=False): col1, col2, col3 = st.columns(3) with col1: target = st.selectbox( "Variable Cible", ['nb_admissions', 'taux_occupation', 'tension_personnel'], index=0, help="Variable à prédire" ) with col2: val_days = st.number_input( "Jours Validation", min_value=15, max_value=60, value=30, help="Nombre de jours pour le set de validation" ) with col3: test_days = st.number_input( "Jours Test", min_value=15, max_value=60, value=30, help="Nombre de jours pour le test final" ) # ===== ENTRAÎNEMENT ===== if 'forecaster' not in st.session_state or st.session_state.get('forecast_target') != target: st.info("🚀 **Lancement de l'entraînement des 3 modèles...**") forecaster = HospitalForecaster(df) try: # Préparation des données with st.spinner(" Préparation des données (Train/Val/Test)..."): split_info = forecaster.prepare_data( target_col=target, val_days=val_days, test_days=test_days ) # Affichage des plages de dates st.markdown("### Plages de Dates des Sets") col1, col2, col3 = st.columns(3) with col1: st.metric( "🟢 Train", f"{split_info['train']['n_samples']} jours", f"{split_info['train']['start'].strftime('%d/%m/%Y')} → {split_info['train']['end'].strftime('%d/%m/%Y')}" ) with col2: st.metric( "🟡 Validation", f"{split_info['val']['n_samples']} jours", f"{split_info['val']['start'].strftime('%d/%m/%Y')} → {split_info['val']['end'].strftime('%d/%m/%Y')}" ) with col3: st.metric( "🔴 Test", f"{split_info['test']['n_samples']} jours", f"{split_info['test']['start'].strftime('%d/%m/%Y')} → {split_info['test']['end'].strftime('%d/%m/%Y')}" ) st.divider() # Entraînement des modèles metrics = forecaster.train_all_models() # Stockage dans session_state st.session_state['forecaster'] = forecaster st.session_state['metrics'] = metrics st.session_state['split_info'] = split_info st.session_state['forecast_target'] = target st.success("✅ **Entraînement terminé avec succès !**") except Exception as e: st.error(f"❌ **Erreur lors de l'entraînement :** {str(e)}") import traceback with st.expander(" Détails de l'erreur"): st.code(traceback.format_exc()) return # ===== RÉCUPÉRATION DES RÉSULTATS ===== forecaster = st.session_state['forecaster'] metrics = st.session_state['metrics'] split_info = st.session_state['split_info'] best = forecaster.best_model_name # ===== ONGLETS PRINCIPAUX ===== tab_future, tab_audit, tab_export = st.tabs([ " Prédictions Futures", " Audit & Comparaison", " Export & Données" ]) # ================================================================ # ONGLET 1 : PRÉDICTIONS FUTURES # ================================================================ with tab_future: st.markdown("### Projections Récursives avec Saisonnalité") col_h1, col_h2 = st.columns([3, 1]) with col_h1: horizon_label = st.radio( "Horizon de prédiction", ["1 Mois", "3 Mois", "6 Mois", "12 Mois"], horizontal=True, index=1 ) with col_h2: show_confidence = st.checkbox("Intervalle de confiance", value=True) months_map = {"1 Mois": 1, "3 Mois": 3, "6 Mois": 6, "12 Mois": 12} months = months_map[horizon_label] with st.spinner(" Calcul de la trajectoire récursive avec saisonnalité..."): f_dates, f_values, conf_intervals = forecaster.generate_future_forecast_recursive(months) # Graphique des prédictions futures fig = go.Figure() # Historique récent (90 derniers jours) hist = df.iloc[-90:] fig.add_trace(go.Scatter( x=hist['date'], y=hist[target], name=" Historique Réel", line=dict(color='gray', width=2), opacity=0.6 )) # Prédictions futures fig.add_trace(go.Scatter( x=f_dates, y=f_values, name=f" Projection {best}", line=dict(color='#00E676', width=3), mode='lines+markers', marker=dict(size=5) )) # Intervalle de confiance if show_confidence: fig.add_trace(go.Scatter( x=f_dates + f_dates[::-1], y=list(conf_intervals['upper']) + list(conf_intervals['lower'][::-1]), fill='toself', fillcolor='rgba(0, 230, 118, 0.2)', line=dict(color='rgba(255,255,255,0)'), name='Intervalle de confiance (95%)', showlegend=True )) # Ligne verticale "Aujourd'hui" last_hist_date = hist['date'].max() # --- FIX COMPATIBILITÉ PANDAS 2.x --- # On convertit le Timestamp en millisecondes (float) pour éviter # l'erreur "Addition/subtraction of integers with Timestamp" # car Plotly utilise sum() en interne qui commence par 0. vline_x = last_hist_date.timestamp() * 1000 # ------------------------------------ fig.add_vline( x=vline_x, # On utilise la version convertie ici line_width=2, line_dash="dash", line_color="white", annotation_text="Aujourd'hui", annotation_position="top" ) fig.update_layout( title=f"Projection {horizon_label} - {target.replace('_', ' ').title()}", template="plotly_dark", height=500, hovermode="x unified", xaxis_title="Date", yaxis_title=target.replace('_', ' ').title() ) st.plotly_chart(fig, use_container_width=True) # Statistiques de la projection col1, col2, col3, col4 = st.columns(4) col1.metric("Moyenne Prévue", f"{np.mean(f_values):.1f}") col2.metric("Maximum Prévu", f"{np.max(f_values):.1f}") col3.metric("Minimum Prévu", f"{np.min(f_values):.1f}") col4.metric("Tendance", " Hausse" if f_values[-1] > f_values[0] else " Baisse", f"{((f_values[-1] / f_values[0]) - 1) * 100:+.1f}%") # Détection d'anomalies futures st.markdown("#### 🚨 Alertes Préventives") alerts = forecaster.detect_future_anomalies(f_values) if alerts: st.warning(f"⚠️ **{len(alerts)} pic(s) détecté(s)** dans les {months} prochains mois") for alert in alerts[:5]: # Top 5 severity_emoji = "🔴" if alert['severity'] == 'HIGH' else "🟠" st.write(f"{severity_emoji} **Jour +{alert['day']}** : Valeur prévue = **{alert['value']:.0f}** " f"(seuil normal : {alert['threshold']:.0f})") else: st.success("✅ Aucun pic anormal détecté dans la période") st.caption(f" Prédiction générée par **{best}** | Fiabilité : **{metrics[best]['R²']*100:.0f}%**") # ================================================================ # ONGLET 2 : AUDIT & COMPARAISON # ================================================================ with tab_audit: st.markdown("### Le Champion de la Prédiction") # Annonce du vainqueur col_winner, col_reset = st.columns([4, 1]) with col_winner: st.success(f""" ### Gagnant : **{best}** Sur les **{split_info['test']['n_samples']} derniers jours** de test, ce modèle a montré la meilleure précision avec une erreur moyenne de seulement **{metrics[best]['MAE']:.2f}** unités par jour. **Fiabilité globale :** {metrics[best]['R²']*100:.1f}% (Le modèle explique {metrics[best]['R²']*100:.1f}% des variations observées) """) with col_reset: if st.button(" Réinitialiser", help="Relancer l'entraînement"): for key in ['forecaster', 'metrics', 'split_info', 'forecast_target']: if key in st.session_state: del st.session_state[key] st.rerun() st.divider() # ===== TABLEAU COMPARATIF ===== st.markdown("### Tableau Comparatif Détaillé") comparison_data = [] for model_name in ['XGBoost', 'LightGBM', 'Random Forest']: m = metrics[model_name] comparison_data.append({ 'Modèle': f"{'🏆 ' if model_name == best else ''}{model_name}", 'R² (Fiabilité %)': f"{m['R²']*100:.2f}%", 'RMSE': f"{m['RMSE']:.2f}", 'MAE (Erreur Moy.)': f"{m['MAE']:.2f}", 'MAPE (Erreur %)': f"{m['MAPE']:.2f}%" if not np.isnan(m['MAPE']) else "N/A" }) comparison_df = pd.DataFrame(comparison_data) st.dataframe( comparison_df, use_container_width=True, hide_index=True ) # Explications des métriques with st.expander(" Comprendre les Métriques"): st.markdown(""" **R² (Coefficient de Détermination) :** - Mesure de 0% à 100% : plus c'est proche de 100%, mieux c'est - Indique quelle part de la variation est expliquée par le modèle - Exemple : 85% = le modèle comprend 85% des raisons des fluctuations **RMSE (Root Mean Squared Error) :** - Erreur moyenne au carré, puis racine carrée - Pénalise fortement les grandes erreurs - Plus la valeur est basse, mieux c'est - Utile en santé car les grosses erreurs sont graves **MAE (Mean Absolute Error) :** - Erreur moyenne en valeur absolue - S'exprime dans l'unité de la cible (ex: nombre d'admissions) - Plus facile à interpréter : "en moyenne, je me trompe de X unités" - Plus robuste aux valeurs extrêmes que RMSE **MAPE (Mean Absolute Percentage Error) :** - Erreur moyenne en pourcentage - Indépendant de l'échelle des données - Exemple : 5% = en moyenne, l'erreur est de 5% de la valeur réelle """) st.divider() # ===== GRAPHIQUE COMPARATIF SUR TEST SET ===== st.markdown("### Comparaison Visuelle sur le Test Set") fig_comp = go.Figure() # Réalité (en blanc épais) fig_comp.add_trace(go.Scatter( x=forecaster.dates_test, y=forecaster.y_test, name=" RÉALITÉ", line=dict(color='white', width=4, dash='dot'), mode='lines+markers', marker=dict(size=6) )) # Les 3 modèles colors = { 'XGBoost': '#1f77b4', 'LightGBM': '#ff7f0e', 'Random Forest': '#2ca02c' } for model_name in ['XGBoost', 'LightGBM', 'Random Forest']: pred = forecaster.predictions['test'][model_name] is_best = (model_name == best) fig_comp.add_trace(go.Scatter( x=forecaster.dates_test, y=pred, name=f"{'🏆 ' if is_best else ''}{model_name}", line=dict( color=colors[model_name], width=3 if is_best else 1.5, dash='solid' if is_best else 'dash' ), opacity=1.0 if is_best else 0.5 )) fig_comp.update_layout( title="Prédictions vs Réalité sur la Période de Test", template="plotly_dark", height=500, hovermode="x unified", xaxis_title="Date", yaxis_title=target.replace('_', ' ').title() ) st.plotly_chart(fig_comp, use_container_width=True) st.caption(" **Plus une ligne colorée est proche de la ligne blanche, meilleur est le modèle**") st.divider() # ===== SHAP : INTERPRÉTABILITÉ ===== st.markdown("### Comprendre les Décisions de l'IA (SHAP)") st.info(""" **SHAP** (SHapley Additive exPlanations) révèle **quelles variables influencent le plus** les prédictions du modèle. Plus la barre est longue, plus cette feature est importante pour les décisions de l'IA. """) if SHAP_AVAILABLE: with st.spinner(" Calcul des valeurs SHAP (analyse d'importance)..."): explainer, shap_values, X_sample = forecaster.compute_shap_values(n_samples=200) if shap_values is not None: # Summary plot (importance globale) fig_shap, ax = plt.subplots(figsize=(10, 6)) fig_shap.patch.set_facecolor('#0e1117') ax.set_facecolor('#0e1117') shap.summary_plot( shap_values, X_sample, max_display=15, plot_type="bar", show=False, color='#00E676' ) # Styling pour fond sombre ax.tick_params(colors='white', labelsize=11) ax.xaxis.label.set_color('white') ax.yaxis.label.set_color('white') ax.set_xlabel('Impact sur la prédiction', color='white', fontsize=12) ax.set_title(f'Top 15 Features les Plus Importantes - {best}', color='white', fontsize=14, fontweight='bold') st.pyplot(fig_shap) plt.close() # Explications textuelles st.markdown("#### Interprétation") # Top 5 features feature_importance = pd.DataFrame({ 'Feature': X_sample.columns, 'Importance': np.abs(shap_values).mean(axis=0) }).sort_values('Importance', ascending=False).head(5) st.write("**Les 5 facteurs clés identifiés par l'IA :**") for idx, row in feature_importance.iterrows(): feature = row['Feature'] # Traduction en français if 'lag' in feature: explanation = f" Historique récent ({feature})" elif 'mean' in feature or 'rolling' in feature: explanation = f" Tendance sur plusieurs jours ({feature})" elif 'taux' in feature: explanation = f" Taux ou ratio ({feature})" elif 'personnel' in feature: explanation = f" Ressources humaines ({feature})" elif 'scenario' in feature or 'intensity' in feature: explanation = f" Scénario/événement ({feature})" elif 'mois' in feature or 'jour' in feature: explanation = f" Facteur temporel ({feature})" else: explanation = f" {feature}" st.write(f"- {explanation}") else: st.warning("⚠️ Impossible de calculer les valeurs SHAP pour ce modèle") else: st.warning("⚠️ **SHAP n'est pas installé.** Installez avec : `pip install shap`") # ================================================================ # ONGLET 3 : EXPORT & DONNÉES # ================================================================ with tab_export: st.markdown("### 📥 Téléchargement des Résultats") col1, col2 = st.columns(2) with col1: # Export des métriques st.markdown("#### Métriques des Modèles") metrics_df = pd.DataFrame(metrics).T metrics_csv = metrics_df.to_csv(index=True).encode('utf-8') st.download_button( label="📥 Télécharger Métriques (CSV)", data=metrics_csv, file_name=f'model_metrics_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv', mime='text/csv' ) st.dataframe(metrics_df, use_container_width=True) with col2: # Export des prédictions sur test st.markdown("#### Prédictions Test Set") predictions_df = pd.DataFrame({ 'date': forecaster.dates_test, 'reel': forecaster.y_test, 'XGBoost': forecaster.predictions['test']['XGBoost'], 'LightGBM': forecaster.predictions['test']['LightGBM'], 'Random_Forest': forecaster.predictions['test']['Random Forest'] }) predictions_csv = predictions_df.to_csv(index=False).encode('utf-8') st.download_button( label="📥 Télécharger Prédictions (CSV)", data=predictions_csv, file_name=f'predictions_test_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv', mime='text/csv' ) st.dataframe(predictions_df.head(10), use_container_width=True) # Export des prédictions futures st.markdown("#### Prédictions Futures") if 'f_dates' in locals() and 'f_values' in locals(): future_df = pd.DataFrame({ 'date': f_dates, 'prediction': f_values, 'lower_bound': conf_intervals['lower'], 'upper_bound': conf_intervals['upper'] }) future_csv = future_df.to_csv(index=False).encode('utf-8') st.download_button( label="📥 Télécharger Prédictions Futures (CSV)", data=future_csv, file_name=f'predictions_futures_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv', mime='text/csv' ) st.dataframe(future_df.head(30), use_container_width=True) # Point d'entrée si appelé directement if __name__ == "__main__": # Pour test standalone show_forecasting_module(None, None)