Prévisions de prix (`kadi.market.forecasting`)¶

La classe Forecasting prédit le prix futur d'une culture à partir de l'historique de prix. Le modèle est une régression linéaire enrichie de features saisonnières (harmoniques de Fourier), entraînée à la demande sur les données fournies. Si l'historique est insuffisant (moins de 20 observations), un fallback simulé clairement signalé est retourné.

Importation directe¶

from kadi.market import Forecasting

Initialisation¶

Forecasting est créée automatiquement par Market et accessible via market.forecast. Pour une instanciation directe :

from kadi.market import Forecasting

forecast = Forecasting()

Le module n'a pas d'état persistant : le modèle est entraîné à chaque appel de predict() sur les données fournies.

Méthodes¶

`predict(crop, market, ahead, ci, hist)`¶

Prédit le prix futur d'une culture sur un marché donné.

Pipeline interne :

Validation de l'historique (minimum 20 points). Si insuffisant, bascule sur le fallback simulé.
Construction des features : tendance linéaire + 2 harmoniques de Fourier (annuelle et semi-annuelle sur 365/182.5 jours).
Entraînement d'une régression linéaire sur l'historique complet.
Calcul du RMSE par validation croisée temporelle (TimeSeriesSplit, 3 folds).
Calcul de l'intervalle de prévision basé sur le RMSE réel, avec croissance en racine carrée du temps.

import kadi as kd

marche = kd.Market(lat=9.3, lon=2.3, location="Parakou")

# Récupération de l'historique
df = marche.pricing.fetch("maize", "parakou", days=365)

# Prévision à 30 jours avec intervalle de confiance 90%
prev = marche.forecast.predict(
    crop="maize",
    market="parakou",
    ahead=30,
    ci=0.9,
    hist=df,
)

print(f"Prix prédit  : {prev['predicted_price']:.2f} XOF/kg")
print(f"Intervalle   : [{prev['low_90']:.2f}, {prev['high_90']:.2f}]")
print(f"RMSE         : {prev['rmse']} XOF/kg")
print(f"Modèle       : {prev['model_used']}")
print(f"Historique   : {prev['nb_history_pts']} points")
print(f"Simulé       : {prev['is_simulated']}")
print(f"Confiance    : {prev['confidence_score']:.2f}")

Via la façade Market (pipeline complet en un seul appel) :

# Market.predict() gère automatiquement la récupération de l'historique
prev = marche.predict("maize", ahead=30, confidence_interval=0.9, days=365)

print(f"Prix prédit dans 30j : {prev['predicted_price']} XOF/kg")
print(f"Horizon utilisé      : {prev['ahead']} jours")

Paramètres :

Nom	Type	Défaut	Description
`crop`	`str`	requis	Code de la culture (ex: `'maize'`)
`market`	`str`	requis	Nom normalisé du marché (ex: `'cotonou'`)
`ahead`	`int`	`7`	Horizon de prévision en jours
`ci`	`float`	`0.9`	Niveau de confiance : `0.9` (90%) ou `0.95` (95%)
`hist`	`pd.DataFrame`	`None`	Historique avec colonnes `'date'` et `'price'`

Retour : dict

Clé	Type	Description
`predicted_price`	`float`	Prix predit en XOF/kg
`low_90`	`float`	Borne inferieure de l'intervalle de confiance
`high_90`	`float`	Borne superieure de l'intervalle de confiance
`confidence`	`float`	Niveau de confiance utilise (0.9 ou 0.95)
`model_used`	`str`	`'linear_regression_fourier'` ou `'fallback_simule'`
`rmse`	`float`	RMSE en XOF/kg (None si fallback simule)
`is_simulated`	`bool`	True si les donnees source sont simulees
`confidence_score`	`float`	Score de fiabilite (0.0 a 1.0)
`nb_history_pts`	`int`	Nombre de points d'historique utilises
`days_ahead`	`int`	Horizon de prevision utilise

Score de confiance :

Le score de confiance est calculé selon :

