Pricing (`kadi.market.pricing`)¶

The Pricing class handles the acquisition, normalization, and analysis of agricultural market price data in Benin. It is used internally by the Market facade, but can also be instantiated directly.

Direct Importing¶

from kadi.market import Pricing

Initialization¶

Pricing is created automatically by the Market facade and is accessible via market.pricing. For direct instantiation:

from kadi.market import Pricing

# Without clients (automatic simulation mode)
pricing = Pricing()

# With WFP client and dynamic exchange rates
from kadi._sources import WFPClient, ExchangeRateClient
pricing = Pricing(
    wfp=WFPClient(),
    exchange=ExchangeRateClient(),
    sim=False,
)

Parameters:

Name	Type	Default	Description
`wfp`	`WFPClient`	`None`	WFP client for HAPI HumData API
`exchange`	`ExchangeRateClient`	`None`	Dynamic exchange rate client
`sim`	`bool`	`False`	If True, forces simulation mode for all methods

Public Attributes:

Attribute	Type	Description
`client`	`WFPClient`	Injected WFP client (or None)
`sim`	`bool`	Active simulation mode flag

Methods¶

`fetch(crop, market, days, sim)`¶

Retrieves historical prices for a crop at a given market location.

In real mode, queries the HAPI HumData API. In simulation mode (no client configured or sim=True), generates dummy prices following a normal distribution centered on 300 XOF/kg, explicitly tagged sim=True and confidence_score=0.1.

import kadi as kd

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

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

print(df.dtypes)
print(df.tail(5))

Parameters:

Name	Type	Default	Description
`crop`	`str`	required	Crop code (e.g. `'maize'`, `'rice'`)
`market`	`str`	required	Normalized market name (e.g. `'cotonou'`)
`days`	`int`	`365`	Number of historical days to fetch
`sim`	`bool`	`None`	Overrides instance simulation mode if provided

Returns: pd.DataFrame

Column	Type	Description
`date`	`datetime`	Observation date
`price`	`float`	Price in XOF/kg
`unit`	`str`	Original unit (e.g. `'XOF/kg'`)
`sim`	`bool`	True if dummy/simulated data
`source`	`str`	Data source identifier
`fetched_at`	`str`	ISO 8601 collection timestamp
`confidence_score`	`float`	Confidence score (0.0 to 1.0)

`convert_unit(value, unit, crop)`¶

Converts a price value to the standard unit XOF/kg.

# XOF/Tonne -> XOF/kg
price_kg = market.pricing.convert_unit(180_000.0, "XOF/Tonne")
print(f"{price_kg:.2f} XOF/kg")   # 180.0 XOF/kg

# USD/kg -> XOF/kg (via Frankfurter exchange rate)
price_kg = market.pricing.convert_unit(0.45, "USD/kg")

# XOF/sac (bag weight depends on crop)
price_kg = market.pricing.convert_unit(14_000.0, "XOF/sac", crop="maize")

Parameters:

Name	Type	Default	Description
`value`	`float`	required	Price value to convert
`unit`	`str`	required	Original unit
`crop`	`str`	`None`	Crop code (for container weight lookups)

Supported Units:

Unit	Conversion
`XOF/kg`	No conversion (target unit)
`XOF/Tonne`	Division by 1000
`USD/kg`	Multiplication by USD/XOF rate
`EUR/kg`	Multiplication by EUR/XOF rate (fixed WAEMU peg)
`XOF/sac`	Division by bag weight in kg (crop-dependent)
`XOF/boisseau`, `XOF/tine`, `XOF/caisse`	Division by container weight

If the unit is unknown, the value is returned unmodified.

Returns: float - Price in XOF/kg.

`anomalies(series, z)`¶

Detects price anomalies in a series using the Z-score method.

df = market.pricing.fetch("rice", "cotonou", days=180)
df = market.pricing.anomalies(df, z=3.0)

nb_anomalies = df["is_anomaly"].sum()
print(f"Detected anomalies: {nb_anomalies}")
print(df[df["is_anomaly"]][["date", "price"]])

Parameters:

Name	Type	Default	Description
`series`	`pd.DataFrame`	required	DataFrame containing a `'price'` column
`z`	`float`	`3.0`	Z-score threshold (3 = 99.7% of normal distribution)

Returns: pd.DataFrame - Input DataFrame with an added boolean column is_anomaly.

`fill_gaps(series, max_gap)`¶

Fills missing values in a price series using linear interpolation, capped at a configurable number of consecutive days.

df = market.pricing.fetch("sorghum", "natitingou", days=365)
df = market.pricing.fill_gaps(df, max_gap=7)

Parameters:

Name	Type	Default	Description
`series`	`pd.DataFrame`	required	DataFrame containing a `'price'` column
`max_gap`	`int`	`7`	Maximum consecutive days to interpolate

Gaps larger than max_gap consecutive days remain NaN.

Returns: pd.DataFrame - DataFrame with short gaps filled.

`source(series)`¶

Identifies the data source of a price series.

df = market.pricing.fetch("maize", "parakou", days=90)
src = market.pricing.source(df)
print(f"Source: {src}")   # 'wfp-vam' or 'simulated'

Parameters:

Name	Type	Description
`series`	`pd.DataFrame`	DataFrame returned by `fetch()`

Returns: str - 'wfp-vam', 'ratin', 'scrape-local', or 'simulated'.

`seasonality(historique, min_observations_par_mois)`¶

Calculates 12 monthly seasonal price indices for agricultural produce using the ratio-to-mean method.

An index greater than 1 indicates a high-price month (lean/hungry season). An index lower than 1 indicates a cheap month (post-harvest season).

df = market.pricing.fetch("maize", "parakou", days=730)
season = market.pricing.seasonality(df)

print(f"Peak month    : {season['mois_pic']}")
print(f"Trough month  : {season['mois_creux']}")
print(f"Average price : {season['prix_moyen_global']:.2f} XOF/kg")
print(f"Confidence    : {season['confiance']:.2f}")

# Index per month
for month, index in season["indices"].items():
    if index:
        print(f"  Month {month:2d} : {index:.3f}")

Parameters:

Name	Type	Default	Description
`historique`	`pd.DataFrame`	required	DataFrame with `'date'` and `'price'` columns
`min_observations_par_mois`	`int`	`2`	Minimum observations required to include a month

Returns: dict

Key	Type	Description
`indices`	`dict[int, float]`	Monthly seasonal indices (1 to 12). None if insufficient data
`mois_pic`	`list[int]`	Months where index exceeds 1.05 (5% above mean)
`mois_creux`	`list[int]`	Months where index is below 0.95
`prix_moyen_global`	`float`	Average price over entire period (XOF/kg)
`prix_moyen_par_mois`	`dict[int, float]`	Raw average price per month
`nb_observations`	`int`	Total count of valid observations
`nb_mois_couverts`	`int`	Count of months meeting `min_observations_par_mois`
`confiance`	`float`	Confidence score (0.0 to 1.0)
`sim`	`bool`	True if history contains simulated data
`message`	`str`	Warning message if data is insufficient. None otherwise

