Decision Support (`kadi.market.decision_support`)¶

The Advisor class converts price forecasts and market data into operational recommendations: spatial arbitrage between markets, storage decision timing, and crop portfolio optimization.

It is used internally by the Market facade and accessible via market.advisor. It can also be instantiated directly.

Direct Importing¶

from kadi.market import Advisor

Initialization¶

Advisor is created automatically by Market and is accessible via market.advisor. For direct instantiation:

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

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

Parameters:

Name	Type	Default	Description
`forecast`	`Forecasting`	`None`	Forecasting instance to estimate future prices
`logistics`	`Logistics`	`None`	Logistics instance for transport costs
`pricing`	`Pricing`	`None`	Pricing instance for real market prices

If pricing is None, methods fall back to a default price of 300 XOF/kg with is_simulated=True and confidence_score=0.0.

Public Attributes:

Attribute	Type	Description
`forecast`	`Forecasting`	Price forecasting module
`logistics`	`Logistics`	Logistics module
`pricing`	`Pricing`	Pricing module

Methods¶

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

Evaluates the profitability of a physical merchandise transfer between two markets.

Formula:

Net Gain = (destination_price - origin_price) * 1000 * qty - transport_cost

The minimum profitability threshold is configurable in config.py (key logistics.seuil_rentabilite_pct, default: 10%).

import kadi as kd

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

# Is it profitable to transport 10 tonnes of maize from Parakou to Cotonou?
decision = market.advisor.arbitrage(
    crop="maize",
    m_from="Parakou",
    to="Cotonou",
    qty=10.0,
)

print(decision["recommandation"])           # 'TRANSPORTER' or 'NE PAS TRANSPORTER'
print(f"Total net gain: {decision['gain_net_total_cfa']:,.0f} XOF")
print(f"Net gain      : {decision['gain_net_percent']:.1f}%")
print(f"Transport cost: {decision['frais_logistiques_total']:,.0f} XOF")
print(f"Rain forecast : {decision['prob_pluie'] * 100:.0f}%")
print(f"Confidence    : {decision['confidence_score']:.2f}")

Parameters:

Name	Type	Default	Description
`crop`	`str`	required	Crop code (e.g. `'maize'`)
`m_from`	`str`	`'Parakou'`	Purchase market
`to`	`str`	`'Cotonou'`	Sale market
`qty`	`float`	`1.0`	Quantity to transport in metric tonnes

Returns: dict

Key	Type	Description
`recommandation`	`str`	`'TRANSPORTER'` or `'NE PAS TRANSPORTER'`
`gain_net_total_cfa`	`float`	Total net gain in XOF
`gain_net_percent`	`float`	Net gain as % of invested capital
`frais_logistiques_total`	`float`	Total transport costs in XOF
`prix_origine_xof_kg`	`float`	Purchase market price in XOF/kg
`prix_destination_xof_kg`	`float`	Sale market price in XOF/kg
`is_simulated`	`bool`	True if used prices are dummy/simulated
`confidence_score`	`float`	Recommendation confidence score (0-1)
`prob_pluie`	`float`	Rain probability used for logistics calculation

Confidence Score:

Calculated as a weighted combination of three factors:

score = 0.5 * price_confidence
      + 0.3 * (0 if simulated, 1 if real)
      + 0.2 * min(1, |net_gain_pct| / 30)

Exceptions:

No exception is raised if data is missing: the method falls back to a default price of 300 XOF/kg with is_simulated=True.

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

Evaluates whether it is more profitable to store a harvest or sell immediately.

Net Expected Gain Formula per Tonne:

E = E[P(t+n)] - P(t) - C_storage(n) - C_opportunity(n) - theta * Var(P)

Where: - C_storage = 3,200 XOF/tonne/month (storage, loss, bags) - C_opportunity = 1.5%/month of current price (working capital lock-up) - theta = 0.04 (risk aversion coefficient)

# Store 5 tonnes of yam for 3 months or sell now?
storage = market.advisor.store_sell(
    crop="yam",
    market="Abomey",
    price=250_000.0,    # Current price in XOF/tonne
    qty=5.0,
    months=3,
)

print(storage["recommandation_binaire"])   # 'STOCKER' or 'VENDRE IMMÉDIATEMENT'
print(f"Total net margin   : {storage['marge_nette_cfa']:,.0f} XOF")
print(f"Margin per tonne   : {storage['marge_nette_par_tonne']:,.0f} XOF")
print(f"Estimated future P : {storage['prix_futur_estime']:,.0f} XOF/tonne")
print(f"Horizon            : {storage['horizon_mois']} months")
print(f"Confidence         : {storage['confidence_score']:.2f}")