score_source = 0.10 si is_simulated, 0.85 sinon
facteur_volume = min(1.0, nb_pts / 100.0)
confidence_score = score_source * (0.5 + 0.5 * facteur_volume)

Un score de 0.85 est le maximum atteignable avec des données réelles. Un score de 0.0 indique un fallback entièrement simulé.

Fallback simulé :

Activé si hist est absent ou contient moins de 20 observations valides. Retourne predicted_price=300.0 XOF/kg avec is_simulated=True et confidence_score=0.0. Ne pas utiliser pour des décisions commerciales.

Méthode dépréciée¶

Ancienne méthode	Remplacée par	Depuis
`predict_price(crop, market, days_ahead, historique)`	`predict(crop, market, ahead, hist)`	v1.2.0

Classe dépréciée¶

Ancien nom	Remplacé par	Depuis
`MarketForecasting`	`Forecasting`	v1.2.0

Forecasting ¶

Classe gérant la prévision de prix agricoles à partir de l'historique réel.

Le modèle principal est une régression linéaire enrichie de features saisonnières (harmoniques de Fourier). Il est entraîné à chaque appel de predict() sur l'historique fourni, sans état persistant entre les appels. Cela garantit que les prévisions sont toujours basées sur les données les plus récentes disponibles.

Quand l'historique est insuffisant (< 20 points) ou absent, la méthode bascule sur un fallback simulé clairement signalé par is_simulated=True.

Source code in kadi/market/forecasting.py

