Tarification (`kadi.market.pricing`)¶

La classe Pricing gère l'acquisition, la normalisation et l'analyse des données de prix de marché agricole au Bénin. Elle est utilisée en interne par la façade Market, mais peut aussi être instanciée directement.

Importation directe¶

from kadi.market import Pricing

Initialisation¶

Pricing est créée automatiquement par la façade Market et est accessible via market.pricing. Pour une instanciation directe :

from kadi.market import Pricing

# Sans client (mode simulation automatique)
pricing = Pricing()

# Avec client WFP et taux de change dynamiques
from kadi._sources import WFPClient, ExchangeRateClient
pricing = Pricing(
    wfp=WFPClient(),
    exchange=ExchangeRateClient(),
    sim=False,
)

Paramètres :

Nom	Type	Défaut	Description
`wfp`	`WFPClient`	`None`	Client WFP pour l'API HAPI HumData
`exchange`	`ExchangeRateClient`	`None`	Client de taux de change dynamiques
`sim`	`bool`	`False`	Si True, force le mode simulation pour toutes les méthodes

Attributs publics :

Attribut	Type	Description
`client`	`WFPClient`	Client WFP injecté (ou None)
`sim`	`bool`	Mode simulation actif

Méthodes¶

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

Récupère les prix historiques d'une culture sur un marché.

En mode réel, interroge l'API HAPI HumData. En mode simulation (aucun client configuré ou sim=True), génère des prix fictifs par distribution normale centrée sur 300 XOF/kg, clairement marqués sim=True et confidence_score=0.1.

import kadi as kd

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

# Récupération de l'historique (données réelles ou simulées)
df = marche.pricing.fetch("maize", "parakou", days=365)

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

Paramètres :

Nom	Type	Défaut	Description
`crop`	`str`	requis	Code de la culture (ex: `'maize'`, `'rice'`)
`market`	`str`	requis	Nom normalisé du marché (ex: `'cotonou'`)
`days`	`int`	`365`	Nombre de jours d'historique
`sim`	`bool`	`None`	Surcharge le mode de l'instance

Retour : pd.DataFrame

Colonne	Type	Description
`date`	`datetime`	Date de l'observation
`price`	`float`	Prix en XOF/kg
`unit`	`str`	Unité d'origine (ex: `'XOF/kg'`)
`sim`	`bool`	True si donnée fictive
`source`	`str`	Identifiant de la source
`fetched_at`	`str`	Horodatage ISO 8601 de la collecte
`confidence_score`	`float`	Score de confiance (0.0 a 1.0)

`convert_unit(value, unit, crop)`¶

Convertit une valeur de prix vers l'unité standard XOF/kg.

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

# USD/kg -> XOF/kg (taux de change Frankfurter)
prix_kg = marche.pricing.convert_unit(0.45, "USD/kg")

# XOF/sac (poids depend de la culture)
prix_kg = marche.pricing.convert_unit(14_000.0, "XOF/sac", crop="maize")

Paramètres :

Nom	Type	Défaut	Description
`value`	`float`	requis	Valeur du prix à convertir
`unit`	`str`	requis	Unité d'origine
`crop`	`str`	`None`	Code de la culture (pour les poids de contenants)

Unités supportées :

Unité	Conversion
`XOF/kg`	Pas de conversion (unité cible)
`XOF/Tonne`	Division par 1000
`USD/kg`	Multiplication par le taux USD/XOF
`EUR/kg`	Multiplication par le taux EUR/XOF (fixe UEMOA)
`XOF/sac`	Division par le poids du sac (culture-dépendant)
`XOF/boisseau`, `XOF/tine`, `XOF/caisse`	Division par le poids du contenant

Si l'unité est inconnue, la valeur est retournée sans modification.

Retour : float - Prix en XOF/kg.

`anomalies(series, z)`¶

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

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

nb = df["is_anomaly"].sum()
print(f"Anomalies détectées : {nb}")
print(df[df["is_anomaly"]][["date", "price"]])

Paramètres :

Nom	Type	Défaut	Description
`series`	`pd.DataFrame`	requis	DataFrame avec une colonne `'price'`
`z`	`float`	`3.0`	Seuil du Z-score (3 = 99.7% de la distribution)

Retour : pd.DataFrame - DataFrame d'entrée avec une colonne booléenne is_anomaly ajoutée.

`fill_gaps(series, max_gap)`¶

Comble les valeurs manquantes dans une série de prix par interpolation linéaire, limitée à un nombre configurable de jours consécutifs.

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

Paramètres :

Nom	Type	Défaut	Description
`series`	`pd.DataFrame`	requis	DataFrame avec une colonne `'price'`
`max_gap`	`int`	`7`	Nombre maximum de jours à interpoler

Les trous de plus de max_gap jours consécutifs restent NaN.

Retour : pd.DataFrame - DataFrame avec les trous courts comblés.

`source(series)`¶

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

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

Paramètres :

Nom	Type	Description
`series`	`pd.DataFrame`	DataFrame retourné par `fetch()`

Retour : str - 'wfp-vam', 'ratin', 'scrape-local' ou 'simulated'.

`seasonality(historique, min_observations_par_mois)`¶

Calcule les 12 indices saisonniers mensuels des prix agricoles par la méthode des ratios.

Un indice supérieur à 1 indique un mois de prix élevés (période de soudure). Inférieur à 1 indique un mois bon marché (période post-récolte).

df = marche.pricing.fetch("maize", "parakou", days=730)
saison = marche.pricing.seasonality(df)

print(f"Mois de pic   : {saison['mois_pic']}")
print(f"Mois de creux : {saison['mois_creux']}")
print(f"Prix moyen    : {saison['prix_moyen_global']:.2f} XOF/kg")
print(f"Confiance     : {saison['confiance']:.2f}")

# Indice par mois
for mois, indice in saison["indices"].items():
    if indice:
        print(f"  Mois {mois:2d} : {indice:.3f}")

Paramètres :

Nom	Type	Défaut	Description
`historique`	`pd.DataFrame`	requis	DataFrame avec colonnes `'date'` et `'price'`
`min_observations_par_mois`	`int`	`2`	Observations minimales pour inclure un mois

Retour : dict

Clé	Type	Description
`indices`	`dict[int, float]`	Indices saisonniers par mois (1 a 12). None si données insuffisantes
`mois_pic`	`list[int]`	Mois dont l'indice depasse 1.05 (5% au-dessus de la moyenne)
`mois_creux`	`list[int]`	Mois dont l'indice est sous 0.95
`prix_moyen_global`	`float`	Prix moyen sur toute la periode (XOF/kg)
`prix_moyen_par_mois`	`dict[int, float]`	Prix brut moyen par mois
`nb_observations`	`int`	Nombre total d'observations valides
`nb_mois_couverts`	`int`	Mois avec au moins `min_observations_par_mois` entrees
`confiance`	`float`	Score de confiance (0.0 a 1.0)
`sim`	`bool`	True si l'historique contient des donnees simulees
`message`	`str`	Avertissement si donnees insuffisantes. None sinon

Exceptions :

ValueError : historique vide, ou colonnes 'date' / 'price' absentes.

Méthodes dépréciées¶

Ancienne méthode	Remplacée par	Depuis
`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,
    }

Tarification (kadi.market.pricing)¶

Importation directe¶

Initialisation¶

Méthodes¶

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)¶

Méthodes dépréciées¶

Pricing ¶

__init__ ¶

__getattr__ ¶

fetch ¶

convert_unit ¶

anomalies ¶

fill_gaps ¶

source ¶

seasonality ¶

Tarification (`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 ¶