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'], }, }