class Forecasting:
    """
    Classe gérant la prévision de prix agricoles à partir de l'historique réel.

    Le modèle principal est une régression linéaire enrichie de features
    saisonnières (harmoniques de Fourier). Il est entraîné à chaque appel
    de predict() sur l'historique fourni, sans état persistant entre
    les appels. Cela garantit que les prévisions sont toujours basées sur
    les données les plus récentes disponibles.

    Quand l'historique est insuffisant (< 20 points) ou absent, la méthode
    bascule sur un fallback simulé clairement signalé par is_simulated=True.
    """

    def __init__(self):
        """
        Initialise le module de prévision.

        Aucun état persistant n'est conservé entre les appels : le modèle
        est entraîné à la demande dans predict().
        """
        # Modèle de régression linéaire (sans état entre les appels)
        self._model = LinearRegression()

    # ------------------------------------------------------------------
    # Rétrocompatibilité : méthodes publiques renommées
    # ------------------------------------------------------------------


    def __getattr__(self, name: str):
        """Intercepte les accès aux anciens noms de méthodes publiques.

        Délègue vers le nouveau nom et émet un DeprecationWarning.

        Args:
            name (str): Nom de l'attribut ou méthode demandé.

        Returns:
            callable: La méthode correspondante sous son nouveau nom.

        Raises:
            AttributeError: Si le nom n'est ni nouveau ni ancien.
        """
        # Vérification dans la table de rétrocompatibilité
        if name in _DEPRECATED_METHODS:
            new_name = _DEPRECATED_METHODS[name]
            warnings.warn(
                f"Forecasting.{name}() est obsolète et sera supprimé dans "
                f"KadiPy v2.0. Utilisez Forecasting.{new_name}() à la place.",
                DeprecationWarning,
                stacklevel=2,
            )
            return getattr(self, new_name)
        raise AttributeError(f"'Forecasting' n'a pas d'attribut '{name}'.")

    # ------------------------------------------------------------------
    # Méthodes privées : préparation des données et entraînement
    # ------------------------------------------------------------------

    def _build_features(self, series: np.ndarray) -> np.ndarray:
        """
        Construit la matrice de features temporelles pour la régression.

        Les features générées sont :
            - t          : indice séquentiel (0, 1, 2, …) - capture la tendance
            - sin_365    : harmonique annuelle (sin) - pic de saison des récoltes
            - cos_365    : harmonique annuelle (cos)
            - sin_182    : harmonique semi-annuelle (sin) - deuxième cycle de culture
            - cos_182    : harmonique semi-annuelle (cos)

        Toutes les features sont construites à partir de l'indice temporel
        pour être extrapolables vers le futur (pas besoin de prix futurs).

        Args:
            series (np.ndarray): Vecteur d'indices temporels (entiers).

        Returns:
            np.ndarray: Matrice de shape (n, 5) contenant les 5 features.
        """
        # Indice temporel normalisé vers [0, 2*pi] sur une année (365 jours)
        t = series.astype(float)
        angle_annuel = 2.0 * np.pi * t / 365.0
        angle_semi_annuel = 2.0 * np.pi * t / 182.5

        # Empilement des features en colonnes
        return np.column_stack([
            t,                          # Tendance linéaire
            np.sin(angle_annuel),       # Saisonnalité annuelle (sinus)
            np.cos(angle_annuel),       # Saisonnalité annuelle (cosinus)
            np.sin(angle_semi_annuel),  # Saisonnalité semi-annuelle (sinus)
            np.cos(angle_semi_annuel),  # Saisonnalité semi-annuelle (cosinus)
        ])

    def _cv_rmse(
        self, X: np.ndarray, y: np.ndarray
    ) -> float:
        """
        Calcule le RMSE moyen par validation croisée temporelle.

        Utilise TimeSeriesSplit pour respecter l'ordre chronologique :
        chaque fold d'entraînement ne contient que des données antérieures
        au fold de test, ce qui évite le data leakage.

        Args:
            X (np.ndarray): Matrice de features de shape (n, p).
            y (np.ndarray): Vecteur de prix cibles de shape (n,).

        Returns:
            float: RMSE moyen sur les folds de test. Retourne NaN si le
                nombre de points est insuffisant pour la validation croisée.
        """
        # Vérification du nombre minimal de points pour la cross-validation
        nb_points = len(y)
        if nb_points < _CV_N_SPLITS * 2:
            logger.debug(
                f"Trop peu de points ({nb_points}) pour la validation croisée "
                f"({_CV_N_SPLITS} folds). RMSE non calculé."
            )
            return float("nan")

        # Validation croisée avec préservation de l'ordre chronologique
        tscv = TimeSeriesSplit(n_splits=_CV_N_SPLITS)
        erreurs = []

        for train_idx, test_idx in tscv.split(X):
            X_train, X_test = X[train_idx], X[test_idx]
            y_train, y_test = y[train_idx], y[test_idx]

            modele_fold = LinearRegression()
            modele_fold.fit(X_train, y_train)
            y_pred = modele_fold.predict(X_test)

            # Calcul du RMSE sur ce fold de test
            rmse_fold = np.sqrt(mean_squared_error(y_test, y_pred))
            erreurs.append(rmse_fold)

        return float(np.mean(erreurs))

    def _clean_df(self, hist: pd.DataFrame) -> pd.DataFrame:
        """
        Nettoie et prépare le DataFrame historique pour l'entraînement.

        Opérations effectuées :
            1. Sélection des colonnes nécessaires
            2. Tri chronologique
            3. Suppression des valeurs manquantes
            4. Suppression des prix non positifs (protection contre les anomalies)

        Args:
            hist (pd.DataFrame): DataFrame brut avec au moins 'date' et 'price'.

        Returns:
            pd.DataFrame: DataFrame nettoyé, trié par date, sans valeurs aberrantes.
                Peut être vide si toutes les lignes sont filtrées.
        """
        if "date" not in hist.columns or "price" not in hist.columns:
            logger.warning(
                "L'historique fourni doit contenir les colonnes 'date' et 'price'."
            )
            return pd.DataFrame()

        # Copie pour ne pas modifier le DataFrame d'entrée
        df = hist[["date", "price"]].copy()

        # Conversion de la colonne date en datetime si nécessaire
        df["date"] = pd.to_datetime(df["date"], errors="coerce")

        # Suppression des lignes avec date ou prix manquants
        df = df.dropna(subset=["date", "price"])

        # Tri chronologique (indispensable pour la cross-validation temporelle)
        df = df.sort_values("date").reset_index(drop=True)

        # Suppression des prix non positifs (prix nuls ou négatifs sont des anomalies)
        df = df[df["price"] > 0].reset_index(drop=True)

        return df

    def _fallback(
        self, ahead: int, ci: float
    ) -> dict:
        """
        Génère une prévision de repli simulée quand l'historique est insuffisant.

        Cette méthode est clairement signalée par is_simulated=True et
        confidence_score=0.0. Elle ne doit jamais être présentée comme une
        prévision fiable.

        Args:
            ahead (int): Horizon de prévision en jours.
            ci (float): Niveau de confiance demandé (0.9 ou 0.95).

        Returns:
            dict: Prévision simulée avec tous les champs standards.
        """
        # Prix de référence béninois (utilisé uniquement en fallback simulé)
        prix_reference = 300.0

        # Volatilité simulée croissante avec le temps (GARCH-like)
        volatilite = prix_reference * 0.05 * np.sqrt(ahead)

        # Facteur z-score pour l'intervalle de confiance
        z_score = 1.645 if ci == 0.9 else 1.96
        marge = volatilite * z_score

        logger.warning(
            "predict() : historique insuffisant ou absent. "
            "Prévision simulée retournée (is_simulated=True, confidence_score=0.0). "
            "Cette prévision ne doit pas être utilisée pour des décisions commerciales."
        )

        return {
            "predicted_price": round(prix_reference, 2),
            "low_90": round(max(0.0, prix_reference - marge), 2),
            "high_90": round(prix_reference + marge, 2),
            "confidence": ci,
            "model_used": "fallback_simule",
            "rmse": None,
            "is_simulated": True,
            "confidence_score": 0.0,
            "nb_history_pts": 0,
            "days_ahead": ahead,
        }

    # ------------------------------------------------------------------
    # Méthode publique principale
    # ------------------------------------------------------------------

    def predict(
        self,
        crop: str,
        market: str,
        ahead: int = 7,
        ci: float = 0.9,
        hist: pd.DataFrame = None,
        **kwargs,
    ) -> dict:
        """
        Prédit le prix futur d'une culture sur un marché donné.

        La méthode utilise une régression linéaire avec features temporelles
        et saisonnières (harmoniques de Fourier), entraînée sur l'historique
        fourni en paramètre. Le RMSE est calculé par validation croisée
        temporelle (TimeSeriesSplit).

        Si l'historique est absent ou contient moins de 20 observations
        valides, la méthode bascule sur un fallback simulé et signale
        clairement is_simulated=True avec confidence_score=0.0.

        Args:
            crop (str): Code de la culture (ex: 'maize', 'rice').
            market (str): Nom normalisé du marché (ex: 'cotonou').
            ahead (int, optional): Horizon de prévision en jours.
                Défaut : 7. La précision décroît avec l'horizon.
            ci (float, optional): Niveau de confiance pour l'intervalle de
                prévision. Valeurs supportées : 0.9 (90%) et 0.95 (95%).
                Défaut : 0.9.
            hist (pd.DataFrame, optional): DataFrame avec au moins les colonnes
                'date' (datetime) et 'price' (float en XOF/kg). Typiquement
                fourni par Pricing.fetch(). Si None ou insuffisant, le
                fallback simulé est activé.
            **kwargs: Anciens arguments (days_ahead, historique) pour rétrocompatibilité.

        Returns:
            dict: Dictionnaire contenant :
                - 'predicted_price' (float) : prix prédit en XOF/kg
                - 'low_90' (float)          : borne inférieure de l'intervalle
                - 'high_90' (float)         : borne supérieure de l'intervalle
                - 'confidence' (float)      : niveau de confiance (0.9 ou 0.95)
                - 'model_used' (str)        : identifiant du modèle utilisé
                - 'rmse' (float ou None)    : RMSE réel en XOF/kg (None si simulé)
                - 'is_simulated' (bool)     : True si les données source sont simulées
                - 'confidence_score' (float): score de fiabilité de la prévision (0-1)
                - 'nb_history_pts' (int)    : nombre de points d'historique utilisés
                - 'days_ahead' (int)        : horizon de prévision utilisé
        """
        # Prise en charge des anciens noms de paramètres pour la rétrocompatibilité
        if "days_ahead" in kwargs:
            ahead = kwargs.pop("days_ahead")
        if "historique" in kwargs:
            hist = kwargs.pop("historique")

        # ------------------------------------------------------------------
        # Étape 1 : vérification de la disponibilité de l'historique
        # ------------------------------------------------------------------
        if hist is None or hist.empty:
            logger.info(
                f"predict({crop}/{market}) : aucun historique fourni. "
                "Fallback simulé activé."
            )
            return self._fallback(ahead, ci)

        # Nettoyage et préparation du DataFrame
        df_propre = self._clean_df(hist)

        # Vérification du seuil minimum d'observations pour le modèle
        if len(df_propre) < _MIN_HISTORY_POINTS:
            logger.info(
                f"predict({crop}/{market}) : historique insuffisant "
                f"({len(df_propre)} points < {_MIN_HISTORY_POINTS} requis). "
                "Fallback simulé activé."
            )
            return self._fallback(ahead, ci)

        # ------------------------------------------------------------------
        # Étape 2 : construction des features et entraînement
        # ------------------------------------------------------------------

        # Création des indices temporels séquentiels pour les données d'entraînement
        nb_pts = len(df_propre)
        indices_train = np.arange(nb_pts)

        # Construction de la matrice de features et du vecteur cible
        X_train = self._build_features(indices_train)
        y_train = df_propre["price"].values

        # Entraînement du modèle sur la totalité de l'historique disponible
        self._model.fit(X_train, y_train)

        # ------------------------------------------------------------------
        # Étape 3 : prédiction pour l'horizon demandé
        # ------------------------------------------------------------------

        # L'indice du point futur est la continuation séquentielle de l'historique
        indice_futur = np.array([nb_pts - 1 + ahead])
        X_futur = self._build_features(indice_futur)

        # Prix prédit par le modèle entraîné
        prix_predit = float(self._model.predict(X_futur)[0])

        # Protection contre les prix négatifs (peut arriver en fin de tendance)
        prix_predit = max(0.0, prix_predit)

        # ------------------------------------------------------------------
        # Étape 4 : calcul du RMSE par validation croisée temporelle
        # ------------------------------------------------------------------
        rmse = self._cv_rmse(X_train, y_train)

        # ------------------------------------------------------------------
        # Étape 5 : calcul de l'intervalle de prévision
        # ------------------------------------------------------------------

        # Facteur z-score pour l'intervalle de confiance demandé
        z_score = 1.645 if ci == 0.9 else 1.96

        # Marge basée sur le RMSE réel si disponible, sinon 5% du prix prédit
        if not np.isnan(rmse):
            # L'incertitude croît avec l'horizon (racine carrée du temps)
            marge = z_score * rmse * np.sqrt(ahead / 7.0)
        else:
            # Fallback de marge si la cross-validation a échoué
            marge = prix_predit * 0.05 * np.sqrt(ahead)

        borne_basse = round(max(0.0, prix_predit - marge), 2)
        borne_haute = round(prix_predit + marge, 2)

        # ------------------------------------------------------------------
        # Étape 6 : propagation du flag is_simulated depuis la source
        # ------------------------------------------------------------------

        # Si l'historique est partiellement simulé, le signal se propage
        est_simule = False
        if "is_simulated" in hist.columns:
            est_simule = bool(hist["is_simulated"].any())

        # Calcul du score de confiance basé sur la source et la taille de l'historique
        # Formule : score_source * facteur_volume_données
        score_source = 0.1 if est_simule else 0.85
        facteur_volume = min(1.0, nb_pts / 100.0)
        score_confiance = round(score_source * (0.5 + 0.5 * facteur_volume), 3)

        # ------------------------------------------------------------------
        # Étape 7 : construction du dictionnaire de retour
        # ------------------------------------------------------------------
        rmse_arrondi = round(rmse, 2) if not np.isnan(rmse) else None

        resultat = {
            "predicted_price": round(prix_predit, 2),
            "low_90": borne_basse,
            "high_90": borne_haute,
            "confidence": ci,
            "model_used": "linear_regression_fourier",
            "rmse": rmse_arrondi,
            "is_simulated": est_simule,
            "confidence_score": score_confiance,
            "nb_history_pts": nb_pts,
            "days_ahead": ahead,
        }

        logger.info(
            f"Prévision {crop}/{market} à {ahead}j : "
            f"{prix_predit:.0f} XOF/kg "
            f"[{borne_basse:.0f} - {borne_haute:.0f}], "
            f"RMSE={rmse_arrondi}, is_simulated={est_simule}, "
            f"nb_pts={nb_pts}"
        )

        return resultat

