Blame for src/Analytics/VortexOutFlux.py

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