Price Forecasting (`kadi.market.forecasting`)¶

The Forecasting class predicts future crop prices based on historical price data. The model uses linear regression enriched with seasonal features (Fourier harmonics), trained on demand using the provided historical dataset. If historical data is insufficient (fewer than 20 observations), a clearly marked simulated fallback is returned.

Direct Importing¶

from kadi.market import Forecasting

Initialization¶

Forecasting is created automatically by Market and is accessible via market.forecast. For direct instantiation:

from kadi.market import Forecasting

forecast = Forecasting()

The module has no persistent state: the model is trained on each call to predict() using the provided data.

Methods¶

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

Predicts future price for a crop at a given market location.

Internal Pipeline:

History validation (minimum 20 observations required). If insufficient, switches to simulated fallback.
Feature engineering: linear trend + 2 Fourier harmonics (annual and semi-annual cycles over 365 / 182.5 days).
Linear regression training on complete history.
RMSE calculation via time series cross-validation (TimeSeriesSplit, 3 folds).
Prediction interval calculation based on real RMSE, expanding with square root of time horizon.

import kadi as kd

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

# Fetching price history
df = market.pricing.fetch("maize", "parakou", days=365)

# 30-day forecast with 90% confidence interval
prediction = market.forecast.predict(
    crop="maize",
    market="parakou",
    ahead=30,
    ci=0.9,
    hist=df,
)

print(f"Predicted price : {prediction['predicted_price']:.2f} XOF/kg")
print(f"Interval        : [{prediction['low_90']:.2f}, {prediction['high_90']:.2f}]")
print(f"RMSE            : {prediction['rmse']} XOF/kg")
print(f"Model           : {prediction['model_used']}")
print(f"History points  : {prediction['nb_history_pts']}")
print(f"Simulated       : {prediction['is_simulated']}")
print(f"Confidence      : {prediction['confidence_score']:.2f}")

Via the Market facade (full pipeline in a single call):

# Market.predict() handles fetching history automatically
prediction = market.predict("maize", ahead=30, confidence_interval=0.9, days=365)

print(f"Predicted price in 30d : {prediction['predicted_price']} XOF/kg")
print(f"Horizon used           : {prediction['ahead']} days")

Parameters:

Name	Type	Default	Description
`crop`	`str`	required	Crop code (e.g. `'maize'`)
`market`	`str`	required	Normalized market name (e.g. `'cotonou'`)
`ahead`	`int`	`7`	Forecast horizon in days
`ci`	`float`	`0.9`	Confidence level: `0.9` (90%) or `0.95` (95%)
`hist`	`pd.DataFrame`	`None`	History DataFrame with `'date'` and `'price'` columns

Returns: dict

Key	Type	Description
`predicted_price`	`float`	Predicted price in XOF/kg
`low_90`	`float`	Lower bound of confidence interval
`high_90`	`float`	Upper bound of confidence interval
`confidence`	`float`	Confidence level used (0.9 or 0.95)
`model_used`	`str`	`'linear_regression_fourier'` or `'fallback_simule'`
`rmse`	`float`	RMSE in XOF/kg (None if simulated fallback)
`is_simulated`	`bool`	True if source data is simulated
`confidence_score`	`float`	Reliability score (0.0 to 1.0)
`nb_history_pts`	`int`	Number of historical data points used
`days_ahead`	`int`	Forecast horizon used

Confidence Score:

The confidence score is calculated as follows:

source_score = 0.10 if is_simulated else 0.85
volume_factor = min(1.0, nb_pts / 100.0)
confidence_score = source_score * (0.5 + 0.5 * volume_factor)

A score of 0.85 is the maximum achievable with real data. A score of 0.0 indicates a fully simulated fallback.

Simulated Fallback:

Activated if hist is missing or contains fewer than 20 valid observations. Returns predicted_price=300.0 XOF/kg with is_simulated=True and confidence_score=0.0. Must not be used for actual commercial decisions.

Deprecated Method¶

Legacy Method	Replaced By	Since
`predict_price(crop, market, days_ahead, historique)`	`predict(crop, market, ahead, hist)`	v1.2.0

Deprecated Class¶

Legacy Class	Replaced By	Since
`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

Price Forecasting (kadi.market.forecasting)¶

Direct Importing¶

Initialization¶

Methods¶

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

Deprecated Method¶

Deprecated Class¶

Forecasting ¶

__init__ ¶

__getattr__ ¶

predict ¶

Price Forecasting (`kadi.market.forecasting`)¶

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

init ¶

getattr ¶