init ¶

__init__()

Initialise le module de prévision.

Aucun état persistant n'est conservé entre les appels : le modèle est entraîné à la demande dans predict().

Source code in kadi/market/forecasting.py

def __init__(self):
    """
    Initialise le module de prévision.

    Aucun état persistant n'est conservé entre les appels : le modèle
    est entraîné à la demande dans predict().
    """
    # Modèle de régression linéaire (sans état entre les appels)
    self._model = LinearRegression()

getattr ¶

__getattr__(name: str)

Intercepte les accès aux anciens noms de méthodes publiques.

Délègue vers le nouveau nom et émet un DeprecationWarning.

Parameters:

Name	Type	Description	Default
`name`	`str`	Nom de l'attribut ou méthode demandé.	required

Returns:

Name	Type	Description
`callable`		La méthode correspondante sous son nouveau nom.

Raises:

Type	Description
`AttributeError`	Si le nom n'est ni nouveau ni ancien.

Source code in kadi/market/forecasting.py

def __getattr__(self, name: str):
    """Intercepte les accès aux anciens noms de méthodes publiques.

    Délègue vers le nouveau nom et émet un DeprecationWarning.

    Args:
        name (str): Nom de l'attribut ou méthode demandé.

    Returns:
        callable: La méthode correspondante sous son nouveau nom.

    Raises:
        AttributeError: Si le nom n'est ni nouveau ni ancien.
    """
    # Vérification dans la table de rétrocompatibilité
    if name in _DEPRECATED_METHODS:
        new_name = _DEPRECATED_METHODS[name]
        warnings.warn(
            f"Forecasting.{name}() est obsolète et sera supprimé dans "
            f"KadiPy v2.0. Utilisez Forecasting.{new_name}() à la place.",
            DeprecationWarning,
            stacklevel=2,
        )
        return getattr(self, new_name)
    raise AttributeError(f"'Forecasting' n'a pas d'attribut '{name}'.")