Parameters:

Name	Type	Default	Description
`crop`	`str`	required	Crop code
`market`	`str`	`'Parakou'`	Reference market for forecasting
`price`	`float`	`300000.0`	Current price in XOF per tonne
`qty`	`float`	`1.0`	Quantity in tonnes
`months`	`int`	`None`	Storage horizon in months. If None, reads default from config.py (default: 3 months)

Returns: dict

Key	Type	Description
`recommandation_binaire`	`str`	`'STOCKER'` or `'VENDRE IMMEDIATEMENT'`
`marge_nette_cfa`	`float`	Total expected net gain in XOF
`marge_nette_par_tonne`	`float`	Expected net gain per tonne
`prix_futur_estime`	`float`	Predicted price at horizon (XOF/tonne)
`horizon_mois`	`int`	Storage horizon used
`is_simulated`	`bool`	True if forecast data is simulated
`confidence_score`	`float`	Confidence score (0-1)

If no forecast module is available, the method applies a conservative 15% increase assumption on the current price.

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

Optimizes crop allocation over available arable land.

Uses scipy.optimize.linprog (HiGHS method) to maximize expected revenue under land capacity and diversification constraints. If scipy is unavailable, a heuristic fallback is applied.

Optimization Model:

Maximize: sum(price_i * yield_i * x_i)
Subject to: sum(x_i) <= land_ha
            0 <= x_i <= 0.7 * land_ha  (minimum 30% diversification)

Climate Adjustments (key drought_severity):

Severity	Effect
`'severe'`	cowpea yield * 1.3, maize and rice * 0.7
`'moderate'`	maize yield * 0.85
`'mild'` or `'no_drought'`	no adjustment

# Optimization under mild climate conditions
portfolio_decision = market.advisor.optimize(
    land_ha=10.0,
    climate={"drought_severity": "mild"},
    market={"maize": 285.0, "cowpea": 580.0, "sorghum": 210.0},
)

print(f"Method          : {portfolio_decision['methode']}")
print(f"Expected revenue: {portfolio_decision['revenu_attendu_cfa']:,.0f} XOF")
print(f"Confidence      : {portfolio_decision['confidence_score']:.2f}")
for crop, ha in portfolio_decision["repartition_hectares"].items():
    print(f"  {crop} : {ha:.1f} ha")

# Severe drought scenario (favors cowpea)
drought_decision = market.advisor.optimize(
    land_ha=5.0,
    climate={"drought_severity": "severe"},
    market={"maize": 285.0, "cowpea": 620.0, "sorghum": 210.0},
)

Parameters:

Name	Type	Default	Description
`land_ha`	`float`	`1.0`	Available arable land in hectares
`climate`	`dict`	`None`	Climate forecast dictionary (see table below)
`market`	`dict`	`None`	Current median prices per crop in XOF/kg
`yields`	`dict`	`None`	Expected yields in t/ha. If None, uses FAO/INSAE Benin reference yields

Keys in climate:

Key	Type	Description
`drought_severity`	`str`	Severity: `'no_drought'`, `'mild'`, `'moderate'`, `'severe'`
`secheresse_anticipee`	`bool`	V1 backward compatibility: True maps to severity `'severe'`
`prob_pluie_7j`	`float`	7-day rain probability

Reference Yields (FAO/INSAE Benin):

Crop	Yield (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

Returns: dict

Key	Type	Description
`repartition_hectares`	`dict[str, float]`	Allocated land per crop (ha)
`revenu_attendu_cfa`	`float`	Total expected revenue in XOF
`recommandation`	`str`	Explanatory text of the decision
`methode`	`str`	`'scipy_linprog'` or `'heuristique'`
`confidence_score`	`float`	Confidence score (0.75 normal, 0.55 drought, 0.3 heuristic)

Heuristic Fallback:

Activated if scipy is missing or if no crop has a known price. Default allocation is: maize 50%, soybean 30%, cowpea 20%. Under severe drought: maize 30%, soybean 30%, cowpea 40%.

Deprecated Methods¶

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

Deprecated Class¶

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

Decision Support (kadi.market.decision_support)¶

Direct Importing¶

Initialization¶

Methods¶

arbitrage(crop, m_from, to, qty)¶

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

optimize(land_ha, climate, market, yields)¶

Deprecated Methods¶

Deprecated Class¶

Advisor ¶

__init__ ¶

__getattr__ ¶

arbitrage ¶

store_sell ¶

optimize ¶

Decision Support (`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 ¶