Spaces:
Running
Running
| import pandas as pd | |
| import numpy as np | |
| from sklearn.linear_model import LinearRegression | |
| from sklearn.metrics import mean_absolute_error, mean_squared_error | |
| from dateutil.relativedelta import relativedelta | |
| class VortexOutFlux: | |
| """ | |
| Analyse et prédit les flux de sortie mensuels via régression linéaire. | |
| Fournit également les données de comparaison réel vs prédit | |
| et les signaux d'alerte d'investissement. | |
| """ | |
| def __init__(self, df_forecasting): | |
| self.df = df_forecasting.copy() | |
| self._prepare_data() | |
| # ========================================================================= | |
| # PRÉPARATION | |
| # ========================================================================= | |
| def _prepare_data(self): | |
| self.df['Date'] = pd.to_datetime(self.df['Date'], errors='coerce') | |
| self.df.dropna(subset=['Date', 'Montant_Total_Sortie'], inplace=True) | |
| self.df = self.df[self.df['Montant_Total_Sortie'] != 0] | |
| self.df = self.df.sort_values('Date').reset_index(drop=True) | |
| self.df['date_ordinal'] = self.df['Date'].apply(lambda x: x.toordinal()) | |
| # ========================================================================= | |
| # MODÈLE DE RÉGRESSION | |
| # ========================================================================= | |
| def train_linear_model(self): | |
| """ | |
| Entraîne le modèle de régression linéaire. | |
| Returns: | |
| dict : modèle sklearn + métriques de performance | |
| """ | |
| if len(self.df) < 2: | |
| return {'error': 'Données insuffisantes (minimum 2 points requis)'} | |
| X = self.df[['date_ordinal']].values | |
| y = self.df['Montant_Total_Sortie'].values | |
| model_lr = LinearRegression() | |
| model_lr.fit(X, y) | |
| y_pred = model_lr.predict(X) | |
| residuals = y - y_pred | |
| mae = mean_absolute_error(y, y_pred) | |
| rmse = np.sqrt(mean_squared_error(y, y_pred)) | |
| ss_res = np.sum(residuals ** 2) | |
| ss_tot = np.sum((y - np.mean(y)) ** 2) | |
| r_squared = 1 - (ss_res / ss_tot) if ss_tot > 0 else 0 | |
| r_coeff = np.sqrt(r_squared) if r_squared >= 0 else 0 | |
| std_error = np.sqrt(np.sum(residuals ** 2) / (len(y) - 2)) if len(y) > 2 else np.std(residuals) | |
| prediction_interval = 1.96 * std_error | |
| return { | |
| 'model_lr': model_lr, | |
| 'mae': mae, | |
| 'rmse': rmse, | |
| 'pente': model_lr.coef_[0], | |
| 'r_squared': r_squared, | |
| 'r_coefficient': r_coeff, | |
| 'historical_predictions': y_pred, | |
| 'residuals': residuals, | |
| 'std_error': std_error, | |
| 'prediction_interval': prediction_interval, | |
| } | |
| # ========================================================================= | |
| # PRÉDICTIONS FUTURES | |
| # ========================================================================= | |
| def predict_next_months(self, n_months=3): | |
| """ | |
| Prédit les n prochains mois avec intervalles de confiance à 95%. | |
| Returns: | |
| dict : { 'predictions': DataFrame, 'models': dict } | |
| """ | |
| models = self.train_linear_model() | |
| if 'error' in models: | |
| return models | |
| last_date = self.df['Date'].max() | |
| future_dates = [] | |
| for i in range(1, n_months + 1): | |
| fd = last_date + relativedelta(months=i) | |
| future_dates.append(fd.replace(day=1)) | |
| future_dates = pd.to_datetime(future_dates) | |
| future_ordinals = np.array([d.toordinal() for d in future_dates]).reshape(-1, 1) | |
| predictions = models['model_lr'].predict(future_ordinals) | |
| n = len(self.df) | |
| X_mean = self.df['date_ordinal'].mean() | |
| X_std = self.df['date_ordinal'].std() | |
| lower_bounds, upper_bounds = [], [] | |
| for idx, ordinal in enumerate(future_ordinals.flatten()): | |
| distance = (ordinal - X_mean) / X_std if X_std > 0 else 0 | |
| adjustment = np.sqrt(1 + 1 / n + distance ** 2) | |
| margin = models['prediction_interval'] * adjustment | |
| pred_val = predictions[idx] | |
| lower_bounds.append(max(0, pred_val - margin)) | |
| upper_bounds.append(pred_val + margin) | |
| predictions_df = pd.DataFrame({ | |
| 'Date': future_dates, | |
| 'Montant_Predit': predictions, | |
| 'Borne_Inf': lower_bounds, | |
| 'Borne_Sup': upper_bounds, | |
| }) | |
| return {'predictions': predictions_df, 'models': models} | |
| # ========================================================================= | |
| # TABLEAU DE COMPARAISON (réel vs prédit — sur un exercice) | |
| # ========================================================================= | |
| def build_comparison_table(self, models, cutoff_date, end_date=None): | |
| """ | |
| Construit le tableau comparatif réel vs prédit de cutoff_date (inclus) | |
| à end_date (exclu, None = sans limite) — en pratique les bornes d'un | |
| exercice (voir core/exercice.bornes). | |
| Colonnes produites : | |
| - Mois : str formaté (ex. "Janvier 2026") | |
| - Capital_Reel : float — montant réel de la feuille Forecasting | |
| - Capital_Predit : float — montant prédit par le modèle | |
| - Evolution_Pct : float — variation réelle (mois N vs mois N-1) en % | |
| Returns: | |
| pd.DataFrame | |
| """ | |
| cutoff = pd.to_datetime(cutoff_date) | |
| masque = self.df['Date'] >= cutoff | |
| if end_date is not None: | |
| masque &= self.df['Date'] < pd.to_datetime(end_date) | |
| df_cut = self.df[masque].copy() | |
| if df_cut.empty: | |
| return pd.DataFrame(columns=['Mois', 'Capital_Reel', 'Capital_Predit', 'Evolution_Pct']) | |
| # Prédictions sur les dates historiques filtrées | |
| ordinals = df_cut['date_ordinal'].values.reshape(-1, 1) | |
| df_cut['Capital_Predit'] = models['model_lr'].predict(ordinals) | |
| # Evolution réelle mois N vs mois N-1 | |
| df_all = self.df.copy() | |
| df_all['prev_reel'] = df_all['Montant_Total_Sortie'].shift(1) | |
| df_cut = df_cut.merge(df_all[['Date', 'prev_reel']], on='Date', how='left') | |
| df_cut['Evolution_Pct'] = df_cut.apply( | |
| lambda r: ((r['Montant_Total_Sortie'] - r['prev_reel']) / r['prev_reel'] * 100) | |
| if pd.notna(r['prev_reel']) and r['prev_reel'] != 0 else None, | |
| axis=1 | |
| ) | |
| df_cut['Mois'] = df_cut['Date'].dt.strftime('%B %Y').str.capitalize() | |
| return df_cut[['Mois', 'Montant_Total_Sortie', 'Capital_Predit', 'Evolution_Pct']].rename( | |
| columns={'Montant_Total_Sortie': 'Capital_Reel'} | |
| ).reset_index(drop=True) | |
| # ========================================================================= | |
| # CARTE MOIS SUIVANT + SIGNAL D'ALERTE | |
| # ========================================================================= | |
| def compute_next_month_signal(self, models): | |
| """ | |
| Calcule les données pour la carte du mois suivant et le signal d'investissement. | |
| Returns: | |
| dict : | |
| - mois_predit : str | |
| - capital_predit : float | |
| - capital_reel_actuel : float | |
| - mois_actuel : str | |
| - evolution_pct : float | |
| - signal : 'REDUIRE' | 'MAINTENIR' | 'AUGMENTER' | |
| - message_signal : str | |
| - seuil_reduction : float | |
| - seuil_augmentation : float | |
| """ | |
| last_date = self.df['Date'].max() | |
| next_month = (last_date + relativedelta(months=1)).replace(day=1) | |
| next_ordinal = np.array([[next_month.toordinal()]]) | |
| capital_predit = float(models['model_lr'].predict(next_ordinal)[0]) | |
| last_row = self.df.iloc[-1] | |
| capital_reel_actuel = float(last_row['Montant_Total_Sortie']) | |
| mois_actuel = last_row['Date'].strftime('%B %Y').capitalize() | |
| evolution_pct = ( | |
| ((capital_predit - capital_reel_actuel) / capital_reel_actuel * 100) | |
| if capital_reel_actuel != 0 else 0.0 | |
| ) | |
| SEUIL_REDUCTION = -10.0 | |
| SEUIL_AUGMENTATION = +10.0 | |
| if evolution_pct <= SEUIL_REDUCTION: | |
| signal = 'REDUIRE' | |
| message_signal = ( | |
| f"Le modèle anticipe une baisse de {abs(evolution_pct):.1f}% des sorties " | |
| f"en {next_month.strftime('%B %Y').capitalize()} " | |
| f"par rapport à {mois_actuel}. " | |
| f"Recommandation : réduire les engagements de capital pour ce mois." | |
| ) | |
| elif evolution_pct >= SEUIL_AUGMENTATION: | |
| signal = 'AUGMENTER' | |
| message_signal = ( | |
| f"Le modèle anticipe une hausse de {evolution_pct:.1f}% des sorties " | |
| f"en {next_month.strftime('%B %Y').capitalize()} " | |
| f"par rapport à {mois_actuel}. " | |
| f"Recommandation : prévoir une allocation de capital supérieure." | |
| ) | |
| else: | |
| signal = 'MAINTENIR' | |
| message_signal = ( | |
| f"Le modèle anticipe une variation limitée de {evolution_pct:+.1f}% " | |
| f"en {next_month.strftime('%B %Y').capitalize()}. " | |
| f"Recommandation : maintenir le niveau d'investissement actuel." | |
| ) | |
| return { | |
| 'mois_predit': next_month.strftime('%B %Y').capitalize(), | |
| 'capital_predit': capital_predit, | |
| 'capital_reel_actuel': capital_reel_actuel, | |
| 'mois_actuel': mois_actuel, | |
| 'evolution_pct': evolution_pct, | |
| 'signal': signal, | |
| 'message_signal': message_signal, | |
| 'seuil_reduction': SEUIL_REDUCTION, | |
| 'seuil_augmentation': SEUIL_AUGMENTATION, | |
| } | |
| # ========================================================================= | |
| # INDICE DE CONFIANCE — NOUVEAU | |
| # ========================================================================= | |
| def compute_confidence_score(self, models): | |
| """ | |
| Calcule un indice de confiance (0–100%) pour la prédiction du mois suivant, | |
| basé sur l'écart moyen entre les données réelles et la courbe de régression. | |
| Méthode : | |
| - MAPE (Mean Absolute Percentage Error) sur tout l'historique → composante principale | |
| - CV (Coefficient de variation des résidus) → mesure la dispersion des écarts | |
| - R² du modèle → bonus/malus de fiabilité globale | |
| Score final : | |
| score = 100 | |
| − clamp(MAPE, 0, 60) * 1.0 # pénalité écart relatif moyen | |
| − clamp(CV, 0, 40) * 0.5 # pénalité instabilité des écarts | |
| + clamp(R², 0, 1) * 10 # bonus fiabilité modèle | |
| Returns: | |
| dict : | |
| - score : float (0–100) | |
| - mape : float MAPE en % | |
| - mae_moy : float écart absolu moyen en XOF | |
| - cv_residus : float coefficient de variation des résidus en % | |
| - r_squared : float | |
| - interpretation : str — texte lisible | |
| """ | |
| y_reel = self.df['Montant_Total_Sortie'].values.astype(float) | |
| y_pred = models['historical_predictions'].astype(float) | |
| residus = y_reel - y_pred | |
| # ── MAPE (on ignore les mois à zéro pour éviter division par zéro) | |
| mask = y_reel != 0 | |
| if mask.sum() > 0: | |
| mape = float(np.mean(np.abs(residus[mask] / y_reel[mask])) * 100) | |
| else: | |
| mape = 100.0 | |
| # ── MAE absolu moyen en XOF | |
| mae_moy = float(np.mean(np.abs(residus))) | |
| # ── Coefficient de variation des résidus | |
| # CV = écart-type des résidus / moyenne des valeurs réelles * 100 | |
| mean_reel = float(np.mean(np.abs(y_reel))) | |
| cv_residus = (float(np.std(residus)) / mean_reel * 100) if mean_reel > 0 else 100.0 | |
| r_squared = models['r_squared'] | |
| # ── Calcul du score (clampé entre 0 et 100) | |
| penalite_mape = min(mape, 60.0) * 1.0 | |
| penalite_cv = min(cv_residus, 40.0) * 0.5 | |
| bonus_r2 = min(max(r_squared, 0.0), 1.0) * 10.0 | |
| score = 100.0 - penalite_mape - penalite_cv + bonus_r2 | |
| score = max(0.0, min(100.0, score)) | |
| # ── Interprétation textuelle | |
| if score >= 70: | |
| interpretation = ( | |
| "Confiance ELEVEE — Les ecarts historiques entre le reel et le modele " | |
| "sont faibles et reguliers. La prediction du mois suivant est fiable." | |
| ) | |
| elif score >= 40: | |
| interpretation = ( | |
| "Confiance MODEREE — Des ecarts significatifs ont ete observes par le passe. " | |
| "La prediction est indicative ; une marge de manœuvre est recommandee." | |
| ) | |
| else: | |
| interpretation = ( | |
| "Confiance FAIBLE — Les ecarts entre le reel et le modele sont importants " | |
| "et/ou irreguliers. La prediction du mois suivant doit etre interpretee avec prudence." | |
| ) | |
| return { | |
| 'score': score, | |
| 'mape': mape, | |
| 'mae_moy': mae_moy, | |
| 'cv_residus': cv_residus, | |
| 'r_squared': r_squared, | |
| 'interpretation': interpretation, | |
| } | |
| # ========================================================================= | |
| # INTERPRÉTATION | |
| # ========================================================================= | |
| def get_interpretation(self, models): | |
| pente = models['pente'] | |
| r_squared = models['r_squared'] | |
| r_coefficient = models['r_coefficient'] | |
| mae = models['mae'] | |
| rmse = models['rmse'] | |
| pente_mensuelle = pente * 30.44 | |
| moyenne_flux = self.df['Montant_Total_Sortie'].mean() | |
| pct_variation = (abs(pente_mensuelle) / moyenne_flux * 100) if moyenne_flux > 0 else 0 | |
| tendance = 'CROISSANCE' if pente > 0 else 'DECROISSANCE' | |
| direction = 'a la hausse' if pente > 0 else 'a la baisse' | |
| impact = 'augmentation' if pente > 0 else 'diminution' | |
| if r_coefficient >= 0.8: | |
| force = 'tres forte' | |
| elif r_coefficient >= 0.6: | |
| force = 'forte' | |
| elif r_coefficient >= 0.4: | |
| force = 'moderee' | |
| else: | |
| force = 'faible' | |
| if r_squared >= 0.7: | |
| qualite = 'excellent' | |
| elif r_squared >= 0.5: | |
| qualite = 'bon' | |
| elif r_squared >= 0.3: | |
| qualite = 'acceptable' | |
| else: | |
| qualite = 'faible' | |
| return { | |
| 'tendance': tendance, | |
| 'direction': direction, | |
| 'impact': impact, | |
| 'pente_mensuelle': pente_mensuelle, | |
| 'force_correlation': force, | |
| 'qualite_modele': qualite, | |
| 'pct_variation': pct_variation, | |
| 'message_principal': ( | |
| f"Les flux de sortie montrent une tendance {direction} avec une {impact} " | |
| f"moyenne de {abs(pente_mensuelle):,.0f} XOF par mois ({pct_variation:.1f}% du flux moyen)." | |
| ), | |
| 'message_fiabilite': ( | |
| f"Qualite du modele : {qualite}. R² = {r_squared:.2%}, " | |
| f"correlation {force} (R = {r_coefficient:.3f})." | |
| ), | |
| 'message_precision': ( | |
| f"Erreur moyenne (MAE) : {mae:,.0f} XOF. " | |
| f"Erreur quadratique (RMSE) : {rmse:,.0f} XOF." | |
| ), | |
| } | |
| # ========================================================================= | |
| # DONNÉES DE VISUALISATION | |
| # ========================================================================= | |
| def generate_visualization_data(self, predictions_result): | |
| if 'error' in predictions_result: | |
| return predictions_result | |
| predictions_df = predictions_result['predictions'] | |
| models = predictions_result['models'] | |
| residuals = self.df['Montant_Total_Sortie'].values - models['historical_predictions'] | |
| return { | |
| 'historical': { | |
| 'dates': self.df['Date'].tolist(), | |
| 'values': self.df['Montant_Total_Sortie'].tolist(), | |
| 'predictions_hist': models['historical_predictions'].tolist(), | |
| }, | |
| 'future': { | |
| 'dates': predictions_df['Date'].tolist(), | |
| 'predictions': predictions_df['Montant_Predit'].tolist(), | |
| 'lower_bound': predictions_df['Borne_Inf'].tolist(), | |
| 'upper_bound': predictions_df['Borne_Sup'].tolist(), | |
| }, | |
| 'residuals': { | |
| 'values': residuals.tolist(), | |
| 'predictions': models['historical_predictions'].tolist(), | |
| }, | |
| 'interpretations': self.get_interpretation(models), | |
| 'metrics': { | |
| 'mae': models['mae'], | |
| 'rmse': models['rmse'], | |
| 'r_squared': models['r_squared'], | |
| 'r_coefficient': models['r_coefficient'], | |
| 'pente': models['pente'], | |
| }, | |
| } |