predict ¶

predict(crop: str, market: str, ahead: int = 7, ci: float = 0.9, hist: DataFrame = None, **kwargs) -> dict

Prédit le prix futur d'une culture sur un marché donné.

La méthode utilise une régression linéaire avec features temporelles et saisonnières (harmoniques de Fourier), entraînée sur l'historique fourni en paramètre. Le RMSE est calculé par validation croisée temporelle (TimeSeriesSplit).

Si l'historique est absent ou contient moins de 20 observations valides, la méthode bascule sur un fallback simulé et signale clairement is_simulated=True avec confidence_score=0.0.

Parameters:

Name	Type	Description	Default
`crop`	`str`	Code de la culture (ex: 'maize', 'rice').	required
`market`	`str`	Nom normalisé du marché (ex: 'cotonou').	required
`ahead`	`int`	Horizon de prévision en jours. Défaut : 7. La précision décroît avec l'horizon.	`7`
`ci`	`float`	Niveau de confiance pour l'intervalle de prévision. Valeurs supportées : 0.9 (90%) et 0.95 (95%). Défaut : 0.9.	`0.9`
`hist`	`DataFrame`	DataFrame avec au moins les colonnes 'date' (datetime) et 'price' (float en XOF/kg). Typiquement fourni par Pricing.fetch(). Si None ou insuffisant, le fallback simulé est activé.	`None`
`**kwargs`		Anciens arguments (days_ahead, historique) pour rétrocompatibilité.	`{}`

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant : - 'predicted_price' (float) : prix prédit en XOF/kg - 'low_90' (float) : borne inférieure de l'intervalle - 'high_90' (float) : borne supérieure de l'intervalle - 'confidence' (float) : niveau de confiance (0.9 ou 0.95) - 'model_used' (str) : identifiant du modèle utilisé - 'rmse' (float ou None) : RMSE réel en XOF/kg (None si simulé) - 'is_simulated' (bool) : True si les données source sont simulées - 'confidence_score' (float): score de fiabilité de la prévision (0-1) - 'nb_history_pts' (int) : nombre de points d'historique utilisés - 'days_ahead' (int) : horizon de prévision utilisé