Exceptions:

ValueError: if historical DataFrame is empty, or missing 'date' / 'price' columns.

Deprecated Methods¶

Legacy Method	Replaced By	Since
`fetch_prices(crop, market, days_back)`	`fetch(crop, market, days)`	v1.2.0
`normalize_units(value, unit)`	`convert_unit(value, unit)`	v1.2.0
`detect_anomalies(series, threshold)`	`anomalies(series, z)`	v1.2.0
`interpolate_gaps(series, max_gap)`	`fill_gaps(series, max_gap)`	v1.2.0
`get_data_source(series)`	`source(series)`	v1.2.0

Pricing ¶

Classe gérant l'ingestion, la normalisation et la détection d'anomalies pour les données de prix de marché agricole au Bénin.

Source code in kadi/market/pricing.py

class Pricing:
    """
    Classe gérant l'ingestion, la normalisation et la détection d'anomalies
    pour les données de prix de marché agricole au Bénin.
    """

    def __init__(
        self,
        wfp=None,
        exchange=None,
        sim: bool = False,
        **kwargs,
    ):
        """
        Initialise le module de tarification.

        Args:
            wfp (WFPClient, optional): Instance de WFPClient pour récupérer les données de prix.
            exchange (ExchangeRateClient, optional): Instance d'ExchangeRateClient
                pour les taux de change dynamiques. Si None, utilise config.EXCHANGE_RATES.
            sim (bool, optional): Si True, force le mode simulation pour
                toutes les opérations de tarification. Défaut : False.
            sim (bool, optional): Si True, force le mode simulation.
            **kwargs: Arguments obsolètes acceptés pour rétrocompatibilité.
        """
        # Prise en charge des anciens noms d'arguments
        if "wfp_client" in kwargs:
            wfp = kwargs.pop("wfp_client")
        if "exchange_client" in kwargs:
            exchange = kwargs.pop("exchange_client")
        if "simulated" in kwargs:
            sim = kwargs.pop("simulated")

        # Instance du client WFP DataBridges
        self.client = wfp

        # Mode simulation explicite
        self.sim = sim

        # Récupération des taux de change dynamiques ou statiques
        if exchange is not None:
            # Le client retourne toujours un dictionnaire valide (fallback intégré)
            self._fx = exchange.get_rates()
            logger.debug(
                f"Taux de change chargés depuis ExchangeRateClient : "
                f"{self._fx}"
            )
        else:
            # Pas de client : on utilise les taux statiques de config.py
            self._fx = dict(_EXCHANGE_RATES_FALLBACK)
            logger.debug(
                "ExchangeRateClient non fourni. Taux de repli de config.py utilisés."
            )

    # ------------------------------------------------------------------
    # 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"Pricing.{name}() est obsolète et sera supprimé dans "
                f"KadiPy v2.0. Utilisez Pricing.{new_name}() à la place.",
                DeprecationWarning,
                stacklevel=2,
            )
            return getattr(self, new_name)
        raise AttributeError(f"'Pricing' n'a pas d'attribut '{name}'.") 

    def fetch(
        self,
        crop: str,
        market: str,
        days: int = 365,
        sim: bool = None,
    ) -> pd.DataFrame:
        """
        Récupère les prix historiques pour une culture et un marché donnés.

        Le paramètre ``sim`` permet de choisir explicitement entre
        les données réelles de l'API HAPI HumData (PAM) et des données
        générées mathématiquement à des fins de test ou de démonstration.

        En mode simulé, aucun appel réseau n'est effectué. Les données générées
        suivent une distribution normale centrée sur 300 XOF/kg (ecart-type 20)
        et sont clairement marquées ``sim=True`` avec un score de
        confiance de 0.1 pour signaler leur origine fictive.

        Args:
            crop (str): Code de la culture (ex: 'maize', 'rice').
            market (str): Nom normalisé du marché (ex: 'cotonou').
            days (int, optional): Nombre de jours d'historique. Défaut à 365.
            sim (bool, optional): Surcharge le mode simulation.
                Si None, hérite de self.sim. Défaut : None.

        Returns:
            pd.DataFrame: DataFrame avec les colonnes 'date', 'price', 'unit',
                'sim', 'source', 'fetched_at', 'confidence_score'.
        """
        # Résolution du mode simulation : paramètre local ou instance
        mode_simule = self.sim if sim is None else sim

        # Calcul de la plage de dates pour la requête API
        end_date = datetime.date.today()
        start_date = end_date - datetime.timedelta(days=days)
        time_range = (start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d"))

        # --- Mode simulation explicitement demandé par l'utilisateur ---
        if mode_simule:
            logger.info(
                f"Mode simulation activé pour '{crop}' / '{market}'. "
                "Aucune requête réseau ne sera effectuée. "
                "Les données retournées sont fictives (sim=True, "
                "confidence_score=0.1). Ne pas utiliser pour des décisions "
                "commerciales réelles."
            )
            return self._simulate(crop, market, days, end_date)

        if self.client is not None:
            # Appel au client WFP (qui gère lui-même le retry, le cache et le fallback)
            df_prices = self.client.get_market_prices(market, crop, time_range)
        else:
            # Pas de client configuré : on génère des données de simulation
            logger.warning(
                f"Aucun client WFP configuré pour '{crop}' / '{market}'. "
                "Données simulées utilisées en remplacement "
                "(sim=True, confidence_score=0.1)."
            )
            df_prices = self._simulate(crop, market, days, end_date)

        # Rétrocompatibilité : synchronisation des colonnes sim et is_simulated
        if "sim" in df_prices.columns and "is_simulated" not in df_prices.columns:
            df_prices["is_simulated"] = df_prices["sim"]
        elif "is_simulated" in df_prices.columns and "sim" not in df_prices.columns:
            df_prices["sim"] = df_prices["is_simulated"]

        return df_prices

    def _simulate(
        self,
        crop: str,
        market: str,
        days: int,
        end_date: datetime.date,
    ) -> pd.DataFrame:
        """
        Génère un DataFrame de prix fictifs par distribution normale.

        Les prix générés oscillent autour de 300 XOF/kg avec un ecart-type
        de 20 XOF, ce qui est representatif des cereales seches au Benin.
        Toutes les colonnes de metadonnees sont positionnees pour signaler
        explicitement l'origine simulee des donnees.

        Args:
            crop (str): Code de la culture (pour les logs).
            market (str): Nom du marche (pour les logs).
            days (int): Nombre de jours de données à générer.
            end_date (datetime.date): Date de fin de la serie.

        Returns:
            pd.DataFrame: Données simulées avec sim=True.
        """
        # Série de dates quotidiennes terminant aujourd'hui
        dates = pd.date_range(end=end_date, periods=days, freq="D")

        # Prix générés par distribution normale centrée sur 300 XOF/kg
        prix_aleatoires = np.random.normal(loc=300, scale=20, size=days)
        maintenant = datetime.datetime.now(datetime.timezone.utc).isoformat()

        logger.debug(
            f"Données simulées générées pour '{crop}' / '{market}' "
            f"({days} observations, moy=300 XOF/kg, ecart-type=20)."
        )

        return pd.DataFrame({
            "date": dates,
            "price": prix_aleatoires,
            "unit": "XOF/kg",
            # Marqueurs explicites d'origine simulée
            "sim": True,
            "source": "simulated",
            "fetched_at": maintenant,
            "confidence_score": 0.1,
        })

    def convert_unit(self, value: float, unit: str, crop: str = None) -> float:
        """
        Convertit une valeur de prix vers le standard XOF/kg.

        Gère les conversions suivantes :
        - XOF/Tonne    -> XOF/kg (division par 1000)
        - USD/kg       -> XOF/kg (multiplication par le taux de change)
        - EUR/kg       -> XOF/kg (multiplication par le taux de change)
        - XOF/sac      -> XOF/kg (division par le poids du sac en kg)
        - XOF/boisseau -> XOF/kg (division par le poids du boisseau)
        - XOF/tine     -> XOF/kg
        - XOF/caisse   -> XOF/kg (pour les produits frais comme la tomate)
        - XOF/kg       -> sans changement (déjà à l'unité standard)

        Si l'unité est inconnue, la valeur est retournée sans modification
        et un avertissement est enregistré.

        Args:
            value (float): La valeur du prix à convertir.
            unit (str): L'unité d'origine (ex: 'XOF/Tonne', 'USD/kg', 'XOF/sac').
            crop (str, optional): Code de la culture, utilisé pour les poids
                de contenants spécifiques (ex: poids d'un sac de maïs vs riz).

        Returns:
            float: Le prix normalisé en XOF/kg.
        """
        valeur = float(value)

        # Normalisation de l'unité pour la comparaison
        unite = unit.strip().lower()

        # --- Conversion des tonnes ---
        if "tonne" in unite or "/t" == unite[-2:]:
            # 1 tonne = 1000 kg
            return valeur / 1000.0

        # --- Conversion des devises étrangères ---
        if unite.startswith("usd"):
            # USD vers XOF : lecture des taux de l'instance
            taux = self._fx.get("USD_TO_XOF", _EXCHANGE_RATES_FALLBACK["USD_TO_XOF"])
            return valeur * taux

        if unite.startswith("eur"):
            # EUR vers XOF (taux fixe UEMOA ou taux dynamique)
            taux = self._fx.get("EUR_TO_XOF", _EXCHANGE_RATES_FALLBACK["EUR_TO_XOF"])
            return valeur * taux

        # --- Conversion des contenants locaux ---
        for conteneur in ("sac", "boisseau", "tine", "caisse", "boite", "panier"):
            if conteneur in unite:
                poids_kg = get_container_weight_kg(conteneur, crop)
                if poids_kg > 0:
                    return valeur / poids_kg
                # Poids inconnu : avertissement et valeur inchangée
                logger.warning(
                    f"Poids inconnu pour le conteneur '{conteneur}' / culture '{crop}'. "
                    "Le prix est retourné sans conversion."
                )
                return valeur

        # --- Unité déjà en XOF/kg ---
        if "xof/kg" in unite or unite == "kg":
            return valeur

        # --- Unité inconnue ---
        logger.warning(
            f"Unité inconnue : '{unit}'. "
            "Le prix est retourné sans conversion."
        )
        return valeur

    def anomalies(self, series: pd.DataFrame, z: float = 3.0) -> pd.DataFrame:
        """
        Détecte les anomalies dans une série de prix par la méthode du Z-score.

        Un prix est considéré comme anormal si son Z-score absolu dépasse
        le seuil configuré (par défaut : 3, soit environ 99.7% de la distribution).

        Args:
            series (pd.DataFrame): DataFrame contenant une colonne 'price'.
            z (float, optional): Seuil d'anomalie. Défaut à 3.0.

        Returns:
            pd.DataFrame: DataFrame original avec une colonne booléenne 'is_anomaly'.
        """
        # Copie pour ne pas modifier le DataFrame d'entrée
        df_result = series.copy()

        if "price" not in df_result.columns:
            logger.warning("Colonne 'price' absente du DataFrame. Aucune détection effectuée.")
            df_result["is_anomaly"] = False
            return df_result

        # Calcul des statistiques de la série
        moyenne = df_result["price"].mean()
        ecart_type = df_result["price"].std()

        if ecart_type > 0:
            # Calcul du Z-score pour chaque observation
            z_scores = (df_result["price"] - moyenne) / ecart_type
            # Marquage des anomalies au-delà du seuil
            df_result["is_anomaly"] = np.abs(z_scores) > z
        else:
            # Série constante : aucune variabilité, donc pas d'anomalie
            df_result["is_anomaly"] = False

        return df_result

    def fill_gaps(self, series: pd.DataFrame, max_gap: int = 7) -> pd.DataFrame:
        """
        Comble les valeurs manquantes dans une série de prix par interpolation linéaire.

        L'interpolation est limitée à un nombre configurable de jours consécutifs.
        Au-delà de cette limite, les valeurs restent manquantes (NaN) pour signaler
        un trou important dans les données.

        Args:
            series (pd.DataFrame): DataFrame contenant une colonne 'price'.
            max_gap (int, optional): Nombre maximum de jours à interpoler. Défaut à 7.

        Returns:
            pd.DataFrame: DataFrame avec les trous courts comblés par interpolation.
        """
        # Copie pour préserver le DataFrame d'entrée
        df_interpolated = series.copy()

        if "price" in df_interpolated.columns:
            # Interpolation linéaire avec limite sur la longueur des trous
            df_interpolated["price"] = df_interpolated["price"].interpolate(
                method="linear",
                limit=max_gap,
                limit_direction="both",
            )

        return df_interpolated

    def source(self, series: pd.DataFrame) -> str:
        """
        Identifie la source des données d'une série de prix.

        Si le DataFrame contient la colonne ``sim``, la méthode
        retourne 'simulated' ou 'wfp-vam' selon le cas.

        Args:
            series (pd.DataFrame): DataFrame retourné par fetch().

        Returns:
            str: La source identifiée ('wfp-vam', 'ratin', 'scrape-local', 'simulated').
        """
        # Vérification de la colonne sim
        if "sim" in series.columns:
            if series["sim"].any():
                return "simulated"

        # Par défaut, on suppose WFP VAM (source primaire du V1)
        return "wfp-vam"

    def seasonality(
        self,
        historique: pd.DataFrame,
        min_observations_par_mois: int = 2,
    ) -> dict:
        """
        Calcule l'indice saisonnier mensuel des prix agricoles.

        La méthode utilise la décomposition par ratios : pour chaque mois,
        l'indice est le rapport entre le prix moyen de ce mois et le prix
        moyen global sur toute la période. Un indice supérieur à 1 indique
        un mois de prix élevés (pénurie), inférieur à 1 un mois bon marché
        (période post-récolte).

        L'historique doit couvrir au moins 12 mois pour que les indices
        soient fiables. En dessous de ce seuil, les indices sont calculés
        mais le champ ``confiance`` sera faible.

        Args:
            historique (pd.DataFrame): DataFrame avec au minimum les colonnes
                'date' (datetime ou str) et 'price' (float, en XOF/kg).
                Typiquement retourné par ``fetch_prices()``.
            min_observations_par_mois (int, optional): Nombre minimal
                d'observations pour qu'un mois soit inclus dans le calcul.
                Les mois en dessous de ce seuil sont marqués NaN.
                Défaut : 2.

        Returns:
            dict: Dictionnaire contenant les champs suivants :

                - ``indices`` (dict[int, float | None]) : dictionnaire des
                  12 indices saisonniers, indexé par numéro de mois (1 à 12).
                  La valeur est None si le mois a moins de
                  ``min_observations_par_mois`` entrées.
                - ``mois_pic`` (list[int]) : liste des mois où l'indice
                  dépasse 1.05 (5% au-dessus de la moyenne), triés par
                  indice décroissant.
                - ``mois_creux`` (list[int]) : liste des mois où l'indice
                  est en dessous de 0.95, triés par indice croissant.
                - ``prix_moyen_global`` (float) : prix moyen sur toute la
                  période historique, en XOF/kg.
                - ``prix_moyen_par_mois`` (dict[int, float | None]) :
                  prix moyen brut par mois, avant normalisation.
                - ``nb_observations`` (int) : nombre total d'observations
                  valides utilisées pour le calcul.
                - ``nb_mois_couverts`` (int) : nombre de mois avec au moins
                  ``min_observations_par_mois`` entrées.
                - ``confiance`` (float) : score de confiance de 0 à 1.
                  Reflète la densité des données (1.0 = 2+ ans de données
                  hebdomadaires, 0.0 = moins d'un mois de données).
                - ``sim`` (bool) : True si l'historique source
                  contient des données simulées.
                - ``message`` (str | None) : avertissement si les données
                  sont insuffisantes pour un calcul fiable. None sinon.

        Raises:
            ValueError: Si l'historique est vide ou ne contient pas les
                colonnes 'date' et 'price'.
        """
        # --- Validation des entrées ---
        if historique is None or historique.empty:
            raise ValueError(
                "L'historique fourni est vide. "
                "Utilisez fetch_prices() pour obtenir des données avant "
                "d'appeler seasonality()."
            )

        colonnes_requises = {"date", "price"}
        if not colonnes_requises.issubset(historique.columns):
            raise ValueError(
                f"L'historique doit contenir les colonnes {colonnes_requises}. "
                f"Colonnes reçues : {set(historique.columns)}."
            )

        # --- Préparation du DataFrame ---
        # Copie pour ne pas modifier le DataFrame d'entrée
        df = historique[["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 ou invalides
        df = df.dropna(subset=["date", "price"])
        df = df[df["price"] > 0]

        if df.empty:
            raise ValueError(
                "Aucune observation valide après nettoyage de l'historique "
                "(vérifiez les colonnes 'date' et 'price')."
            )

        # Extraction du numéro de mois (1 = janvier, 12 = décembre)
        df["mois"] = df["date"].dt.month

        # --- Calcul des statistiques par mois ---
        # Comptage des observations disponibles par mois
        compte_par_mois = df.groupby("mois")["price"].count()

        # Prix moyen brut par mois (tous mois confondus)
        moyenne_par_mois = df.groupby("mois")["price"].mean()

        # Prix moyen global sur toute la période (référence de normalisation)
        prix_moyen_global = float(df["price"].mean())

        # Noms des mois en français pour les messages lisibles
        _NOMS_MOIS = {
            1: "janvier", 2: "février", 3: "mars", 4: "avril",
            5: "mai", 6: "juin", 7: "juillet", 8: "août",
            9: "septembre", 10: "octobre", 11: "novembre", 12: "décembre",
        }

        # --- Construction des indices saisonniers ---
        indices = {}
        prix_moyen_par_mois = {}

        for mois in range(1, 13):
            nb_obs = int(compte_par_mois.get(mois, 0))
            prix_moyen_par_mois[mois] = (
                round(float(moyenne_par_mois[mois]), 2)
                if mois in moyenne_par_mois.index
                else None
            )

            if nb_obs >= min_observations_par_mois and prix_moyen_global > 0:
                # Indice = ratio entre prix moyen du mois et prix moyen global
                # Indice > 1 : mois cher, < 1 : mois bon marché
                indice = float(moyenne_par_mois[mois]) / prix_moyen_global
                indices[mois] = round(indice, 4)
            else:
                # Données insuffisantes pour ce mois
                indices[mois] = None

        # --- Identification des mois de pic et de creux ---
        # Seuil de 5% au-dessus/en dessous de la moyenne pour éviter le bruit
        _SEUIL_PIC = 1.05
        _SEUIL_CREUX = 0.95

        mois_pic = sorted(
            [m for m, idx in indices.items() if idx is not None and idx >= _SEUIL_PIC],
            key=lambda m: indices[m],
            reverse=True,
        )
        mois_creux = sorted(
            [m for m, idx in indices.items() if idx is not None and idx <= _SEUIL_CREUX],
            key=lambda m: indices[m],
        )

        # --- Calcul du score de confiance ---
        nb_mois_couverts = sum(1 for v in indices.values() if v is not None)
        nb_observations = len(df)

        # La confiance dépend de deux facteurs :
        # 1. La couverture mensuelle : 12 mois = 1.0, 0 mois = 0.0
        facteur_couverture = nb_mois_couverts / 12.0
        # 2. La densité des données : on considère 104 obs (2 ans hebdo) comme optimal
        facteur_densite = min(1.0, nb_observations / 104.0)
        confiance = round((facteur_couverture + facteur_densite) / 2.0, 3)

        # --- Propagation du flag sim depuis la source ---
        est_simule = False
        if "sim" in historique.columns:
            est_simule = bool(historique["sim"].any())
        elif "is_simulated" in historique.columns:
            est_simule = bool(historique["is_simulated"].any())

        # --- Message d'avertissement si données insuffisantes ---
        message = None
        if nb_mois_couverts < 6:
            message = (
                f"Seulement {nb_mois_couverts} mois couverts sur 12. "
                "Les indices saisonniers calculés sont peu fiables. "
                "Il est recommandé d'avoir au moins 12 mois d'historique."
            )
        elif nb_mois_couverts < 12:
            manquants = [_NOMS_MOIS[m] for m in range(1, 13) if indices[m] is None]
            message = (
                f"Données manquantes pour : {', '.join(manquants)}. "
                "L'indice saisonnier est None pour ces mois."
            )

        if est_simule:
            avert_simul = "Attention : l'historique est simulé. Les indices ne reflètent pas la réalité du marché."
            message = f"{avert_simul} {message}" if message else avert_simul

        logger.info(
            f"Saisonnalité calculée : {nb_mois_couverts}/12 mois couverts, "
            f"{nb_observations} observations, confiance={confiance}, "
            f"pic={mois_pic}, creux={mois_creux}."
        )

        return {
            "indices": indices,
            "mois_pic": mois_pic,
            "mois_creux": mois_creux,
            "prix_moyen_global": round(prix_moyen_global, 2),
            "prix_moyen_par_mois": prix_moyen_par_mois,
            "nb_observations": nb_observations,
            "nb_mois_couverts": nb_mois_couverts,
            "confiance": confiance,
            "sim": est_simule,
            "is_simulated": est_simule,
            "message": message,
        }

init ¶

__init__(wfp=None, exchange=None, sim: bool = False, **kwargs)

Initialise le module de tarification.

Parameters:

Name	Type	Description	Default
`wfp`	`WFPClient`	Instance de WFPClient pour récupérer les données de prix.	`None`
`exchange`	`ExchangeRateClient`	Instance d'ExchangeRateClient pour les taux de change dynamiques. Si None, utilise config.EXCHANGE_RATES.	`None`
`sim`	`bool`	Si True, force le mode simulation pour toutes les opérations de tarification. Défaut : False.	`False`
`sim`	`bool`	Si True, force le mode simulation.	`False`
`**kwargs`		Arguments obsolètes acceptés pour rétrocompatibilité.	`{}`

Source code in kadi/market/pricing.py

def __init__(
    self,
    wfp=None,
    exchange=None,
    sim: bool = False,
    **kwargs,
):
    """
    Initialise le module de tarification.

    Args:
        wfp (WFPClient, optional): Instance de WFPClient pour récupérer les données de prix.
        exchange (ExchangeRateClient, optional): Instance d'ExchangeRateClient
            pour les taux de change dynamiques. Si None, utilise config.EXCHANGE_RATES.
        sim (bool, optional): Si True, force le mode simulation pour
            toutes les opérations de tarification. Défaut : False.
        sim (bool, optional): Si True, force le mode simulation.
        **kwargs: Arguments obsolètes acceptés pour rétrocompatibilité.
    """
    # Prise en charge des anciens noms d'arguments
    if "wfp_client" in kwargs:
        wfp = kwargs.pop("wfp_client")
    if "exchange_client" in kwargs:
        exchange = kwargs.pop("exchange_client")
    if "simulated" in kwargs:
        sim = kwargs.pop("simulated")

    # Instance du client WFP DataBridges
    self.client = wfp

    # Mode simulation explicite
    self.sim = sim

    # Récupération des taux de change dynamiques ou statiques
    if exchange is not None:
        # Le client retourne toujours un dictionnaire valide (fallback intégré)
        self._fx = exchange.get_rates()
        logger.debug(
            f"Taux de change chargés depuis ExchangeRateClient : "
            f"{self._fx}"
        )
    else:
        # Pas de client : on utilise les taux statiques de config.py
        self._fx = dict(_EXCHANGE_RATES_FALLBACK)
        logger.debug(
            "ExchangeRateClient non fourni. Taux de repli de config.py utilisés."
        )

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/pricing.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"Pricing.{name}() est obsolète et sera supprimé dans "
            f"KadiPy v2.0. Utilisez Pricing.{new_name}() à la place.",
            DeprecationWarning,
            stacklevel=2,
        )
        return getattr(self, new_name)
    raise AttributeError(f"'Pricing' n'a pas d'attribut '{name}'.") 

fetch ¶

fetch(crop: str, market: str, days: int = 365, sim: bool = None) -> pd.DataFrame

Récupère les prix historiques pour une culture et un marché donnés.

Le paramètre sim permet de choisir explicitement entre les données réelles de l'API HAPI HumData (PAM) et des données générées mathématiquement à des fins de test ou de démonstration.

En mode simulé, aucun appel réseau n'est effectué. Les données générées suivent une distribution normale centrée sur 300 XOF/kg (ecart-type 20) et sont clairement marquées sim=True avec un score de confiance de 0.1 pour signaler leur origine fictive.

Parameters:

Name	Type	Description	Default
`crop`	`str`	Code de la culture (ex: 'maize', 'rice').	required
`market`	`str`	Nom normalisé du marché (ex: 'cotonou').	required
`days`	`int`	Nombre de jours d'historique. Défaut à 365.	`365`
`sim`	`bool`	Surcharge le mode simulation. Si None, hérite de self.sim. Défaut : None.	`None`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec les colonnes 'date', 'price', 'unit', 'sim', 'source', 'fetched_at', 'confidence_score'.

Source code in kadi/market/pricing.py

def fetch(
    self,
    crop: str,
    market: str,
    days: int = 365,
    sim: bool = None,
) -> pd.DataFrame:
    """
    Récupère les prix historiques pour une culture et un marché donnés.

    Le paramètre ``sim`` permet de choisir explicitement entre
    les données réelles de l'API HAPI HumData (PAM) et des données
    générées mathématiquement à des fins de test ou de démonstration.

    En mode simulé, aucun appel réseau n'est effectué. Les données générées
    suivent une distribution normale centrée sur 300 XOF/kg (ecart-type 20)
    et sont clairement marquées ``sim=True`` avec un score de
    confiance de 0.1 pour signaler leur origine fictive.

    Args:
        crop (str): Code de la culture (ex: 'maize', 'rice').
        market (str): Nom normalisé du marché (ex: 'cotonou').
        days (int, optional): Nombre de jours d'historique. Défaut à 365.
        sim (bool, optional): Surcharge le mode simulation.
            Si None, hérite de self.sim. Défaut : None.

    Returns:
        pd.DataFrame: DataFrame avec les colonnes 'date', 'price', 'unit',
            'sim', 'source', 'fetched_at', 'confidence_score'.
    """
    # Résolution du mode simulation : paramètre local ou instance
    mode_simule = self.sim if sim is None else sim

    # Calcul de la plage de dates pour la requête API
    end_date = datetime.date.today()
    start_date = end_date - datetime.timedelta(days=days)
    time_range = (start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d"))

    # --- Mode simulation explicitement demandé par l'utilisateur ---
    if mode_simule:
        logger.info(
            f"Mode simulation activé pour '{crop}' / '{market}'. "
            "Aucune requête réseau ne sera effectuée. "
            "Les données retournées sont fictives (sim=True, "
            "confidence_score=0.1). Ne pas utiliser pour des décisions "
            "commerciales réelles."
        )
        return self._simulate(crop, market, days, end_date)

    if self.client is not None:
        # Appel au client WFP (qui gère lui-même le retry, le cache et le fallback)
        df_prices = self.client.get_market_prices(market, crop, time_range)
    else:
        # Pas de client configuré : on génère des données de simulation
        logger.warning(
            f"Aucun client WFP configuré pour '{crop}' / '{market}'. "
            "Données simulées utilisées en remplacement "
            "(sim=True, confidence_score=0.1)."
        )
        df_prices = self._simulate(crop, market, days, end_date)

    # Rétrocompatibilité : synchronisation des colonnes sim et is_simulated
    if "sim" in df_prices.columns and "is_simulated" not in df_prices.columns:
        df_prices["is_simulated"] = df_prices["sim"]
    elif "is_simulated" in df_prices.columns and "sim" not in df_prices.columns:
        df_prices["sim"] = df_prices["is_simulated"]

    return df_prices

convert_unit ¶

convert_unit(value: float, unit: str, crop: str = None) -> float

Convertit une valeur de prix vers le standard XOF/kg.

Gère les conversions suivantes : - XOF/Tonne -> XOF/kg (division par 1000) - USD/kg -> XOF/kg (multiplication par le taux de change) - EUR/kg -> XOF/kg (multiplication par le taux de change) - XOF/sac -> XOF/kg (division par le poids du sac en kg) - XOF/boisseau -> XOF/kg (division par le poids du boisseau) - XOF/tine -> XOF/kg - XOF/caisse -> XOF/kg (pour les produits frais comme la tomate) - XOF/kg -> sans changement (déjà à l'unité standard)

Si l'unité est inconnue, la valeur est retournée sans modification et un avertissement est enregistré.

Parameters:

Name	Type	Description	Default
`value`	`float`	La valeur du prix à convertir.	required
`unit`	`str`	L'unité d'origine (ex: 'XOF/Tonne', 'USD/kg', 'XOF/sac').	required
`crop`	`str`	Code de la culture, utilisé pour les poids de contenants spécifiques (ex: poids d'un sac de maïs vs riz).	`None`

Returns:

Name	Type	Description
`float`	`float`	Le prix normalisé en XOF/kg.

Source code in kadi/market/pricing.py

def convert_unit(self, value: float, unit: str, crop: str = None) -> float:
    """
    Convertit une valeur de prix vers le standard XOF/kg.

    Gère les conversions suivantes :
    - XOF/Tonne    -> XOF/kg (division par 1000)
    - USD/kg       -> XOF/kg (multiplication par le taux de change)
    - EUR/kg       -> XOF/kg (multiplication par le taux de change)
    - XOF/sac      -> XOF/kg (division par le poids du sac en kg)
    - XOF/boisseau -> XOF/kg (division par le poids du boisseau)
    - XOF/tine     -> XOF/kg
    - XOF/caisse   -> XOF/kg (pour les produits frais comme la tomate)
    - XOF/kg       -> sans changement (déjà à l'unité standard)

    Si l'unité est inconnue, la valeur est retournée sans modification
    et un avertissement est enregistré.

    Args:
        value (float): La valeur du prix à convertir.
        unit (str): L'unité d'origine (ex: 'XOF/Tonne', 'USD/kg', 'XOF/sac').
        crop (str, optional): Code de la culture, utilisé pour les poids
            de contenants spécifiques (ex: poids d'un sac de maïs vs riz).

    Returns:
        float: Le prix normalisé en XOF/kg.
    """
    valeur = float(value)

    # Normalisation de l'unité pour la comparaison
    unite = unit.strip().lower()

    # --- Conversion des tonnes ---
    if "tonne" in unite or "/t" == unite[-2:]:
        # 1 tonne = 1000 kg
        return valeur / 1000.0

    # --- Conversion des devises étrangères ---
    if unite.startswith("usd"):
        # USD vers XOF : lecture des taux de l'instance
        taux = self._fx.get("USD_TO_XOF", _EXCHANGE_RATES_FALLBACK["USD_TO_XOF"])
        return valeur * taux

    if unite.startswith("eur"):
        # EUR vers XOF (taux fixe UEMOA ou taux dynamique)
        taux = self._fx.get("EUR_TO_XOF", _EXCHANGE_RATES_FALLBACK["EUR_TO_XOF"])
        return valeur * taux

    # --- Conversion des contenants locaux ---
    for conteneur in ("sac", "boisseau", "tine", "caisse", "boite", "panier"):
        if conteneur in unite:
            poids_kg = get_container_weight_kg(conteneur, crop)
            if poids_kg > 0:
                return valeur / poids_kg
            # Poids inconnu : avertissement et valeur inchangée
            logger.warning(
                f"Poids inconnu pour le conteneur '{conteneur}' / culture '{crop}'. "
                "Le prix est retourné sans conversion."
            )
            return valeur

    # --- Unité déjà en XOF/kg ---
    if "xof/kg" in unite or unite == "kg":
        return valeur

    # --- Unité inconnue ---
    logger.warning(
        f"Unité inconnue : '{unit}'. "
        "Le prix est retourné sans conversion."
    )
    return valeur

anomalies ¶

anomalies(series: DataFrame, z: float = 3.0) -> pd.DataFrame

Détecte les anomalies dans une série de prix par la méthode du Z-score.

Un prix est considéré comme anormal si son Z-score absolu dépasse le seuil configuré (par défaut : 3, soit environ 99.7% de la distribution).

Parameters:

Name	Type	Description	Default
`series`	`DataFrame`	DataFrame contenant une colonne 'price'.	required
`z`	`float`	Seuil d'anomalie. Défaut à 3.0.	`3.0`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame original avec une colonne booléenne 'is_anomaly'.

Source code in kadi/market/pricing.py

def anomalies(self, series: pd.DataFrame, z: float = 3.0) -> pd.DataFrame:
    """
    Détecte les anomalies dans une série de prix par la méthode du Z-score.

    Un prix est considéré comme anormal si son Z-score absolu dépasse
    le seuil configuré (par défaut : 3, soit environ 99.7% de la distribution).

    Args:
        series (pd.DataFrame): DataFrame contenant une colonne 'price'.
        z (float, optional): Seuil d'anomalie. Défaut à 3.0.

    Returns:
        pd.DataFrame: DataFrame original avec une colonne booléenne 'is_anomaly'.
    """
    # Copie pour ne pas modifier le DataFrame d'entrée
    df_result = series.copy()

    if "price" not in df_result.columns:
        logger.warning("Colonne 'price' absente du DataFrame. Aucune détection effectuée.")
        df_result["is_anomaly"] = False
        return df_result

    # Calcul des statistiques de la série
    moyenne = df_result["price"].mean()
    ecart_type = df_result["price"].std()

    if ecart_type > 0:
        # Calcul du Z-score pour chaque observation
        z_scores = (df_result["price"] - moyenne) / ecart_type
        # Marquage des anomalies au-delà du seuil
        df_result["is_anomaly"] = np.abs(z_scores) > z
    else:
        # Série constante : aucune variabilité, donc pas d'anomalie
        df_result["is_anomaly"] = False

    return df_result

fill_gaps ¶

fill_gaps(series: DataFrame, max_gap: int = 7) -> pd.DataFrame

Comble les valeurs manquantes dans une série de prix par interpolation linéaire.

L'interpolation est limitée à un nombre configurable de jours consécutifs. Au-delà de cette limite, les valeurs restent manquantes (NaN) pour signaler un trou important dans les données.

Parameters:

Name	Type	Description	Default
`series`	`DataFrame`	DataFrame contenant une colonne 'price'.	required
`max_gap`	`int`	Nombre maximum de jours à interpoler. Défaut à 7.	`7`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec les trous courts comblés par interpolation.

Source code in kadi/market/pricing.py

def fill_gaps(self, series: pd.DataFrame, max_gap: int = 7) -> pd.DataFrame:
    """
    Comble les valeurs manquantes dans une série de prix par interpolation linéaire.

    L'interpolation est limitée à un nombre configurable de jours consécutifs.
    Au-delà de cette limite, les valeurs restent manquantes (NaN) pour signaler
    un trou important dans les données.

    Args:
        series (pd.DataFrame): DataFrame contenant une colonne 'price'.
        max_gap (int, optional): Nombre maximum de jours à interpoler. Défaut à 7.

    Returns:
        pd.DataFrame: DataFrame avec les trous courts comblés par interpolation.
    """
    # Copie pour préserver le DataFrame d'entrée
    df_interpolated = series.copy()

    if "price" in df_interpolated.columns:
        # Interpolation linéaire avec limite sur la longueur des trous
        df_interpolated["price"] = df_interpolated["price"].interpolate(
            method="linear",
            limit=max_gap,
            limit_direction="both",
        )

    return df_interpolated

source ¶

source(series: DataFrame) -> str

Identifie la source des données d'une série de prix.

Si le DataFrame contient la colonne sim, la méthode retourne 'simulated' ou 'wfp-vam' selon le cas.

Parameters:

Name	Type	Description	Default
`series`	`DataFrame`	DataFrame retourné par fetch().	required

Returns:

Name	Type	Description
`str`	`str`	La source identifiée ('wfp-vam', 'ratin', 'scrape-local', 'simulated').

Source code in kadi/market/pricing.py

def source(self, series: pd.DataFrame) -> str:
    """
    Identifie la source des données d'une série de prix.

    Si le DataFrame contient la colonne ``sim``, la méthode
    retourne 'simulated' ou 'wfp-vam' selon le cas.

    Args:
        series (pd.DataFrame): DataFrame retourné par fetch().

    Returns:
        str: La source identifiée ('wfp-vam', 'ratin', 'scrape-local', 'simulated').
    """
    # Vérification de la colonne sim
    if "sim" in series.columns:
        if series["sim"].any():
            return "simulated"

    # Par défaut, on suppose WFP VAM (source primaire du V1)
    return "wfp-vam"

seasonality ¶

seasonality(historique: DataFrame, min_observations_par_mois: int = 2) -> dict

Calcule l'indice saisonnier mensuel des prix agricoles.

La méthode utilise la décomposition par ratios : pour chaque mois, l'indice est le rapport entre le prix moyen de ce mois et le prix moyen global sur toute la période. Un indice supérieur à 1 indique un mois de prix élevés (pénurie), inférieur à 1 un mois bon marché (période post-récolte).

L'historique doit couvrir au moins 12 mois pour que les indices soient fiables. En dessous de ce seuil, les indices sont calculés mais le champ confiance sera faible.

Parameters:

Name	Type	Description	Default
`historique`	`DataFrame`	DataFrame avec au minimum les colonnes 'date' (datetime ou str) et 'price' (float, en XOF/kg). Typiquement retourné par `fetch_prices()`.	required
`min_observations_par_mois`	`int`	Nombre minimal d'observations pour qu'un mois soit inclus dans le calcul. Les mois en dessous de ce seuil sont marqués NaN. Défaut : 2.	`2`

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant les champs suivants : `indices` (dict[int, float \| None]) : dictionnaire des 12 indices saisonniers, indexé par numéro de mois (1 à 12). La valeur est None si le mois a moins de `min_observations_par_mois` entrées. `mois_pic` (list[int]) : liste des mois où l'indice dépasse 1.05 (5% au-dessus de la moyenne), triés par indice décroissant. `mois_creux` (list[int]) : liste des mois où l'indice est en dessous de 0.95, triés par indice croissant. `prix_moyen_global` (float) : prix moyen sur toute la période historique, en XOF/kg. `prix_moyen_par_mois` (dict[int, float \| None]) : prix moyen brut par mois, avant normalisation. `nb_observations` (int) : nombre total d'observations valides utilisées pour le calcul. `nb_mois_couverts` (int) : nombre de mois avec au moins `min_observations_par_mois` entrées. `confiance` (float) : score de confiance de 0 à 1. Reflète la densité des données (1.0 = 2+ ans de données hebdomadaires, 0.0 = moins d'un mois de données). `sim` (bool) : True si l'historique source contient des données simulées. `message` (str \| None) : avertissement si les données sont insuffisantes pour un calcul fiable. None sinon.

Raises:

Type	Description
`ValueError`	Si l'historique est vide ou ne contient pas les colonnes 'date' et 'price'.

Source code in kadi/market/pricing.py

def seasonality(
    self,
    historique: pd.DataFrame,
    min_observations_par_mois: int = 2,
) -> dict:
    """
    Calcule l'indice saisonnier mensuel des prix agricoles.

    La méthode utilise la décomposition par ratios : pour chaque mois,
    l'indice est le rapport entre le prix moyen de ce mois et le prix
    moyen global sur toute la période. Un indice supérieur à 1 indique
    un mois de prix élevés (pénurie), inférieur à 1 un mois bon marché
    (période post-récolte).

    L'historique doit couvrir au moins 12 mois pour que les indices
    soient fiables. En dessous de ce seuil, les indices sont calculés
    mais le champ ``confiance`` sera faible.

    Args:
        historique (pd.DataFrame): DataFrame avec au minimum les colonnes
            'date' (datetime ou str) et 'price' (float, en XOF/kg).
            Typiquement retourné par ``fetch_prices()``.
        min_observations_par_mois (int, optional): Nombre minimal
            d'observations pour qu'un mois soit inclus dans le calcul.
            Les mois en dessous de ce seuil sont marqués NaN.
            Défaut : 2.

    Returns:
        dict: Dictionnaire contenant les champs suivants :

            - ``indices`` (dict[int, float | None]) : dictionnaire des
              12 indices saisonniers, indexé par numéro de mois (1 à 12).
              La valeur est None si le mois a moins de
              ``min_observations_par_mois`` entrées.
            - ``mois_pic`` (list[int]) : liste des mois où l'indice
              dépasse 1.05 (5% au-dessus de la moyenne), triés par
              indice décroissant.
            - ``mois_creux`` (list[int]) : liste des mois où l'indice
              est en dessous de 0.95, triés par indice croissant.
            - ``prix_moyen_global`` (float) : prix moyen sur toute la
              période historique, en XOF/kg.
            - ``prix_moyen_par_mois`` (dict[int, float | None]) :
              prix moyen brut par mois, avant normalisation.
            - ``nb_observations`` (int) : nombre total d'observations
              valides utilisées pour le calcul.
            - ``nb_mois_couverts`` (int) : nombre de mois avec au moins
              ``min_observations_par_mois`` entrées.
            - ``confiance`` (float) : score de confiance de 0 à 1.
              Reflète la densité des données (1.0 = 2+ ans de données
              hebdomadaires, 0.0 = moins d'un mois de données).
            - ``sim`` (bool) : True si l'historique source
              contient des données simulées.
            - ``message`` (str | None) : avertissement si les données
              sont insuffisantes pour un calcul fiable. None sinon.

    Raises:
        ValueError: Si l'historique est vide ou ne contient pas les
            colonnes 'date' et 'price'.
    """
    # --- Validation des entrées ---
    if historique is None or historique.empty:
        raise ValueError(
            "L'historique fourni est vide. "
            "Utilisez fetch_prices() pour obtenir des données avant "
            "d'appeler seasonality()."
        )

    colonnes_requises = {"date", "price"}
    if not colonnes_requises.issubset(historique.columns):
        raise ValueError(
            f"L'historique doit contenir les colonnes {colonnes_requises}. "
            f"Colonnes reçues : {set(historique.columns)}."
        )

    # --- Préparation du DataFrame ---
    # Copie pour ne pas modifier le DataFrame d'entrée
    df = historique[["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 ou invalides
    df = df.dropna(subset=["date", "price"])
    df = df[df["price"] > 0]

    if df.empty:
        raise ValueError(
            "Aucune observation valide après nettoyage de l'historique "
            "(vérifiez les colonnes 'date' et 'price')."
        )

    # Extraction du numéro de mois (1 = janvier, 12 = décembre)
    df["mois"] = df["date"].dt.month

    # --- Calcul des statistiques par mois ---
    # Comptage des observations disponibles par mois
    compte_par_mois = df.groupby("mois")["price"].count()

    # Prix moyen brut par mois (tous mois confondus)
    moyenne_par_mois = df.groupby("mois")["price"].mean()

    # Prix moyen global sur toute la période (référence de normalisation)
    prix_moyen_global = float(df["price"].mean())

    # Noms des mois en français pour les messages lisibles
    _NOMS_MOIS = {
        1: "janvier", 2: "février", 3: "mars", 4: "avril",
        5: "mai", 6: "juin", 7: "juillet", 8: "août",
        9: "septembre", 10: "octobre", 11: "novembre", 12: "décembre",
    }

    # --- Construction des indices saisonniers ---
    indices = {}
    prix_moyen_par_mois = {}

    for mois in range(1, 13):
        nb_obs = int(compte_par_mois.get(mois, 0))
        prix_moyen_par_mois[mois] = (
            round(float(moyenne_par_mois[mois]), 2)
            if mois in moyenne_par_mois.index
            else None
        )

        if nb_obs >= min_observations_par_mois and prix_moyen_global > 0:
            # Indice = ratio entre prix moyen du mois et prix moyen global
            # Indice > 1 : mois cher, < 1 : mois bon marché
            indice = float(moyenne_par_mois[mois]) / prix_moyen_global
            indices[mois] = round(indice, 4)
        else:
            # Données insuffisantes pour ce mois
            indices[mois] = None

    # --- Identification des mois de pic et de creux ---
    # Seuil de 5% au-dessus/en dessous de la moyenne pour éviter le bruit
    _SEUIL_PIC = 1.05
    _SEUIL_CREUX = 0.95

    mois_pic = sorted(
        [m for m, idx in indices.items() if idx is not None and idx >= _SEUIL_PIC],
        key=lambda m: indices[m],
        reverse=True,
    )
    mois_creux = sorted(
        [m for m, idx in indices.items() if idx is not None and idx <= _SEUIL_CREUX],
        key=lambda m: indices[m],
    )

    # --- Calcul du score de confiance ---
    nb_mois_couverts = sum(1 for v in indices.values() if v is not None)
    nb_observations = len(df)

    # La confiance dépend de deux facteurs :
    # 1. La couverture mensuelle : 12 mois = 1.0, 0 mois = 0.0
    facteur_couverture = nb_mois_couverts / 12.0
    # 2. La densité des données : on considère 104 obs (2 ans hebdo) comme optimal
    facteur_densite = min(1.0, nb_observations / 104.0)
    confiance = round((facteur_couverture + facteur_densite) / 2.0, 3)

    # --- Propagation du flag sim depuis la source ---
    est_simule = False
    if "sim" in historique.columns:
        est_simule = bool(historique["sim"].any())
    elif "is_simulated" in historique.columns:
        est_simule = bool(historique["is_simulated"].any())

    # --- Message d'avertissement si données insuffisantes ---
    message = None
    if nb_mois_couverts < 6:
        message = (
            f"Seulement {nb_mois_couverts} mois couverts sur 12. "
            "Les indices saisonniers calculés sont peu fiables. "
            "Il est recommandé d'avoir au moins 12 mois d'historique."
        )
    elif nb_mois_couverts < 12:
        manquants = [_NOMS_MOIS[m] for m in range(1, 13) if indices[m] is None]
        message = (
            f"Données manquantes pour : {', '.join(manquants)}. "
            "L'indice saisonnier est None pour ces mois."
        )

    if est_simule:
        avert_simul = "Attention : l'historique est simulé. Les indices ne reflètent pas la réalité du marché."
        message = f"{avert_simul} {message}" if message else avert_simul

    logger.info(
        f"Saisonnalité calculée : {nb_mois_couverts}/12 mois couverts, "
        f"{nb_observations} observations, confiance={confiance}, "
        f"pic={mois_pic}, creux={mois_creux}."
    )

    return {
        "indices": indices,
        "mois_pic": mois_pic,
        "mois_creux": mois_creux,
        "prix_moyen_global": round(prix_moyen_global, 2),
        "prix_moyen_par_mois": prix_moyen_par_mois,
        "nb_observations": nb_observations,
        "nb_mois_couverts": nb_mois_couverts,
        "confiance": confiance,
        "sim": est_simule,
        "is_simulated": est_simule,
        "message": message,
    }

Pricing (kadi.market.pricing)¶

Direct Importing¶

Initialization¶

Methods¶

fetch(crop, market, days, sim)¶

convert_unit(value, unit, crop)¶

anomalies(series, z)¶

fill_gaps(series, max_gap)¶

source(series)¶

seasonality(historique, min_observations_par_mois)¶

Deprecated Methods¶

Pricing ¶

__init__ ¶

__getattr__ ¶

fetch ¶

convert_unit ¶

anomalies ¶

fill_gaps ¶

source ¶

seasonality ¶

Pricing (`kadi.market.pricing`)¶

`fetch(crop, market, days, sim)`¶

`convert_unit(value, unit, crop)`¶

`anomalies(series, z)`¶

`fill_gaps(series, max_gap)`¶

`source(series)`¶

`seasonality(historique, min_observations_par_mois)`¶

init ¶

getattr ¶