Aide à la décision (`kadi.market.decision_support`)¶

La classe Advisor convertit les prévisions de prix et les données de marché en recommandations opérationnelles : arbitrage spatial entre marchés, décision de stockage et optimisation du portefeuille de cultures.

Elle est utilisée en interne par la façade Market et accessible via market.advisor. Elle peut aussi être instanciée directement.

Importation directe¶

from kadi.market import Advisor

Initialisation¶

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

from kadi.market import Advisor, Forecasting, Logistics, Pricing

advisor = Advisor(
    forecast=Forecasting(),
    logistics=Logistics(),
    pricing=Pricing(),
)

Paramètres :

Nom	Type	Défaut	Description
`forecast`	`Forecasting`	`None`	Instance de prévision pour estimer les prix futurs
`logistics`	`Logistics`	`None`	Instance logistique pour les coûts de transport
`pricing`	`Pricing`	`None`	Instance de tarification pour les prix réels

Si pricing est None, les méthodes utilisent un prix de repli de 300 XOF/kg avec is_simulated=True et confidence_score=0.0.

Attributs publics :

Attribut	Type	Description
`forecast`	`Forecasting`	Module de prévision de prix
`logistics`	`Logistics`	Module logistique
`pricing`	`Pricing`	Module de tarification

Méthodes¶

`arbitrage(crop, m_from, to, qty)`¶

Évalue la rentabilité d'un transfert physique de marchandises entre deux marchés.

Formule :

Gain net = (prix_destination - prix_origine) * 1000 * qty - cout_transport

Le seuil de rentabilité minimum est configurable dans config.py (clé logistics.seuil_rentabilite_pct, défaut : 10%).

import kadi as kd

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

# Est-il rentable de transporter 10 t de maïs de Parakou à Cotonou ?
decision = marche.advisor.arbitrage(
    crop="maize",
    m_from="Parakou",
    to="Cotonou",
    qty=10.0,
)

print(decision["recommandation"])           # 'TRANSPORTER' ou 'NE PAS TRANSPORTER'
print(f"Gain net total : {decision['gain_net_total_cfa']:,.0f} XOF")
print(f"Gain net       : {decision['gain_net_percent']:.1f} %")
print(f"Transport      : {decision['frais_logistiques_total']:,.0f} XOF")
print(f"Pluie prévue   : {decision['prob_pluie'] * 100:.0f} %")
print(f"Confiance      : {decision['confidence_score']:.2f}")

Paramètres :

Nom	Type	Défaut	Description
`crop`	`str`	requis	Code de la culture (ex: `'maize'`)
`m_from`	`str`	`'Parakou'`	Marché d'achat
`to`	`str`	`'Cotonou'`	Marché de vente
`qty`	`float`	`1.0`	Quantité à transporter en tonnes métriques

Retour : dict

Clé	Type	Description
`recommandation`	`str`	`'TRANSPORTER'` ou `'NE PAS TRANSPORTER'`
`gain_net_total_cfa`	`float`	Gain net total en XOF
`gain_net_percent`	`float`	Gain net en % du capital investi
`frais_logistiques_total`	`float`	Couts de transport totaux en XOF
`prix_origine_xof_kg`	`float`	Prix au marche d'achat en XOF/kg
`prix_destination_xof_kg`	`float`	Prix au marche de vente en XOF/kg
`is_simulated`	`bool`	True si les prix utilises sont fictifs
`confidence_score`	`float`	Score de confiance de la recommandation (0-1)
`prob_pluie`	`float`	Probabilite de pluie utilisee pour la logistique

Score de confiance :

Calculé comme une combinaison pondérée de trois facteurs :

score = 0.5 * price_confidence
      + 0.3 * (0 si simulé, 1 si réel)
      + 0.2 * min(1, |gain_net_pct| / 30)

Exceptions :

Aucune exception levée en cas d'absence de données : la méthode retourne un prix de repli de 300 XOF/kg avec is_simulated=True.

`store_sell(crop, market, price, qty, months)`¶

Évalue s'il est plus rentable de stocker une récolte ou de la vendre immédiatement.

Formule d'espérance de gain nette par tonne :

E = E[P(t+n)] - P(t) - C_stockage(n) - C_opportunite(n) - theta * Var(P)