Source code in kadi/market/forecasting.py

def predict(
    self,
    crop: str,
    market: str,
    ahead: int = 7,
    ci: float = 0.9,
    hist: pd.DataFrame = None,
    **kwargs,
) -> dict:
    """
    Prédit le prix futur d'une culture sur un marché donné.

    La méthode utilise une régression linéaire avec features temporelles
    et saisonnières (harmoniques de Fourier), entraînée sur l'historique
    fourni en paramètre. Le RMSE est calculé par validation croisée
    temporelle (TimeSeriesSplit).

    Si l'historique est absent ou contient moins de 20 observations
    valides, la méthode bascule sur un fallback simulé et signale
    clairement is_simulated=True avec confidence_score=0.0.

    Args:
        crop (str): Code de la culture (ex: 'maize', 'rice').
        market (str): Nom normalisé du marché (ex: 'cotonou').
        ahead (int, optional): Horizon de prévision en jours.
            Défaut : 7. La précision décroît avec l'horizon.
        ci (float, optional): Niveau de confiance pour l'intervalle de
            prévision. Valeurs supportées : 0.9 (90%) et 0.95 (95%).
            Défaut : 0.9.
        hist (pd.DataFrame, optional): DataFrame avec au moins les colonnes
            'date' (datetime) et 'price' (float en XOF/kg). Typiquement
            fourni par Pricing.fetch(). Si None ou insuffisant, le
            fallback simulé est activé.
        **kwargs: Anciens arguments (days_ahead, historique) pour rétrocompatibilité.

    Returns:
        dict: Dictionnaire contenant :
            - 'predicted_price' (float) : prix prédit en XOF/kg
            - 'low_90' (float)          : borne inférieure de l'intervalle
            - 'high_90' (float)         : borne supérieure de l'intervalle
            - 'confidence' (float)      : niveau de confiance (0.9 ou 0.95)
            - 'model_used' (str)        : identifiant du modèle utilisé
            - 'rmse' (float ou None)    : RMSE réel en XOF/kg (None si simulé)
            - 'is_simulated' (bool)     : True si les données source sont simulées
            - 'confidence_score' (float): score de fiabilité de la prévision (0-1)
            - 'nb_history_pts' (int)    : nombre de points d'historique utilisés
            - 'days_ahead' (int)        : horizon de prévision utilisé
    """
    # Prise en charge des anciens noms de paramètres pour la rétrocompatibilité
    if "days_ahead" in kwargs:
        ahead = kwargs.pop("days_ahead")
    if "historique" in kwargs:
        hist = kwargs.pop("historique")

    # ------------------------------------------------------------------
    # Étape 1 : vérification de la disponibilité de l'historique
    # ------------------------------------------------------------------
    if hist is None or hist.empty:
        logger.info(
            f"predict({crop}/{market}) : aucun historique fourni. "
            "Fallback simulé activé."
        )
        return self._fallback(ahead, ci)

    # Nettoyage et préparation du DataFrame
    df_propre = self._clean_df(hist)

    # Vérification du seuil minimum d'observations pour le modèle
    if len(df_propre) < _MIN_HISTORY_POINTS:
        logger.info(
            f"predict({crop}/{market}) : historique insuffisant "
            f"({len(df_propre)} points < {_MIN_HISTORY_POINTS} requis). "
            "Fallback simulé activé."
        )
        return self._fallback(ahead, ci)

    # ------------------------------------------------------------------
    # Étape 2 : construction des features et entraînement
    # ------------------------------------------------------------------

    # Création des indices temporels séquentiels pour les données d'entraînement
    nb_pts = len(df_propre)
    indices_train = np.arange(nb_pts)

    # Construction de la matrice de features et du vecteur cible
    X_train = self._build_features(indices_train)
    y_train = df_propre["price"].values

    # Entraînement du modèle sur la totalité de l'historique disponible
    self._model.fit(X_train, y_train)

    # ------------------------------------------------------------------
    # Étape 3 : prédiction pour l'horizon demandé
    # ------------------------------------------------------------------

    # L'indice du point futur est la continuation séquentielle de l'historique
    indice_futur = np.array([nb_pts - 1 + ahead])
    X_futur = self._build_features(indice_futur)

    # Prix prédit par le modèle entraîné
    prix_predit = float(self._model.predict(X_futur)[0])

    # Protection contre les prix négatifs (peut arriver en fin de tendance)
    prix_predit = max(0.0, prix_predit)

    # ------------------------------------------------------------------
    # Étape 4 : calcul du RMSE par validation croisée temporelle
    # ------------------------------------------------------------------
    rmse = self._cv_rmse(X_train, y_train)

    # ------------------------------------------------------------------
    # Étape 5 : calcul de l'intervalle de prévision
    # ------------------------------------------------------------------

    # Facteur z-score pour l'intervalle de confiance demandé
    z_score = 1.645 if ci == 0.9 else 1.96

    # Marge basée sur le RMSE réel si disponible, sinon 5% du prix prédit
    if not np.isnan(rmse):
        # L'incertitude croît avec l'horizon (racine carrée du temps)
        marge = z_score * rmse * np.sqrt(ahead / 7.0)
    else:
        # Fallback de marge si la cross-validation a échoué
        marge = prix_predit * 0.05 * np.sqrt(ahead)

    borne_basse = round(max(0.0, prix_predit - marge), 2)
    borne_haute = round(prix_predit + marge, 2)

    # ------------------------------------------------------------------
    # Étape 6 : propagation du flag is_simulated depuis la source
    # ------------------------------------------------------------------

    # Si l'historique est partiellement simulé, le signal se propage
    est_simule = False
    if "is_simulated" in hist.columns:
        est_simule = bool(hist["is_simulated"].any())

    # Calcul du score de confiance basé sur la source et la taille de l'historique
    # Formule : score_source * facteur_volume_données
    score_source = 0.1 if est_simule else 0.85
    facteur_volume = min(1.0, nb_pts / 100.0)
    score_confiance = round(score_source * (0.5 + 0.5 * facteur_volume), 3)

    # ------------------------------------------------------------------
    # Étape 7 : construction du dictionnaire de retour
    # ------------------------------------------------------------------
    rmse_arrondi = round(rmse, 2) if not np.isnan(rmse) else None

    resultat = {
        "predicted_price": round(prix_predit, 2),
        "low_90": borne_basse,
        "high_90": borne_haute,
        "confidence": ci,
        "model_used": "linear_regression_fourier",
        "rmse": rmse_arrondi,
        "is_simulated": est_simule,
        "confidence_score": score_confiance,
        "nb_history_pts": nb_pts,
        "days_ahead": ahead,
    }

    logger.info(
        f"Prévision {crop}/{market} à {ahead}j : "
        f"{prix_predit:.0f} XOF/kg "
        f"[{borne_basse:.0f} - {borne_haute:.0f}], "
        f"RMSE={rmse_arrondi}, is_simulated={est_simule}, "
        f"nb_pts={nb_pts}"
    )

    return resultat

Prévisions de prix (kadi.market.forecasting)¶

Importation directe¶

Initialisation¶

Méthodes¶

predict(crop, market, ahead, ci, hist)¶

Méthode dépréciée¶

Classe dépréciée¶

Forecasting ¶

__init__ ¶

__getattr__ ¶

predict ¶

Prévisions de prix (`kadi.market.forecasting`)¶

`predict(crop, market, ahead, ci, hist)`¶

init ¶

getattr ¶