Où : - C_stockage = 3 200 XOF/tonne/mois (gardiennage, pertes, sacs) - C_opportunite = 1.5%/mois du prix actuel (immobilisation de trésorerie) - theta = 0.04 (coefficient d'aversion au risque)

# Stocker 5 t d'igname pendant 3 mois ou vendre maintenant ?
stockage = marche.advisor.store_sell(
    crop="yam",
    market="Abomey",
    price=250_000.0,    # Prix actuel en XOF/tonne
    qty=5.0,
    months=3,
)

print(stockage["recommandation_binaire"])   # 'STOCKER' ou 'VENDRE IMMÉDIATEMENT'
print(f"Marge nette totale   : {stockage['marge_nette_cfa']:,.0f} XOF")
print(f"Marge par tonne      : {stockage['marge_nette_par_tonne']:,.0f} XOF")
print(f"Prix futur estimé    : {stockage['prix_futur_estime']:,.0f} XOF/tonne")
print(f"Horizon              : {stockage['horizon_mois']} mois")
print(f"Confiance            : {stockage['confidence_score']:.2f}")

Paramètres :

Nom	Type	Défaut	Description
`crop`	`str`	requis	Code de la culture
`market`	`str`	`'Parakou'`	Marche de référence pour la prévision
`price`	`float`	`300000.0`	Prix actuel en XOF par tonne
`qty`	`float`	`1.0`	Quantité en tonnes
`months`	`int`	`None`	Horizon de stockage en mois. Si None, lit la valeur de config.py (défaut : 3 mois)

Retour : dict

Clé	Type	Description
`recommandation_binaire`	`str`	`'STOCKER'` ou `'VENDRE IMMEDIATEMENT'`
`marge_nette_cfa`	`float`	Esperance de gain total en XOF
`marge_nette_par_tonne`	`float`	Esperance de gain par tonne
`prix_futur_estime`	`float`	Prix prevu a l'horizon (XOF/tonne)
`horizon_mois`	`int`	Horizon de stockage utilise
`is_simulated`	`bool`	True si les donnees de prevision sont simulees
`confidence_score`	`float`	Score de confiance (0-1)

Si aucun module forecast n'est disponible, la méthode applique une hypothèse conservative de hausse de 15% sur le prix actuel.

`optimize(land_ha, climate, market, yields)`¶

Optimise la répartition des cultures sur la surface disponible.

Utilise scipy.optimize.linprog (méthode HiGHS) pour maximiser le revenu attendu sous contraintes de surface et de diversification. Si scipy n'est pas disponible, un fallback heuristique est appliqué.

Modèle d'optimisation :

Maximiser : sum(prix_i * rendement_i * x_i)
Sous       : sum(x_i) <= land_ha
             0 <= x_i <= 0.7 * land_ha  (diversification minimale 30%)

Ajustements climatiques (clé drought_severity) :

Severite	Effet
`'severe'`	rendement niébé * 1.3, maïs et riz * 0.7
`'moderate'`	rendement maïs * 0.85
`'mild'` ou `'no_drought'`	pas d'ajustement

# Optimisation avec conditions climatiques modérées
decision_port = marche.advisor.optimize(
    land_ha=10.0,
    climate={"drought_severity": "mild"},
    market={"maize": 285.0, "cowpea": 580.0, "sorghum": 210.0},
)

print(f"Méthode        : {decision_port['methode']}")
print(f"Revenu attendu : {decision_port['revenu_attendu_cfa']:,.0f} XOF")
print(f"Confiance      : {decision_port['confidence_score']:.2f}")
for culture, ha in decision_port["repartition_hectares"].items():
    print(f"  {culture} : {ha:.1f} ha")

# Cas de sécheresse sévère (favorise le niébé)
decision_sec = marche.advisor.optimize(
    land_ha=5.0,
    climate={"drought_severity": "severe"},
    market={"maize": 285.0, "cowpea": 620.0, "sorghum": 210.0},
)

Paramètres :

Nom	Type	Défaut	Description
`land_ha`	`float`	`1.0`	Surface arable disponible en hectares
`climate`	`dict`	`None`	Prévisions climatiques (voir tableau ci-dessous)
`market`	`dict`	`None`	Prix médians actuels par culture en XOF/kg
`yields`	`dict`	`None`	Rendements attendus en t/ha. Si None, utilise les rendements de référence FAO/INSAE Bénin

Clés de climate :

Clé	Type	Description
`drought_severity`	`str`	Severite : `'no_drought'`, `'mild'`, `'moderate'`, `'severe'`
`secheresse_anticipee`	`bool`	Rétrocompatibilité V1 : True = severity `'severe'`
`prob_pluie_7j`	`float`	Probabilite de pluie sur 7 jours

Rendements de référence (FAO/INSAE Bénin) :

Culture	Rendement (t/ha)
`maize`	1.8
`sorghum`	1.2
`millet`	1.0
`rice`	2.5
`cowpea`	0.7
`soybean`	1.2
`yam`	8.0
`cassava`	12.0

Retour : dict

Clé	Type	Description
`repartition_hectares`	`dict[str, float]`	Surface allouée par culture (ha)
`revenu_attendu_cfa`	`float`	Revenu total attendu en XOF
`recommandation`	`str`	Texte explicatif de la décision
`methode`	`str`	`'scipy_linprog'` ou `'heuristique'`
`confidence_score`	`float`	Score de confiance (0.75 conditions normales, 0.55 sécheresse, 0.3 heuristique)

Fallback heuristique :

Activé si scipy est absent ou si aucune culture n'a de prix connu. La répartition par défaut est : maïs 50%, soja 30%, niébé 20%. En cas de sécheresse sévère : maïs 30%, soja 30%, niébé 40%.

Méthodes dépréciées¶

Ancienne méthode	Remplacée par	Depuis
`arbitrage_decision(crop, origine, destination, qty_tons)`	`arbitrage(crop, m_from, to, qty)`	v1.2.0
`storage_vs_sell_now(crop, market, current_price, qty_tons, mois_stockage)`	`store_sell(crop, market, price, qty, months)`	v1.2.0
`portfolio_optimization(available_land_ha, climate_forecast, market_forecast)`	`optimize(land_ha, climate, market)`	v1.2.0

Classe dépréciée¶

Ancien nom	Remplacé par	Depuis
`DecisionSupport`	`Advisor`	v1.2.0

Advisor ¶

Classe convertissant les prévisions de prix et les données de marché en recommandations opérationnelles : arbitrage spatial, stockage, et optimisation de portefeuille de cultures.

Phase 4

Chaque recommandation inclut un 'confidence_score' (0.0 à 1.0) calculé à partir de la qualité des données de prix, du flag is_simulated et de la magnitude du gain estimé.

optimize() utilise scipy.optimize.linprog si disponible, avec un fallback heuristique si scipy est absent.

Source code in kadi/market/decision_support.py

class Advisor:
    """
    Classe convertissant les prévisions de prix et les données de marché
    en recommandations opérationnelles : arbitrage spatial, stockage,
    et optimisation de portefeuille de cultures.

    Phase 4 :
        Chaque recommandation inclut un 'confidence_score' (0.0 à 1.0)
        calculé à partir de la qualité des données de prix, du flag
        is_simulated et de la magnitude du gain estimé.

        optimize() utilise scipy.optimize.linprog si disponible,
        avec un fallback heuristique si scipy est absent.
    """

    def __init__(
        self,
        forecast=None,
        logistics=None,
        pricing=None,
    ):
        """
        Initialise le module d'aide à la décision.

        Args:
            forecast (Forecasting, optional): Instance de Forecasting pour les prévisions.
            logistics (Logistics, optional): Instance de Logistics pour les coûts de transport.
            pricing (Pricing, optional): Instance de Pricing pour les prix réels du marché.
                Si None, les méthodes utiliseront des prix estimés par défaut.
        """
        # Références vers les sous-modules
        self.forecast = forecast
        self.logistics = logistics
        self.pricing = pricing

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

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

        Args:
            name (str): Nom de la méthode demandée.

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

        Raises:
            AttributeError: Si le nom est inconnu.
        """
        import warnings
        if name in _DEPRECATED_METHODS:
            new_name = _DEPRECATED_METHODS[name]
            warnings.warn(
                f"Advisor.{name}() est obsolète et sera supprimé dans "
                f"KadiPy v2.0. Utilisez Advisor.{new_name}() à la place.",
                DeprecationWarning,
                stacklevel=2,
            )
            return getattr(self, new_name)
        raise AttributeError(f"'Advisor' n'a pas d'attribut '{name}'.")

    # ------------------------------------------------------------------
    # Méthodes privées : données de prix et score de confiance
    # ------------------------------------------------------------------

    def _get_price(self, crop: str, market: str) -> tuple:
        """
        Récupère le prix médian actuel pour une culture sur un marché donné.

        Interroge le module pricing pour obtenir les données réelles ou simulées.
        Retourne aussi le flag is_simulated et le confidence_score des données.

        Args:
            crop (str): Code de la culture (ex: 'maize').
            market (str): Nom normalisé du marché (ex: 'cotonou').

        Returns:
            tuple: (prix_median_xof_kg: float, is_simulated: bool, confidence: float)
        """
        if self.pricing is None:
            # Pas de module pricing : retour du prix de repli avec confiance nulle
            logger.warning(
                f"Pas de module pricing disponible pour {crop}/{market}. "
                "Prix de repli utilisé (300 XOF/kg)."
            )
            return 300.0, True, 0.0

        try:
            # Récupération des 30 derniers jours de données
            df = self.pricing.fetch(crop, market, days=30)

            if df.empty or "price" not in df.columns:
                return 300.0, True, 0.0

            # Calcul du prix médian (robuste aux outliers)
            prix_median = float(df["price"].median())

            # Flag de simulation : True si au moins une ligne est simulée
            est_simule = (
                bool(df["is_simulated"].any())
                if "is_simulated" in df.columns
                else True
            )

            # Score de confiance : median des scores individuels dans le DataFrame
            if "confidence_score" in df.columns:
                confiance = float(df["confidence_score"].median())
            else:
                confiance = 0.0 if est_simule else 0.8

            return prix_median, est_simule, confiance

        except Exception as e:
            logger.warning(
                f"Erreur lors de la récupération du prix {crop}/{market}: {e}. "
                "Prix de repli utilisé."
            )
            return 300.0, True, 0.0

    def _confidence(
        self,
        price_confidence: float,
        is_simulated: bool,
        gain_net_pct: float,
    ) -> float:
        """
        Calcule un score de confiance global sur une recommandation financière.

        Formule pondérée :
            score = 0.5 * price_confidence
                  + 0.3 * (0 si simulé, 1 si réel)
                  + 0.2 * min(1, |gain_net_pct| / 30)

        Le troisième terme représente l'idée qu'une recommandation avec un
        gain de +30% ou plus est plus convaincante qu'un gain de +1%.

        Args:
            price_confidence (float): Score de confiance des données de prix (0 à 1).
            is_simulated (bool): True si les données sont fictives.
            gain_net_pct (float): Gain net en % du capital investi (peut être négatif).

        Returns:
            float: Score de confiance global entre 0.0 et 1.0.
        """
        # Composante 1 : confiance des données de prix
        score_prix = 0.5 * max(0.0, min(1.0, price_confidence))

        # Composante 2 : données réelles vs simulées
        score_source = 0.3 * (0.0 if is_simulated else 1.0)

        # Composante 3 : magnitude relative du gain (30% = confiance max)
        score_magnitude = 0.2 * min(1.0, abs(gain_net_pct) / 30.0)

        score = score_prix + score_source + score_magnitude
        return round(max(0.0, min(1.0, score)), 3)

    # ------------------------------------------------------------------
    # API publique : recommandations de marché
    # ------------------------------------------------------------------

    def arbitrage(
        self,
        crop: str,
        m_from: str = "Parakou",
        to: str = "Cotonou",
        qty: float = 1.0,
        **kwargs,
    ) -> dict:
        """
        Évalue la rentabilité d'un transfert physique de marchandises entre deux marchés.

        Utilise les prix réels du marché (via pricing) pour calculer
        la marge brute. Les coûts logistiques (avec ajustement météo si disponible)
        sont calculés par le module logistics.

        Formule :
            Gain net = (prix_destination - prix_origine) * 1000 * qty - cout_transport

        Args:
            crop (str): La culture concernée (ex: 'maize').
            m_from (str): Le marché d'achat (ex: 'Parakou').
            to (str): Le marché de vente (ex: 'Cotonou').
            qty (float): La quantité à transporter en tonnes métriques.
            **kwargs: Anciens nom de paramètres (market_from, market_to, qty_tons).

        Returns:
            dict: Dictionnaire contenant :
                - 'recommandation'       : 'TRANSPORTER' ou 'NE PAS TRANSPORTER'
                - 'gain_net_total_cfa'   : gain net total en XOF
                - 'gain_net_percent'     : gain net en % du capital investi
                - 'frais_logistiques_total' : coûts de transport totaux en XOF
                - 'prix_origine_xof_kg'  : prix au marché d'achat
                - 'prix_destination_xof_kg' : prix au marché de vente
                - 'is_simulated'         : True si les prix utilisés sont fictifs
                - 'confidence_score'     : score de confiance de la recommandation (0-1)
                - 'prob_pluie'           : probabilité de pluie utilisée (0 si sans météo)
        """
        if "market_from" in kwargs:
            m_from = kwargs.pop("market_from")
        if "market_to" in kwargs:
            to = kwargs.pop("market_to")
        if "qty_tons" in kwargs:
            qty = kwargs.pop("qty_tons")

        # Récupération des prix réels (ou simulés) pour les deux marchés
        prix_origine_kg, simule_origine, conf_origine = self._get_price(
            crop, m_from
        )
        prix_destination_kg, simule_destination, conf_destination = self._get_price(
            crop, to
        )

        # Les données sont simulées si l'une des deux sources l'est
        est_simule = simule_origine or simule_destination

        # Confiance combinée sur les prix : moyenne des deux marchés
        price_confidence = (conf_origine + conf_destination) / 2.0

        # Marge brute par tonne (conversion kg -> tonne : *1000)
        marge_brute_tonne = (prix_destination_kg - prix_origine_kg) * 1000

        # Calcul des frais logistiques via le module dédié
        prob_pluie = 0.0
        if self.logistics:
            # On passe le crop pour activer la perte de qualité par culture
            res_logistics = self.logistics.transfer_cost(
                m_from, to, crop=crop
            )
            frais_logistiques_tonne = res_logistics["total_cost_cfa"]
            prob_pluie = res_logistics.get("prob_pluie", 0.0)
        else:
            frais_logistiques_tonne = 30000.0
            logger.warning(
                "Pas de module logistics. Frais de transport de repli : 30 000 XOF/tonne."
            )

        # Gain net par tonne et total sur la quantité transportée
        gain_net_tonne = marge_brute_tonne - frais_logistiques_tonne
        gain_net_total = gain_net_tonne * qty

        # Gain net en % du capital investi (achat + transport)
        cout_total = (prix_origine_kg * 1000 + frais_logistiques_tonne) * qty
        gain_net_pct = (gain_net_total / cout_total * 100) if cout_total > 0 else 0.0

        # Recommandation basée sur le seuil de rentabilité configuré
        recommandation = (
            "TRANSPORTER" if gain_net_pct >= _SEUIL_RENTABILITE else "NE PAS TRANSPORTER"
        )

        # Score de confiance global sur la recommandation
        confidence_score = self._confidence(
            price_confidence, est_simule, gain_net_pct
        )

        return {
            "recommandation": recommandation,
            "gain_net_total_cfa": round(gain_net_total, 2),
            "gain_net_percent": round(gain_net_pct, 2),
            "frais_logistiques_total": round(frais_logistiques_tonne * qty, 2),
            "prix_origine_xof_kg": round(prix_origine_kg, 2),
            "prix_destination_xof_kg": round(prix_destination_kg, 2),
            "is_simulated": est_simule,
            "confidence_score": confidence_score,
            "prob_pluie": round(prob_pluie, 3),
        }

    def store_sell(
        self,
        crop: str,
        market: str = "Parakou",
        price: float = 300000.0,
        qty: float = 1.0,
        months: int = None,
        **kwargs,
    ) -> dict:
        """
        Évalue s'il est plus rentable de stocker ou de vendre immédiatement.

        Phase 4 : l'horizon de stockage est maintenant configurable via le
        paramètre months. La valeur par défaut est lue depuis config.py
        (clé de configuration horizon stockage mois, défaut : 3 mois).

        Formule :
            E = E[P_{t+n}] - P_t - C_stockage(n) - C_opportunite(n) - theta * Var(P)

        Args:
            crop (str): La culture concernée.
            market (str): Le marché de référence.
            price (float): Prix actuel en XOF par tonne.
            qty (float): Quantité récoltée en tonnes.
            months (int, optional): Horizon de stockage en mois.
                Si None, utilise la valeur par defaut configuree (3 mois).

        Returns:
            dict: Dictionnaire contenant :
                - 'recommandation_binaire' : 'STOCKER' ou 'VENDRE IMMÉDIATEMENT'
                - 'marge_nette_cfa'        : espérance de gain total en XOF
                - 'marge_nette_par_tonne'  : espérance de gain par tonne
                - 'prix_futur_estime'      : prix prévu à l'horizon (XOF/tonne)
                - 'horizon_mois'           : horizon de stockage utilisé
                - 'is_simulated'           : True si les données de prévision sont
                  simulées (mode offline ou stub). Propagé depuis predict_price().
                  Vaut True par défaut si aucun module de prévision n'est disponible.
                - 'confidence_score'       : score de confiance (0-1)
        """
        if "current_price" in kwargs:
            price = kwargs.pop("current_price")
        if "qty_tons" in kwargs:
            qty = kwargs.pop("qty_tons")
        if "storage_months" in kwargs:
            months = kwargs.pop("storage_months")
        if "mois_stockage" in kwargs:
            months = kwargs.pop("mois_stockage")

        # Lecture de l'horizon de stockage (paramètre > config > défaut)
        if months is None:
            months = _HORIZON_MOIS_DEFAULT
        jours_stockage = months * 30

        # Valeur par défaut : True si aucun module de prévision n'est disponible
        est_simule = True

        # Récupération du prix futur estimé via le module de prévision
        if self.forecast:
            try:
                prevision = self.forecast.predict(
                    crop, market, ahead=jours_stockage
                )
                # La prévision est en XOF/kg ; on convertit en XOF/tonne
                prix_futur_tonne = prevision["predicted_price"] * 1000.0
                rmse_val = prevision.get("rmse")
                if rmse_val is not None:
                    variance = rmse_val * 1000.0
                else:
                    variance = prevision["predicted_price"] * 0.05 * 1000.0
                # Propagation du flag réel retourné par predict().
                # Par défaut True si la clé est absente (comportement offline conservé).
                est_simule = prevision.get("is_simulated", True)
            except Exception as e:
                logger.warning(
                    f"Erreur lors de la prévision {crop}/{market}: {e}. "
                    "Hypothèse de hausse de 15%."
                )
                prix_futur_tonne = price * 1.15
                variance = price * 0.05
        else:
            # Sans module de prévision : hypothèse conservative de hausse de 15%
            prix_futur_tonne = price * 1.15
            variance = price * 0.05

        # Coûts de stockage : gardiennage, pertes, sacs (par mois par tonne)
        cout_stockage_mensuel_tonne = 3200.0
        c_stockage = cout_stockage_mensuel_tonne * months

        # Coût d'opportunité : immobilisation de la trésorerie à 1.5%/mois
        taux_opportunite_mensuel = 0.015
        c_opportunite = price * (taux_opportunite_mensuel * months)

        # Pénalité de risque (aversion au risque theta)
        theta_risque = 0.04
        penalite_risque = theta_risque * variance

        # Espérance de gain net par tonne
        esperance_gain = (
            prix_futur_tonne
            - price
            - c_stockage
            - c_opportunite
            - penalite_risque
        )

        recommandation = "STOCKER" if esperance_gain > 0 else "VENDRE IMMÉDIATEMENT"

        # Gain net en % du capital actuel (pour le score de confiance)
        gain_pct = (esperance_gain / price * 100) if price > 0 else 0.0

        # Score de confiance (prix simulés = confiance faible)
        confidence_score = self._confidence(
            price_confidence=0.0 if est_simule else 0.7,
            is_simulated=est_simule,
            gain_net_pct=gain_pct,
        )

        return {
            "recommandation_binaire": recommandation,
            "marge_nette_cfa": round(esperance_gain * qty, 2),
            "marge_nette_par_tonne": round(esperance_gain, 2),
            "prix_futur_estime": round(prix_futur_tonne, 2),
            "horizon_mois": months,
            "is_simulated": est_simule,
            "confidence_score": confidence_score,
        }

    def optimize(
        self,
        land_ha: float = 1.0,
        climate: dict = None,
        market: dict = None,
        yields: dict = None,
        **kwargs,
    ) -> dict:
        """
        Optimise la répartition des cultures sur la surface disponible.

        Phase 4 : utilise scipy.optimize.linprog pour maximiser le revenu
        attendu sous contraintes agronomiques et de surface. Si scipy n'est
        pas disponible, utilise un fallback heuristique.

        Modèle d'optimisation (programmation linéaire) :
            - Variables : x_i = hectares alloués à la culture i
            - Objectif  : maximiser sum(prix_i * rendement_i * x_i)
            - Contraintes :
                * sum(x_i) <= land_ha  (surface totale)
                * x_i >= 0
                * x_i <= 0.7 * land_ha  (diversification minimale)

        Ajustements climatiques (SPI de sécheresse) :
            - Sécheresse sévère (drought_severity='severe') :
                rendement_niébé * 1.3 (résistant), rendement_maïs * 0.7
            - Sécheresse modérée ('moderate') :
                rendement_maïs * 0.85

        Args:
            land_ha (float): Surface arable disponible en hectares.
            climate (dict): Prévisions climatiques. Clés attendues :
                - 'secheresse_anticipee' (bool) : compatibilité V1
                - 'drought_severity' (str)      : 'no_drought', 'mild', 'moderate', 'severe'
                - 'prob_pluie_7j' (float)       : probabilité de pluie sur 7 jours
            market (dict): Prix médians actuels par culture (XOF/kg).
                Exemple : {'maize': 285, 'cowpea': 580, 'sorghum': 210}
            yields (dict, optional): Rendements attendus en tonnes/hectare.
                Si None, utilise les rendements de référence béninois.
            **kwargs: Anciens arguments (available_land_ha, climate_forecast, market_forecast).

        Returns:
            dict: Dictionnaire contenant :
                - 'repartition_hectares' : répartition optimale par culture (hectares)
                - 'revenu_attendu_cfa'   : revenu total attendu en XOF
                - 'recommandation'       : texte explicatif
                - 'methode'              : 'scipy_linprog' ou 'heuristique'
                - 'confidence_score'     : score de confiance de l'optimisation
        """

        if "available_land_ha" in kwargs:
            land_ha = kwargs.pop("available_land_ha")
        if "climate_forecast" in kwargs:
            climate = kwargs.pop("climate_forecast")
        if "market_forecast" in kwargs:
            market = kwargs.pop("market_forecast")

        if climate is None:
            climate = {}
        if market is None:
            market = {}

        # Rendements de référence (paramètre > défauts Bénin)
        rends = dict(_RENDEMENTS_BENIN)
        if yields:
            rends.update(yields)

        # Cultures à optimiser : celles pour lesquelles on a un prix de marché
        # On filtre pour ne garder que les cultures connues dans les rendements
        cultures = [
            c for c in market
            if c in rends and market[c] > 0
        ]

        if not cultures:
            # Aucune culture avec données : fallback heuristique complet
            return self._heuristic(
                land_ha, climate, market
            )

        # Ajustement des rendements selon la sévérité de la sécheresse
        severity = climate.get("drought_severity", "mild")
        if climate.get("secheresse_anticipee", False):
            # Rétrocompatibilité V1 : bool -> sécheresse sévère
            severity = "severe"

        rends_ajustes = {}
        for c in cultures:
            r = rends[c]
            if severity == "severe":
                # Sécheresse sévère : niébé résiste, maïs souffre
                if c == "cowpea":
                    r *= 1.3
                elif c in ("maize", "rice"):
                    r *= 0.7
            elif severity == "moderate":
                if c == "maize":
                    r *= 0.85
            rends_ajustes[c] = r

        # Revenus attendus par hectare pour chaque culture (XOF/ha)
        # prix en XOF/kg * rendement en t/ha * 1000 kg/t = XOF/ha
        revenu_par_ha = {
            c: market[c] * rends_ajustes[c] * 1000
            for c in cultures
        }

        # --- Tentative d'optimisation linéaire avec scipy -----------------
        try:
            from scipy.optimize import linprog

            n = len(cultures)
            # linprog minimise -> on inverse les coefficients (maximisation)
            c_obj = [-revenu_par_ha[culture] for culture in cultures]

            # Contraintes d'inégalité : sum(x_i) <= land_ha
            A_ub = [[1.0] * n]
            b_ub = [land_ha]

            # Bornes : 0 <= x_i <= 70% de la surface (diversification minimale)
            max_mono = 0.7 * land_ha
            bounds = [(0.0, max_mono)] * n

            resultat_scipy = linprog(
                c_obj,
                A_ub=A_ub,
                b_ub=b_ub,
                bounds=bounds,
                method="highs",
            )

            if resultat_scipy.success:
                # Extraction de la répartition optimale
                repartition = {
                    c: round(x, 4)
                    for c, x in zip(cultures, resultat_scipy.x)
                }
                # Revenu attendu total = - valeur objectif minimisée
                revenu_total = round(-resultat_scipy.fun, 2)

                # Construction du texte de recommandation
                culture_dominante = max(repartition, key=repartition.get)
                recommandation_texte = (
                    f"Optimisation linéaire ({severity}). "
                    f"Culture dominante recommandée : {culture_dominante} "
                    f"({repartition[culture_dominante]:.1f} ha)."
                )

                return {
                    "repartition_hectares": repartition,
                    "revenu_attendu_cfa": revenu_total,
                    "recommandation": recommandation_texte,
                    "methode": "scipy_linprog",
                    "confidence_score": 0.75 if severity in ("mild", "no_drought") else 0.55,
                }

        except ImportError:
            logger.warning(
                "scipy non disponible. Utilisation du fallback heuristique pour "
                "optimize()."
            )
        except Exception as e:
            logger.warning(
                f"Erreur scipy.optimize.linprog : {e}. Fallback heuristique."
            )

        # --- Fallback heuristique si scipy est absent ou échoue ----------
        return self._heuristic(land_ha, climate, market)

    def _heuristic(
        self,
        land_ha: float = 1.0,
        climate: dict = None,
        market: dict = None,
        **kwargs,
    ) -> dict:
        """Répartition heuristique de secours pour optimize().

        Applique des règles agronomiques simples au lieu de l'optimiseur
        linéaire. Utilisé quand scipy n'est pas disponible ou quand aucune
        culture n'a de prix connus.

        Le revenu attendu est calculé à partir des prix disponibles dans
        market et des rendements de référence FAO/INSAE du Bénin.
        Si aucun prix n'est disponible, un repli de 150 000 XOF par hectare
        est appliqué (valeur prudente pour les céréales sèches béninoises).

        Args:
            land_ha (float): Surface disponible en hectares.
            climate (dict): Prévisions climatiques.
            market (dict): Prix par culture (peut être vide).
            **kwargs: Anciens noms d'arguments gérés pour rétrocompatibilité.
        Returns:
            dict: Répartition heuristique, revenu estimé et recommandation.
        """
        # Prise en charge des anciens noms d'arguments
        if "available_land_ha" in kwargs:
            land_ha = kwargs.pop("available_land_ha")
        if "climate_forecast" in kwargs:
            climate = kwargs.pop("climate_forecast")
        if "market_forecast" in kwargs:
            market = kwargs.pop("market_forecast")

        if climate is None:
            climate = {}
        if market is None:
            market = {}
        # Correspondance entre les noms de cultures français (heuristique)
        # et les clés de _RENDEMENTS_BENIN (anglais, aligné sur WFP/FAO).
        _NOM_CULTURE_VERS_CLE = {
            "maïs": "maize",
            "soja": "soybean",
            "niébé": "cowpea",
        }

        # Rendement de repli (t/ha) si la culture n'est pas dans _RENDEMENTS_BENIN
        _RENDEMENT_DEFAUT = 1.0

        # Repli de revenu (XOF/ha) appliqué si aucun prix de marché n'est disponible
        _REVENU_REPLI_XOF_PAR_HA = 150_000.0

        # Répartition par défaut : maïs 50 %, soja 30 %, niébé 20 %
        repartition = {
            "maïs": 0.5 * land_ha,
            "soja": 0.3 * land_ha,
            "niébé": 0.2 * land_ha,
        }

        # Ajustement en cas de sécheresse sévère : privilégier le niébé
        severity = climate.get("drought_severity", "mild")
        secheresse = (
            climate.get("secheresse_anticipee", False)
            or severity == "severe"
        )

        if secheresse:
            # En sécheresse sévère : favoriser le niébé (plus résistant)
            repartition["maïs"] = 0.3 * land_ha
            repartition["soja"] = 0.3 * land_ha
            repartition["niébé"] = 0.4 * land_ha

        recommandation_texte = (
            "Sécheresse anticipée : privilégier le niébé, culture résistante."
            if secheresse
            else "Conditions normales : répartition équilibrée recommandée."
        )

        # Calcul du revenu attendu à partir des prix de marché disponibles.
        # Formule : surface_ha × rendement_t_ha × 1 000 kg/t × prix_xof/kg
        revenu_total = 0.0
        for nom_fr, surface_ha in repartition.items():
            # Récupération de la clé anglaise pour la table des rendements
            cle_anglaise = _NOM_CULTURE_VERS_CLE.get(nom_fr, nom_fr)

            # Prix de marché en XOF/kg (clé "prix_moyen_xof" ou absence → 0)
            prix_xof_par_kg = market.get(cle_anglaise, {}).get(
                "prix_moyen_xof", 0.0
            )

            # Rendement FAO/INSAE de référence (t/ha)
            rendement_t_ha = _RENDEMENTS_BENIN.get(cle_anglaise, _RENDEMENT_DEFAUT)

            # Contribution au revenu total
            revenu_total += surface_ha * rendement_t_ha * 1_000.0 * prix_xof_par_kg

        # Repli proportionnel à la surface si aucun prix n'est disponible
        if revenu_total == 0.0:
            revenu_total = land_ha * _REVENU_REPLI_XOF_PAR_HA

        return {
            "repartition_hectares": repartition,
            "revenu_attendu_cfa": round(revenu_total, 2),
            "recommandation": recommandation_texte,
            "methode": "heuristique",
            "confidence_score": 0.3,
        }

init ¶

__init__(forecast=None, logistics=None, pricing=None)

Initialise le module d'aide à la décision.

Parameters:

Name	Type	Description	Default
`forecast`	`Forecasting`	Instance de Forecasting pour les prévisions.	`None`
`logistics`	`Logistics`	Instance de Logistics pour les coûts de transport.	`None`
`pricing`	`Pricing`	Instance de Pricing pour les prix réels du marché. Si None, les méthodes utiliseront des prix estimés par défaut.	`None`

Source code in kadi/market/decision_support.py

def __init__(
    self,
    forecast=None,
    logistics=None,
    pricing=None,
):
    """
    Initialise le module d'aide à la décision.

    Args:
        forecast (Forecasting, optional): Instance de Forecasting pour les prévisions.
        logistics (Logistics, optional): Instance de Logistics pour les coûts de transport.
        pricing (Pricing, optional): Instance de Pricing pour les prix réels du marché.
            Si None, les méthodes utiliseront des prix estimés par défaut.
    """
    # Références vers les sous-modules
    self.forecast = forecast
    self.logistics = logistics
    self.pricing = pricing

getattr ¶

__getattr__(name: str)

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

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

Parameters:

Name	Type	Description	Default
`name`	`str`	Nom de la méthode demandée.	required

Returns:

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

Raises:

Type	Description
`AttributeError`	Si le nom est inconnu.

Source code in kadi/market/decision_support.py

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

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

    Args:
        name (str): Nom de la méthode demandée.

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

    Raises:
        AttributeError: Si le nom est inconnu.
    """
    import warnings
    if name in _DEPRECATED_METHODS:
        new_name = _DEPRECATED_METHODS[name]
        warnings.warn(
            f"Advisor.{name}() est obsolète et sera supprimé dans "
            f"KadiPy v2.0. Utilisez Advisor.{new_name}() à la place.",
            DeprecationWarning,
            stacklevel=2,
        )
        return getattr(self, new_name)
    raise AttributeError(f"'Advisor' n'a pas d'attribut '{name}'.")

arbitrage ¶

arbitrage(crop: str, m_from: str = 'Parakou', to: str = 'Cotonou', qty: float = 1.0, **kwargs) -> dict

Évalue la rentabilité d'un transfert physique de marchandises entre deux marchés.

Utilise les prix réels du marché (via pricing) pour calculer la marge brute. Les coûts logistiques (avec ajustement météo si disponible) sont calculés par le module logistics.

Formule

Gain net = (prix_destination - prix_origine) * 1000 * qty - cout_transport

Parameters:

Name	Type	Description	Default
`crop`	`str`	La culture concernée (ex: 'maize').	required
`m_from`	`str`	Le marché d'achat (ex: 'Parakou').	`'Parakou'`
`to`	`str`	Le marché de vente (ex: 'Cotonou').	`'Cotonou'`
`qty`	`float`	La quantité à transporter en tonnes métriques.	`1.0`
`**kwargs`		Anciens nom de paramètres (market_from, market_to, qty_tons).	`{}`

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant : - 'recommandation' : 'TRANSPORTER' ou 'NE PAS TRANSPORTER' - 'gain_net_total_cfa' : gain net total en XOF - 'gain_net_percent' : gain net en % du capital investi - 'frais_logistiques_total' : coûts de transport totaux en XOF - 'prix_origine_xof_kg' : prix au marché d'achat - 'prix_destination_xof_kg' : prix au marché de vente - 'is_simulated' : True si les prix utilisés sont fictifs - 'confidence_score' : score de confiance de la recommandation (0-1) - 'prob_pluie' : probabilité de pluie utilisée (0 si sans météo)

Source code in kadi/market/decision_support.py

def arbitrage(
    self,
    crop: str,
    m_from: str = "Parakou",
    to: str = "Cotonou",
    qty: float = 1.0,
    **kwargs,
) -> dict:
    """
    Évalue la rentabilité d'un transfert physique de marchandises entre deux marchés.

    Utilise les prix réels du marché (via pricing) pour calculer
    la marge brute. Les coûts logistiques (avec ajustement météo si disponible)
    sont calculés par le module logistics.

    Formule :
        Gain net = (prix_destination - prix_origine) * 1000 * qty - cout_transport

    Args:
        crop (str): La culture concernée (ex: 'maize').
        m_from (str): Le marché d'achat (ex: 'Parakou').
        to (str): Le marché de vente (ex: 'Cotonou').
        qty (float): La quantité à transporter en tonnes métriques.
        **kwargs: Anciens nom de paramètres (market_from, market_to, qty_tons).

    Returns:
        dict: Dictionnaire contenant :
            - 'recommandation'       : 'TRANSPORTER' ou 'NE PAS TRANSPORTER'
            - 'gain_net_total_cfa'   : gain net total en XOF
            - 'gain_net_percent'     : gain net en % du capital investi
            - 'frais_logistiques_total' : coûts de transport totaux en XOF
            - 'prix_origine_xof_kg'  : prix au marché d'achat
            - 'prix_destination_xof_kg' : prix au marché de vente
            - 'is_simulated'         : True si les prix utilisés sont fictifs
            - 'confidence_score'     : score de confiance de la recommandation (0-1)
            - 'prob_pluie'           : probabilité de pluie utilisée (0 si sans météo)
    """
    if "market_from" in kwargs:
        m_from = kwargs.pop("market_from")
    if "market_to" in kwargs:
        to = kwargs.pop("market_to")
    if "qty_tons" in kwargs:
        qty = kwargs.pop("qty_tons")

    # Récupération des prix réels (ou simulés) pour les deux marchés
    prix_origine_kg, simule_origine, conf_origine = self._get_price(
        crop, m_from
    )
    prix_destination_kg, simule_destination, conf_destination = self._get_price(
        crop, to
    )

    # Les données sont simulées si l'une des deux sources l'est
    est_simule = simule_origine or simule_destination

    # Confiance combinée sur les prix : moyenne des deux marchés
    price_confidence = (conf_origine + conf_destination) / 2.0

    # Marge brute par tonne (conversion kg -> tonne : *1000)
    marge_brute_tonne = (prix_destination_kg - prix_origine_kg) * 1000

    # Calcul des frais logistiques via le module dédié
    prob_pluie = 0.0
    if self.logistics:
        # On passe le crop pour activer la perte de qualité par culture
        res_logistics = self.logistics.transfer_cost(
            m_from, to, crop=crop
        )
        frais_logistiques_tonne = res_logistics["total_cost_cfa"]
        prob_pluie = res_logistics.get("prob_pluie", 0.0)
    else:
        frais_logistiques_tonne = 30000.0
        logger.warning(
            "Pas de module logistics. Frais de transport de repli : 30 000 XOF/tonne."
        )

    # Gain net par tonne et total sur la quantité transportée
    gain_net_tonne = marge_brute_tonne - frais_logistiques_tonne
    gain_net_total = gain_net_tonne * qty

    # Gain net en % du capital investi (achat + transport)
    cout_total = (prix_origine_kg * 1000 + frais_logistiques_tonne) * qty
    gain_net_pct = (gain_net_total / cout_total * 100) if cout_total > 0 else 0.0

    # Recommandation basée sur le seuil de rentabilité configuré
    recommandation = (
        "TRANSPORTER" if gain_net_pct >= _SEUIL_RENTABILITE else "NE PAS TRANSPORTER"
    )

    # Score de confiance global sur la recommandation
    confidence_score = self._confidence(
        price_confidence, est_simule, gain_net_pct
    )

    return {
        "recommandation": recommandation,
        "gain_net_total_cfa": round(gain_net_total, 2),
        "gain_net_percent": round(gain_net_pct, 2),
        "frais_logistiques_total": round(frais_logistiques_tonne * qty, 2),
        "prix_origine_xof_kg": round(prix_origine_kg, 2),
        "prix_destination_xof_kg": round(prix_destination_kg, 2),
        "is_simulated": est_simule,
        "confidence_score": confidence_score,
        "prob_pluie": round(prob_pluie, 3),
    }

store_sell ¶

store_sell(crop: str, market: str = 'Parakou', price: float = 300000.0, qty: float = 1.0, months: int = None, **kwargs) -> dict

Évalue s'il est plus rentable de stocker ou de vendre immédiatement.

Phase 4 : l'horizon de stockage est maintenant configurable via le paramètre months. La valeur par défaut est lue depuis config.py (clé de configuration horizon stockage mois, défaut : 3 mois).

Formule

E = E[P_{t+n}] - P_t - C_stockage(n) - C_opportunite(n) - theta * Var(P)

Parameters:

Name	Type	Description	Default
`crop`	`str`	La culture concernée.	required
`market`	`str`	Le marché de référence.	`'Parakou'`
`price`	`float`	Prix actuel en XOF par tonne.	`300000.0`
`qty`	`float`	Quantité récoltée en tonnes.	`1.0`
`months`	`int`	Horizon de stockage en mois. Si None, utilise la valeur par defaut configuree (3 mois).	`None`

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant : - 'recommandation_binaire' : 'STOCKER' ou 'VENDRE IMMÉDIATEMENT' - 'marge_nette_cfa' : espérance de gain total en XOF - 'marge_nette_par_tonne' : espérance de gain par tonne - 'prix_futur_estime' : prix prévu à l'horizon (XOF/tonne) - 'horizon_mois' : horizon de stockage utilisé - 'is_simulated' : True si les données de prévision sont simulées (mode offline ou stub). Propagé depuis predict_price(). Vaut True par défaut si aucun module de prévision n'est disponible. - 'confidence_score' : score de confiance (0-1)

Source code in kadi/market/decision_support.py

def store_sell(
    self,
    crop: str,
    market: str = "Parakou",
    price: float = 300000.0,
    qty: float = 1.0,
    months: int = None,
    **kwargs,
) -> dict:
    """
    Évalue s'il est plus rentable de stocker ou de vendre immédiatement.

    Phase 4 : l'horizon de stockage est maintenant configurable via le
    paramètre months. La valeur par défaut est lue depuis config.py
    (clé de configuration horizon stockage mois, défaut : 3 mois).

    Formule :
        E = E[P_{t+n}] - P_t - C_stockage(n) - C_opportunite(n) - theta * Var(P)

    Args:
        crop (str): La culture concernée.
        market (str): Le marché de référence.
        price (float): Prix actuel en XOF par tonne.
        qty (float): Quantité récoltée en tonnes.
        months (int, optional): Horizon de stockage en mois.
            Si None, utilise la valeur par defaut configuree (3 mois).

    Returns:
        dict: Dictionnaire contenant :
            - 'recommandation_binaire' : 'STOCKER' ou 'VENDRE IMMÉDIATEMENT'
            - 'marge_nette_cfa'        : espérance de gain total en XOF
            - 'marge_nette_par_tonne'  : espérance de gain par tonne
            - 'prix_futur_estime'      : prix prévu à l'horizon (XOF/tonne)
            - 'horizon_mois'           : horizon de stockage utilisé
            - 'is_simulated'           : True si les données de prévision sont
              simulées (mode offline ou stub). Propagé depuis predict_price().
              Vaut True par défaut si aucun module de prévision n'est disponible.
            - 'confidence_score'       : score de confiance (0-1)
    """
    if "current_price" in kwargs:
        price = kwargs.pop("current_price")
    if "qty_tons" in kwargs:
        qty = kwargs.pop("qty_tons")
    if "storage_months" in kwargs:
        months = kwargs.pop("storage_months")
    if "mois_stockage" in kwargs:
        months = kwargs.pop("mois_stockage")

    # Lecture de l'horizon de stockage (paramètre > config > défaut)
    if months is None:
        months = _HORIZON_MOIS_DEFAULT
    jours_stockage = months * 30

    # Valeur par défaut : True si aucun module de prévision n'est disponible
    est_simule = True

    # Récupération du prix futur estimé via le module de prévision
    if self.forecast:
        try:
            prevision = self.forecast.predict(
                crop, market, ahead=jours_stockage
            )
            # La prévision est en XOF/kg ; on convertit en XOF/tonne
            prix_futur_tonne = prevision["predicted_price"] * 1000.0
            rmse_val = prevision.get("rmse")
            if rmse_val is not None:
                variance = rmse_val * 1000.0
            else:
                variance = prevision["predicted_price"] * 0.05 * 1000.0
            # Propagation du flag réel retourné par predict().
            # Par défaut True si la clé est absente (comportement offline conservé).
            est_simule = prevision.get("is_simulated", True)
        except Exception as e:
            logger.warning(
                f"Erreur lors de la prévision {crop}/{market}: {e}. "
                "Hypothèse de hausse de 15%."
            )
            prix_futur_tonne = price * 1.15
            variance = price * 0.05
    else:
        # Sans module de prévision : hypothèse conservative de hausse de 15%
        prix_futur_tonne = price * 1.15
        variance = price * 0.05

    # Coûts de stockage : gardiennage, pertes, sacs (par mois par tonne)
    cout_stockage_mensuel_tonne = 3200.0
    c_stockage = cout_stockage_mensuel_tonne * months

    # Coût d'opportunité : immobilisation de la trésorerie à 1.5%/mois
    taux_opportunite_mensuel = 0.015
    c_opportunite = price * (taux_opportunite_mensuel * months)

    # Pénalité de risque (aversion au risque theta)
    theta_risque = 0.04
    penalite_risque = theta_risque * variance

    # Espérance de gain net par tonne
    esperance_gain = (
        prix_futur_tonne
        - price
        - c_stockage
        - c_opportunite
        - penalite_risque
    )

    recommandation = "STOCKER" if esperance_gain > 0 else "VENDRE IMMÉDIATEMENT"

    # Gain net en % du capital actuel (pour le score de confiance)
    gain_pct = (esperance_gain / price * 100) if price > 0 else 0.0

    # Score de confiance (prix simulés = confiance faible)
    confidence_score = self._confidence(
        price_confidence=0.0 if est_simule else 0.7,
        is_simulated=est_simule,
        gain_net_pct=gain_pct,
    )

    return {
        "recommandation_binaire": recommandation,
        "marge_nette_cfa": round(esperance_gain * qty, 2),
        "marge_nette_par_tonne": round(esperance_gain, 2),
        "prix_futur_estime": round(prix_futur_tonne, 2),
        "horizon_mois": months,
        "is_simulated": est_simule,
        "confidence_score": confidence_score,
    }

optimize ¶

optimize(land_ha: float = 1.0, climate: dict = None, market: dict = None, yields: dict = None, **kwargs) -> dict

Optimise la répartition des cultures sur la surface disponible.

Phase 4 : utilise scipy.optimize.linprog pour maximiser le revenu attendu sous contraintes agronomiques et de surface. Si scipy n'est pas disponible, utilise un fallback heuristique.

Modèle d'optimisation (programmation linéaire) : - Variables : x_i = hectares alloués à la culture i - Objectif : maximiser sum(prix_i * rendement_i * x_i) - Contraintes : * sum(x_i) <= land_ha (surface totale) * x_i >= 0 * x_i <= 0.7 * land_ha (diversification minimale)

Ajustements climatiques (SPI de sécheresse) : - Sécheresse sévère (drought_severity='severe') : rendement_niébé * 1.3 (résistant), rendement_maïs * 0.7 - Sécheresse modérée ('moderate') : rendement_maïs * 0.85

Parameters:

Name	Type	Description	Default
`land_ha`	`float`	Surface arable disponible en hectares.	`1.0`
`climate`	`dict`	Prévisions climatiques. Clés attendues : - 'secheresse_anticipee' (bool) : compatibilité V1 - 'drought_severity' (str) : 'no_drought', 'mild', 'moderate', 'severe' - 'prob_pluie_7j' (float) : probabilité de pluie sur 7 jours	`None`
`market`	`dict`	Prix médians actuels par culture (XOF/kg). Exemple : {'maize': 285, 'cowpea': 580, 'sorghum': 210}	`None`
`yields`	`dict`	Rendements attendus en tonnes/hectare. Si None, utilise les rendements de référence béninois.	`None`
`**kwargs`		Anciens arguments (available_land_ha, climate_forecast, market_forecast).	`{}`

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant : - 'repartition_hectares' : répartition optimale par culture (hectares) - 'revenu_attendu_cfa' : revenu total attendu en XOF - 'recommandation' : texte explicatif - 'methode' : 'scipy_linprog' ou 'heuristique' - 'confidence_score' : score de confiance de l'optimisation

Source code in kadi/market/decision_support.py

def optimize(
    self,
    land_ha: float = 1.0,
    climate: dict = None,
    market: dict = None,
    yields: dict = None,
    **kwargs,
) -> dict:
    """
    Optimise la répartition des cultures sur la surface disponible.

    Phase 4 : utilise scipy.optimize.linprog pour maximiser le revenu
    attendu sous contraintes agronomiques et de surface. Si scipy n'est
    pas disponible, utilise un fallback heuristique.

    Modèle d'optimisation (programmation linéaire) :
        - Variables : x_i = hectares alloués à la culture i
        - Objectif  : maximiser sum(prix_i * rendement_i * x_i)
        - Contraintes :
            * sum(x_i) <= land_ha  (surface totale)
            * x_i >= 0
            * x_i <= 0.7 * land_ha  (diversification minimale)

    Ajustements climatiques (SPI de sécheresse) :
        - Sécheresse sévère (drought_severity='severe') :
            rendement_niébé * 1.3 (résistant), rendement_maïs * 0.7
        - Sécheresse modérée ('moderate') :
            rendement_maïs * 0.85

    Args:
        land_ha (float): Surface arable disponible en hectares.
        climate (dict): Prévisions climatiques. Clés attendues :
            - 'secheresse_anticipee' (bool) : compatibilité V1
            - 'drought_severity' (str)      : 'no_drought', 'mild', 'moderate', 'severe'
            - 'prob_pluie_7j' (float)       : probabilité de pluie sur 7 jours
        market (dict): Prix médians actuels par culture (XOF/kg).
            Exemple : {'maize': 285, 'cowpea': 580, 'sorghum': 210}
        yields (dict, optional): Rendements attendus en tonnes/hectare.
            Si None, utilise les rendements de référence béninois.
        **kwargs: Anciens arguments (available_land_ha, climate_forecast, market_forecast).

    Returns:
        dict: Dictionnaire contenant :
            - 'repartition_hectares' : répartition optimale par culture (hectares)
            - 'revenu_attendu_cfa'   : revenu total attendu en XOF
            - 'recommandation'       : texte explicatif
            - 'methode'              : 'scipy_linprog' ou 'heuristique'
            - 'confidence_score'     : score de confiance de l'optimisation
    """

    if "available_land_ha" in kwargs:
        land_ha = kwargs.pop("available_land_ha")
    if "climate_forecast" in kwargs:
        climate = kwargs.pop("climate_forecast")
    if "market_forecast" in kwargs:
        market = kwargs.pop("market_forecast")

    if climate is None:
        climate = {}
    if market is None:
        market = {}

    # Rendements de référence (paramètre > défauts Bénin)
    rends = dict(_RENDEMENTS_BENIN)
    if yields:
        rends.update(yields)

    # Cultures à optimiser : celles pour lesquelles on a un prix de marché
    # On filtre pour ne garder que les cultures connues dans les rendements
    cultures = [
        c for c in market
        if c in rends and market[c] > 0
    ]

    if not cultures:
        # Aucune culture avec données : fallback heuristique complet
        return self._heuristic(
            land_ha, climate, market
        )

    # Ajustement des rendements selon la sévérité de la sécheresse
    severity = climate.get("drought_severity", "mild")
    if climate.get("secheresse_anticipee", False):
        # Rétrocompatibilité V1 : bool -> sécheresse sévère
        severity = "severe"

    rends_ajustes = {}
    for c in cultures:
        r = rends[c]
        if severity == "severe":
            # Sécheresse sévère : niébé résiste, maïs souffre
            if c == "cowpea":
                r *= 1.3
            elif c in ("maize", "rice"):
                r *= 0.7
        elif severity == "moderate":
            if c == "maize":
                r *= 0.85
        rends_ajustes[c] = r

    # Revenus attendus par hectare pour chaque culture (XOF/ha)
    # prix en XOF/kg * rendement en t/ha * 1000 kg/t = XOF/ha
    revenu_par_ha = {
        c: market[c] * rends_ajustes[c] * 1000
        for c in cultures
    }

    # --- Tentative d'optimisation linéaire avec scipy -----------------
    try:
        from scipy.optimize import linprog

        n = len(cultures)
        # linprog minimise -> on inverse les coefficients (maximisation)
        c_obj = [-revenu_par_ha[culture] for culture in cultures]

        # Contraintes d'inégalité : sum(x_i) <= land_ha
        A_ub = [[1.0] * n]
        b_ub = [land_ha]

        # Bornes : 0 <= x_i <= 70% de la surface (diversification minimale)
        max_mono = 0.7 * land_ha
        bounds = [(0.0, max_mono)] * n

        resultat_scipy = linprog(
            c_obj,
            A_ub=A_ub,
            b_ub=b_ub,
            bounds=bounds,
            method="highs",
        )

        if resultat_scipy.success:
            # Extraction de la répartition optimale
            repartition = {
                c: round(x, 4)
                for c, x in zip(cultures, resultat_scipy.x)
            }
            # Revenu attendu total = - valeur objectif minimisée
            revenu_total = round(-resultat_scipy.fun, 2)

            # Construction du texte de recommandation
            culture_dominante = max(repartition, key=repartition.get)
            recommandation_texte = (
                f"Optimisation linéaire ({severity}). "
                f"Culture dominante recommandée : {culture_dominante} "
                f"({repartition[culture_dominante]:.1f} ha)."
            )

            return {
                "repartition_hectares": repartition,
                "revenu_attendu_cfa": revenu_total,
                "recommandation": recommandation_texte,
                "methode": "scipy_linprog",
                "confidence_score": 0.75 if severity in ("mild", "no_drought") else 0.55,
            }

    except ImportError:
        logger.warning(
            "scipy non disponible. Utilisation du fallback heuristique pour "
            "optimize()."
        )
    except Exception as e:
        logger.warning(
            f"Erreur scipy.optimize.linprog : {e}. Fallback heuristique."
        )

    # --- Fallback heuristique si scipy est absent ou échoue ----------
    return self._heuristic(land_ha, climate, market)

Aide à la décision (kadi.market.decision_support)¶

Importation directe¶

Initialisation¶

Méthodes¶

arbitrage(crop, m_from, to, qty)¶

store_sell(crop, market, price, qty, months)¶

optimize(land_ha, climate, market, yields)¶

Méthodes dépréciées¶

Classe dépréciée¶

Advisor ¶

__init__ ¶

__getattr__ ¶

arbitrage ¶

store_sell ¶

optimize ¶

Aide à la décision (`kadi.market.decision_support`)¶

`arbitrage(crop, m_from, to, qty)`¶

`store_sell(crop, market, price, qty, months)`¶

`optimize(land_ha, climate, market, yields)`¶

init ¶

getattr ¶