Logistique (`kadi.market.logistics`)¶

Le module MarketLogistics modélise les frictions du transport de marchandises au Bénin : distances routières réelles, coûts en carburant, tracasseries aux postes de contrôle et perte de qualité des produits. Depuis la Phase 4, il intègre les prévisions météo pour ajuster ses calculs dynamiquement.

Formule de coût de transfert¶

C_transfer = C_info
           + Distance × (gamma_effectif × P_carburant / 100 + mu_checkpoints)
           + C_qualite(culture, distance, pluie)

Terme	Signification	Source
`C_info`	Coût fixe de recherche d'informations	`config.py`
`gamma_effectif`	Coefficient d'état des routes (dynamique)	Formule ci-dessous
`P_carburant`	Prix de l'essence en XOF/litre	Env, GitHub, repli
`mu_checkpoints`	Tracasseries policières par km	`config.py`
`C_qualite`	Perte de valeur marchande	Culture × distance × pluie

Ajustement météo (Phase 4)¶

Quand une weather_session est fournie :

gamma_effectif = gamma_base × (1 + alpha_pluie × prob_pluie)

gamma_base = 1.2 (par défaut)
alpha_pluie = 0.25 (majoration max de 25% si pluie certaine)
prob_pluie = probabilité de pluie demain (0.0 à 1.0)

La perte de qualité dépend aussi de la culture et de la pluie :

C_qualite = facteur_culture × distance_km × (1 + beta_pluie × prob_pluie)

Initialisation¶

from kadi.market.logistics import MarketLogistics

# Sans météo (comportement V1)
logistics = MarketLogistics()

# Avec météo (Phase 4)
from kadi.weather import WeatherSession

ws = WeatherSession(latitude=9.30, longitude=2.08, name="Parakou")
logistics = MarketLogistics(weather_session=ws)

Via la façade Market (recommandé) :

from kadi.market import Market
from kadi.weather import WeatherSession

ws = WeatherSession(latitude=9.30, longitude=2.08, name="Parakou")
marche = Market(lat=9.30, lon=2.08, location="Parakou", weather_session=ws)
# marche.logistics est un MarketLogistics avec météo injectée

Méthodes¶

`get_distance(origine, destination)`¶

Calcule la distance routière entre deux villes béninoises. Utilise OSRM pour la vraie distance de conduite. Les résultats sont mis en cache dans ~/.kadi/osrm_cache.json.

km = logistics.get_distance("Parakou", "Cotonou")
print(f"Distance réelle : {km:.1f} km")

Stratégie de résolution :

Cache local (évite les appels réseau répétés)
Géocodage Nominatim → routage OSRM
Fallback Haversine × 1.3 si OSRM est indisponible
100 km par défaut si le géocodage échoue

`calculate_transfer_cost(origine, destination, prix_carburant, crop)`¶

Calcule le coût total de transport d'un point A vers un point B.

cout = logistics.calculate_transfer_cost(
    origine="Parakou",
    destination="Cotonou",
    crop="tomato",       # Optionnel — active la perte de qualité par culture
)

Paramètres :

Nom	Type	Défaut	Description
`origine`	`str`	requis	Ville de départ
`destination`	`str`	requis	Ville d'arrivée
`prix_carburant`	`float`	`None`	Prix de l'essence en XOF/litre (auto-récupéré si None)
`crop`	`str`	`None`	Culture transportée (influence la perte de qualité)

Retour : dict

Clé	Type	Description
`total_cost_cfa`	`float`	Coût total du transfert en XOF
`prob_pluie`	`float`	Probabilité de pluie utilisée (0 si sans météo)
`gamma_effectif`	`float`	Coefficient de route réellement appliqué
`details`	`dict`	Détail de chaque composante du coût

Détail du résultat :

# Exemple de retour complet
{
    "total_cost_cfa": 47_350.0,
    "prob_pluie": 0.82,
    "gamma_effectif": 1.446,
    "details": {
        "distance_km": 415.2,
        "search_costs": 5000.0,
        "transport_costs": 38_200.0,
        "quality_loss": 4150.0,
        "fuel_price_used": 680.0,
        "gamma_route_base": 1.2,
        "gamma_effectif": 1.446,
        "prob_pluie": 0.82,
        "crop": "tomato",
    }
}

Facteurs de perte de qualité par culture¶

Culture	Facteur (XOF/km/tonne)	Sensibilité
Maïs, Sorgho, Mil	5.0	Très faible (céréales sèches)
Riz	6.0	Faible
Niébé, Soja	7.0	Modérée
Manioc	10.0	Modérée
Igname	12.0	Élevée (sensible aux chocs)
Oignon	20.0	Très élevée
Tomate	25.0	Extrême (légume frais)

Sous la pluie, la perte est majorée de 50% maximum (beta_pluie = 0.5).

Prix du carburant¶

Le module récupère le prix de l'essence selon cette cascade :

Variable d'environnement BENIN_FUEL_PRICE
Fichier en ligne config/fuel_prices.json sur GitHub
Valeur de repli : 680 XOF/litre (configurable dans config.py)

# .env — force le prix manuellement
BENIN_FUEL_PRICE=695

Exemple complet avec météo¶

from kadi.weather import WeatherSession
from kadi.market import Market

ws = WeatherSession(latitude=9.30, longitude=2.08, name="Parakou")
marche = Market(lat=9.30, lon=2.08, location="Parakou", weather_session=ws)

# Coût pour transporter des tomates (culture périssable, sensible à la pluie)
cout = marche.logistics.calculate_transfer_cost(
    "Parakou", "Cotonou", crop="tomato"
)

print(f"Coût total          : {cout['total_cost_cfa']:,.0f} XOF")
print(f"Pluie demain        : {cout['prob_pluie'] * 100:.0f}%")
print(f"Gamma route effectif: {cout['gamma_effectif']:.3f}")
print(f"Perte qualité       : {cout['details']['quality_loss']:,.0f} XOF")

MarketLogistics ¶

Classe modélisant les frictions logistiques réelles au Bénin : coûts de transport, tracasseries aux postes de contrôle, et dégradation de la qualité des marchandises.

Phase 4 - Intégration météo

Si un weather_session est fourni à l'initialisation, le calcul des coûts de transfert utilise la probabilité de pluie du lendemain pour ajuster gamma_route (état des routes) et la perte de qualité. En l'absence de weather_session, le comportement est identique à la V1.

Source code in kadi/market/logistics.py

class MarketLogistics:
    """
    Classe modélisant les frictions logistiques réelles au Bénin :
    coûts de transport, tracasseries aux postes de contrôle,
    et dégradation de la qualité des marchandises.

    Phase 4 - Intégration météo :
        Si un weather_session est fourni à l'initialisation, le calcul
        des coûts de transfert utilise la probabilité de pluie du lendemain
        pour ajuster gamma_route (état des routes) et la perte de qualité.
        En l'absence de weather_session, le comportement est identique à la V1.
    """

    def __init__(self, cache_file: str = None, weather_session=None):
        """
        Initialise le module logistique.

        Charge le cache persistant des coordonnées GPS et des distances
        routières pour éviter des appels réseau redondants.

        Args:
            cache_file (str, optional): Chemin vers le fichier de cache JSON.
                Si None, utilise ``~/.kadi/osrm_cache.json``.
            weather_session (WeatherSession, optional): Session météo
                (kadi.weather.WeatherSession) pour ajuster les coûts selon
                la météo prévue. Si None, pas d'ajustement climatique.
        """
        # Session météo optionnelle (intégration Phase 4)
        self.weather_session = weather_session

        # Initialisation du chemin du fichier de cache
        if cache_file is None:
            cache_dir = os.path.expanduser("~/.kadi")
            os.makedirs(cache_dir, exist_ok=True)
            self.cache_file = os.path.join(cache_dir, "osrm_cache.json")
        else:
            self.cache_file = cache_file

        # Structure du cache en mémoire (coordonnées et distances routières)
        self.cache = {"coords": {}, "distances": {}}

        # Cache en mémoire pour le prix du carburant (évite les appels répétés)
        self._cached_fuel_price: Optional[float] = None

        # Cache en mémoire pour la probabilité de pluie (une seule requête par session)
        self._cached_prob_pluie: Optional[float] = None

        # Chargement du cache depuis le disque
        self._load_cache()

    def _load_cache(self):
        """Charge le cache persistant depuis le fichier JSON s'il existe."""
        if os.path.isfile(self.cache_file):
            try:
                with open(self.cache_file, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    self.cache["coords"] = data.get("coords", {})
                    self.cache["distances"] = data.get("distances", {})
            except Exception as e:
                logger.warning(f"Impossible de lire le cache {self.cache_file}: {e}")

    def _save_cache(self):
        """Sauvegarde le cache en mémoire dans le fichier JSON persistant."""
        try:
            with open(self.cache_file, "w", encoding="utf-8") as f:
                json.dump(self.cache, f, indent=4)
        except Exception as e:
            logger.warning(f"Impossible de sauvegarder le cache {self.cache_file}: {e}")

    def _get_prob_pluie(self) -> float:
        """
        Retourne la probabilité de pluie, mise en cache pour la durée de la session.

        La probabilité est calculée une seule fois par session logistique pour
        éviter des appels répétés au module météo.

        Returns:
            float: Probabilité de pluie entre 0.0 et 1.0.
        """
        if self._cached_prob_pluie is None:
            self._cached_prob_pluie = _obtenir_prob_pluie(self.weather_session)
        return self._cached_prob_pluie

    def _haversine_distance(
        self, lon1: float, lat1: float, lon2: float, lat2: float
    ) -> float:
        """
        Calcule la distance orthodromique entre deux points GPS
        en utilisant la formule de Haversine.

        Utilisée comme fallback si OSRM est indisponible. La distance
        vol d'oiseau est multipliée par un facteur de tortuosité (1.3)
        pour approximer la distance réelle sur les routes béninoises.

        Args:
            lon1 (float): Longitude du point de départ.
            lat1 (float): Latitude du point de départ.
            lon2 (float): Longitude du point d'arrivée.
            lat2 (float): Latitude du point d'arrivée.

        Returns:
            float: Distance routière approximée en kilomètres.
        """
        # Rayon de la Terre en kilomètres
        R = 6371.0

        # Conversion des coordonnées en radians
        lat1_r = math.radians(lat1)
        lon1_r = math.radians(lon1)
        lat2_r = math.radians(lat2)
        lon2_r = math.radians(lon2)

        dlon = lon2_r - lon1_r
        dlat = lat2_r - lat1_r

        # Formule de Haversine
        a = (
            math.sin(dlat / 2) ** 2
            + math.cos(lat1_r) * math.cos(lat2_r) * math.sin(dlon / 2) ** 2
        )
        c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

        distance_vol_oiseau = R * c

        # Facteur de tortuosité : les routes béninoises font environ 30%
        # de plus que la distance à vol d'oiseau en moyenne
        return distance_vol_oiseau * 1.3

    def _geocode_city(self, city_name: str) -> Optional[Tuple[float, float]]:
        """
        Interroge Nominatim (OpenStreetMap) pour obtenir les coordonnées GPS
        d'une ville. Respecte la limite de taux de 1 requête/seconde.

        Le résultat est mis en cache pour éviter des appels répétés.

        Args:
            city_name (str): Nom de la ville à géocoder (ex: 'Cotonou').

        Returns:
            tuple: (longitude, latitude) ou None si la ville est introuvable.
        """
        city_key = city_name.strip().lower()

        # Retour depuis le cache si déjà résolu
        if city_key in self.cache["coords"]:
            return tuple(self.cache["coords"][city_key])

        # Respect du rate-limit Nominatim avant toute requête
        _respecter_rate_limit_nominatim()

        url = "https://nominatim.openstreetmap.org/search"
        params = {
            "q": f"{city_name}, Benin",
            "format": "json",
            "limit": 1,
        }
        # Identification de l'application (requis par les conditions d'utilisation Nominatim)
        headers = {"User-Agent": "KadiPy/0.1.0 (Agritech Research, Benin)"}

        try:
            response = requests.get(url, params=params, headers=headers, timeout=5)
            response.raise_for_status()
            data = response.json()

            if data and len(data) > 0:
                lon = float(data[0]["lon"])
                lat = float(data[0]["lat"])
                # Mise en cache du résultat
                self.cache["coords"][city_key] = [lon, lat]
                self._save_cache()
                return (lon, lat)
            else:
                logger.warning(f"Ville '{city_name}' introuvable via Nominatim.")

        except Exception as e:
            logger.warning(f"Erreur lors du géocodage de '{city_name}': {e}")

        return None

    def _fetch_osrm_distance(
        self, lon1: float, lat1: float, lon2: float, lat2: float
    ) -> Optional[float]:
        """
        Interroge le serveur public OSRM pour calculer la distance routière
        réelle entre deux points GPS.

        Args:
            lon1 (float): Longitude du point de départ.
            lat1 (float): Latitude du point de départ.
            lon2 (float): Longitude du point d'arrivée.
            lat2 (float): Latitude du point d'arrivée.

        Returns:
            float: Distance en kilomètres, ou None si la requête a échoué.
        """
        url = (
            f"http://router.project-osrm.org/route/v1/driving/"
            f"{lon1},{lat1};{lon2},{lat2}?overview=false"
        )
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            data = response.json()

            if data.get("code") == "Ok" and data.get("routes"):
                distance_metres = data["routes"][0]["distance"]
                return distance_metres / 1000.0

        except Exception as e:
            logger.warning(f"Erreur lors de la requête OSRM : {e}")

        return None

    def get_distance(self, origine: str, destination: str) -> float:
        """
        Récupère la distance routière réelle entre deux villes béninoises.

        Stratégie en cascade :
        1. Cache local (résultat d'un appel précédent)
        2. Géocodage Nominatim + routage OSRM
        3. Fallback Haversine (vol d'oiseau * 1.3) si OSRM échoue
        4. Valeur par défaut 100 km si le géocodage échoue

        Args:
            origine (str): Nom de la ville de départ (ex: 'Parakou').
            destination (str): Nom de la ville d'arrivée (ex: 'Cotonou').

        Returns:
            float: Distance estimée en kilomètres.
        """
        orig_key = origine.strip().lower()
        dest_key = destination.strip().lower()
        route_key = f"{orig_key}_{dest_key}"
        rev_route_key = f"{dest_key}_{orig_key}"

        # 1. Vérification du cache (aller et retour)
        if route_key in self.cache["distances"]:
            return self.cache["distances"][route_key]
        if rev_route_key in self.cache["distances"]:
            return self.cache["distances"][rev_route_key]

        # 2. Géocodage des deux villes
        coords_orig = self._geocode_city(origine)
        coords_dest = self._geocode_city(destination)

        if not coords_orig or not coords_dest:
            logger.warning(
                f"Géocodage impossible pour '{origine}' -> '{destination}'. "
                "Valeur de repli : 100 km."
            )
            return 100.0

        # 3. Requête OSRM pour la vraie distance routière
        dist = self._fetch_osrm_distance(
            coords_orig[0], coords_orig[1],
            coords_dest[0], coords_dest[1],
        )

        # 4. Fallback Haversine si OSRM est indisponible
        if dist is None:
            logger.warning(
                f"OSRM indisponible pour '{origine}' -> '{destination}'. "
                "Utilisation de la formule Haversine (distance approximée)."
            )
            dist = self._haversine_distance(
                coords_orig[0], coords_orig[1],
                coords_dest[0], coords_dest[1],
            )

        # Mise en cache du résultat
        self.cache["distances"][route_key] = dist
        self._save_cache()

        return dist

    def _fetch_fuel_price(self) -> float:
        """
        Récupère le prix actuel du carburant (essence) au Bénin.

        Stratégie en cascade :
        1. Variable d'environnement BENIN_FUEL_PRICE (priorité absolue)
        2. Cache en mémoire (évite les appels répétés)
        3. Fichier de configuration sur GitHub (config/fuel_prices.json)
        4. Valeur de repli depuis la configuration (680 XOF/litre par défaut)

        Returns:
            float: Prix de l'essence en XOF par litre.
        """
        # Priorité 1 : variable d'environnement
        env_price = os.getenv("BENIN_FUEL_PRICE")
        if env_price is not None:
            try:
                return float(env_price)
            except ValueError:
                logger.warning(
                    f"Valeur BENIN_FUEL_PRICE invalide dans .env : '{env_price}'. "
                    "Passage à la source suivante."
                )

        # Priorité 2 : cache en mémoire
        if self._cached_fuel_price is not None:
            return self._cached_fuel_price

        # Priorité 3 : configuration en ligne (GitHub)
        url = "https://raw.githubusercontent.com/delsDin/kadipy/main/config/fuel_prices.json"
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            data = response.json()
            price = float(data.get("benin", {}).get("essence", None))
            self._cached_fuel_price = price
            return price
        except Exception as e:
            logger.warning(
                f"Impossible de récupérer le prix du carburant en ligne ({e}). "
                "Utilisation de la valeur de repli."
            )

        # Priorité 4 : valeur de repli depuis la configuration
        prix_repli = _LOGISTICS_CONFIG.get("prix_carburant_fallback_xof", 680.0)
        self._cached_fuel_price = prix_repli
        return prix_repli

    def calculate_transfer_cost(
        self,
        origine: str,
        destination: str,
        prix_carburant: float = None,
        crop: str = None,
    ) -> dict:
        """
        Calcule le coût total de transfert d'un point A vers un point B.

        Phase 4 - Intégration météo :
            Si un weather_session a été fourni à l'initialisation, le calcul
            ajuste automatiquement :
            - gamma_route selon la probabilité de pluie prévue demain.
            - La perte de qualité selon la culture et la météo.

        Formule :
            C_transfer = C_info
                       + Distance * (gamma_effectif * P_carburant / 100 + mu_checkpoints)
                       + C_qualite(culture, distance, pluie)

        Où :
        - C_info = coût fixe de recherche d'informations (appels, déplacements)
        - gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie)
        - P_carburant = prix du litre d'essence en XOF
        - mu_checkpoints = coût moyen des tracasseries par km
        - C_qualite = perte de valeur marchande (variable par culture et météo)

        Args:
            origine (str): Ville de départ (ex: 'Parakou').
            destination (str): Ville de destination (ex: 'Cotonou').
            prix_carburant (float, optional): Prix du litre d'essence en XOF.
                Si None, récupéré automatiquement depuis les sources configurées.
            crop (str, optional): Code de la culture transportée (ex: 'maize').
                Influence la perte de qualité. Si None, utilise le facteur par défaut.

        Returns:
            dict: Dictionnaire contenant le coût total et le détail par poste :
                - 'total_cost_cfa' : coût total du transfert en XOF
                - 'details'        : sous-dictionnaire avec chaque composante
                - 'prob_pluie'     : probabilité de pluie utilisée (0 si pas de météo)
                - 'gamma_effectif' : gamma_route réellement appliqué
        """
        # Récupération du prix du carburant si non fourni
        if prix_carburant is None:
            prix_carburant = self._fetch_fuel_price()

        # Récupération des coefficients depuis la configuration centrale
        c_info = _LOGISTICS_CONFIG.get("c_info_xof", 5000.0)
        gamma_route_base = _LOGISTICS_CONFIG.get("gamma_route", 1.2)
        mu_checkpoints = _LOGISTICS_CONFIG.get("mu_checkpoints_xof_per_km", 15.0)

        # Probabilité de pluie (0.0 si pas de module météo disponible)
        prob_pluie = self._get_prob_pluie()

        # Coefficient de route effectif, ajusté selon la météo
        gamma_effectif = _calculer_gamma_effectif(gamma_route_base, prob_pluie)

        # Calcul de la distance routière entre les deux villes
        d_ab = self.get_distance(origine, destination)

        # Calcul du coût lié à la distance (carburant + tracasseries)
        cout_distance = d_ab * ((gamma_effectif * prix_carburant / 100.0) + mu_checkpoints)

        # Perte de qualité dynamique (par culture, distance et météo)
        c_qualite = _calculer_perte_qualite(crop, d_ab, prob_pluie)

        # Coût de transfert total
        c_transfer = c_info + cout_distance + c_qualite

        # Construction du résultat détaillé
        resultat = {
            "total_cost_cfa": round(c_transfer, 2),
            "prob_pluie": round(prob_pluie, 3),
            "gamma_effectif": round(gamma_effectif, 4),
            "details": {
                "distance_km": round(d_ab, 2),
                "search_costs": c_info,
                "transport_costs": round(cout_distance, 2),
                "quality_loss": c_qualite,
                "fuel_price_used": prix_carburant,
                "gamma_route_base": gamma_route_base,
                "gamma_effectif": round(gamma_effectif, 4),
                "prob_pluie": round(prob_pluie, 3),
                "crop": crop or "_default",
            },
        }

        return resultat

init ¶

__init__(cache_file: str = None, weather_session=None)

Initialise le module logistique.

Charge le cache persistant des coordonnées GPS et des distances routières pour éviter des appels réseau redondants.

Parameters:

Name	Type	Description	Default
`cache_file`	`str`	Chemin vers le fichier de cache JSON. Si None, utilise `~/.kadi/osrm_cache.json`.	`None`
`weather_session`	`WeatherSession`	Session météo (kadi.weather.WeatherSession) pour ajuster les coûts selon la météo prévue. Si None, pas d'ajustement climatique.	`None`

Source code in kadi/market/logistics.py

def __init__(self, cache_file: str = None, weather_session=None):
    """
    Initialise le module logistique.

    Charge le cache persistant des coordonnées GPS et des distances
    routières pour éviter des appels réseau redondants.

    Args:
        cache_file (str, optional): Chemin vers le fichier de cache JSON.
            Si None, utilise ``~/.kadi/osrm_cache.json``.
        weather_session (WeatherSession, optional): Session météo
            (kadi.weather.WeatherSession) pour ajuster les coûts selon
            la météo prévue. Si None, pas d'ajustement climatique.
    """
    # Session météo optionnelle (intégration Phase 4)
    self.weather_session = weather_session

    # Initialisation du chemin du fichier de cache
    if cache_file is None:
        cache_dir = os.path.expanduser("~/.kadi")
        os.makedirs(cache_dir, exist_ok=True)
        self.cache_file = os.path.join(cache_dir, "osrm_cache.json")
    else:
        self.cache_file = cache_file

    # Structure du cache en mémoire (coordonnées et distances routières)
    self.cache = {"coords": {}, "distances": {}}

    # Cache en mémoire pour le prix du carburant (évite les appels répétés)
    self._cached_fuel_price: Optional[float] = None

    # Cache en mémoire pour la probabilité de pluie (une seule requête par session)
    self._cached_prob_pluie: Optional[float] = None

    # Chargement du cache depuis le disque
    self._load_cache()

get_distance ¶

get_distance(origine: str, destination: str) -> float

Récupère la distance routière réelle entre deux villes béninoises.

Stratégie en cascade : 1. Cache local (résultat d'un appel précédent) 2. Géocodage Nominatim + routage OSRM 3. Fallback Haversine (vol d'oiseau * 1.3) si OSRM échoue 4. Valeur par défaut 100 km si le géocodage échoue

Parameters:

Name	Type	Description	Default
`origine`	`str`	Nom de la ville de départ (ex: 'Parakou').	required
`destination`	`str`	Nom de la ville d'arrivée (ex: 'Cotonou').	required

Returns:

Name	Type	Description
`float`	`float`	Distance estimée en kilomètres.

Source code in kadi/market/logistics.py

def get_distance(self, origine: str, destination: str) -> float:
    """
    Récupère la distance routière réelle entre deux villes béninoises.

    Stratégie en cascade :
    1. Cache local (résultat d'un appel précédent)
    2. Géocodage Nominatim + routage OSRM
    3. Fallback Haversine (vol d'oiseau * 1.3) si OSRM échoue
    4. Valeur par défaut 100 km si le géocodage échoue

    Args:
        origine (str): Nom de la ville de départ (ex: 'Parakou').
        destination (str): Nom de la ville d'arrivée (ex: 'Cotonou').

    Returns:
        float: Distance estimée en kilomètres.
    """
    orig_key = origine.strip().lower()
    dest_key = destination.strip().lower()
    route_key = f"{orig_key}_{dest_key}"
    rev_route_key = f"{dest_key}_{orig_key}"

    # 1. Vérification du cache (aller et retour)
    if route_key in self.cache["distances"]:
        return self.cache["distances"][route_key]
    if rev_route_key in self.cache["distances"]:
        return self.cache["distances"][rev_route_key]

    # 2. Géocodage des deux villes
    coords_orig = self._geocode_city(origine)
    coords_dest = self._geocode_city(destination)

    if not coords_orig or not coords_dest:
        logger.warning(
            f"Géocodage impossible pour '{origine}' -> '{destination}'. "
            "Valeur de repli : 100 km."
        )
        return 100.0

    # 3. Requête OSRM pour la vraie distance routière
    dist = self._fetch_osrm_distance(
        coords_orig[0], coords_orig[1],
        coords_dest[0], coords_dest[1],
    )

    # 4. Fallback Haversine si OSRM est indisponible
    if dist is None:
        logger.warning(
            f"OSRM indisponible pour '{origine}' -> '{destination}'. "
            "Utilisation de la formule Haversine (distance approximée)."
        )
        dist = self._haversine_distance(
            coords_orig[0], coords_orig[1],
            coords_dest[0], coords_dest[1],
        )

    # Mise en cache du résultat
    self.cache["distances"][route_key] = dist
    self._save_cache()

    return dist

calculate_transfer_cost ¶

calculate_transfer_cost(origine: str, destination: str, prix_carburant: float = None, crop: str = None) -> dict

Calcule le coût total de transfert d'un point A vers un point B.

Phase 4 - Intégration météo

Si un weather_session a été fourni à l'initialisation, le calcul ajuste automatiquement : - gamma_route selon la probabilité de pluie prévue demain. - La perte de qualité selon la culture et la météo.

Formule

C_transfer = C_info + Distance * (gamma_effectif * P_carburant / 100 + mu_checkpoints) + C_qualite(culture, distance, pluie)

Où : - C_info = coût fixe de recherche d'informations (appels, déplacements) - gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie) - P_carburant = prix du litre d'essence en XOF - mu_checkpoints = coût moyen des tracasseries par km - C_qualite = perte de valeur marchande (variable par culture et météo)

Parameters:

Name	Type	Description	Default
`origine`	`str`	Ville de départ (ex: 'Parakou').	required
`destination`	`str`	Ville de destination (ex: 'Cotonou').	required
`prix_carburant`	`float`	Prix du litre d'essence en XOF. Si None, récupéré automatiquement depuis les sources configurées.	`None`
`crop`	`str`	Code de la culture transportée (ex: 'maize'). Influence la perte de qualité. Si None, utilise le facteur par défaut.	`None`

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant le coût total et le détail par poste : - 'total_cost_cfa' : coût total du transfert en XOF - 'details' : sous-dictionnaire avec chaque composante - 'prob_pluie' : probabilité de pluie utilisée (0 si pas de météo) - 'gamma_effectif' : gamma_route réellement appliqué

Source code in kadi/market/logistics.py

def calculate_transfer_cost(
    self,
    origine: str,
    destination: str,
    prix_carburant: float = None,
    crop: str = None,
) -> dict:
    """
    Calcule le coût total de transfert d'un point A vers un point B.

    Phase 4 - Intégration météo :
        Si un weather_session a été fourni à l'initialisation, le calcul
        ajuste automatiquement :
        - gamma_route selon la probabilité de pluie prévue demain.
        - La perte de qualité selon la culture et la météo.

    Formule :
        C_transfer = C_info
                   + Distance * (gamma_effectif * P_carburant / 100 + mu_checkpoints)
                   + C_qualite(culture, distance, pluie)

    Où :
    - C_info = coût fixe de recherche d'informations (appels, déplacements)
    - gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie)
    - P_carburant = prix du litre d'essence en XOF
    - mu_checkpoints = coût moyen des tracasseries par km
    - C_qualite = perte de valeur marchande (variable par culture et météo)

    Args:
        origine (str): Ville de départ (ex: 'Parakou').
        destination (str): Ville de destination (ex: 'Cotonou').
        prix_carburant (float, optional): Prix du litre d'essence en XOF.
            Si None, récupéré automatiquement depuis les sources configurées.
        crop (str, optional): Code de la culture transportée (ex: 'maize').
            Influence la perte de qualité. Si None, utilise le facteur par défaut.

    Returns:
        dict: Dictionnaire contenant le coût total et le détail par poste :
            - 'total_cost_cfa' : coût total du transfert en XOF
            - 'details'        : sous-dictionnaire avec chaque composante
            - 'prob_pluie'     : probabilité de pluie utilisée (0 si pas de météo)
            - 'gamma_effectif' : gamma_route réellement appliqué
    """
    # Récupération du prix du carburant si non fourni
    if prix_carburant is None:
        prix_carburant = self._fetch_fuel_price()

    # Récupération des coefficients depuis la configuration centrale
    c_info = _LOGISTICS_CONFIG.get("c_info_xof", 5000.0)
    gamma_route_base = _LOGISTICS_CONFIG.get("gamma_route", 1.2)
    mu_checkpoints = _LOGISTICS_CONFIG.get("mu_checkpoints_xof_per_km", 15.0)

    # Probabilité de pluie (0.0 si pas de module météo disponible)
    prob_pluie = self._get_prob_pluie()

    # Coefficient de route effectif, ajusté selon la météo
    gamma_effectif = _calculer_gamma_effectif(gamma_route_base, prob_pluie)

    # Calcul de la distance routière entre les deux villes
    d_ab = self.get_distance(origine, destination)

    # Calcul du coût lié à la distance (carburant + tracasseries)
    cout_distance = d_ab * ((gamma_effectif * prix_carburant / 100.0) + mu_checkpoints)

    # Perte de qualité dynamique (par culture, distance et météo)
    c_qualite = _calculer_perte_qualite(crop, d_ab, prob_pluie)

    # Coût de transfert total
    c_transfer = c_info + cout_distance + c_qualite

    # Construction du résultat détaillé
    resultat = {
        "total_cost_cfa": round(c_transfer, 2),
        "prob_pluie": round(prob_pluie, 3),
        "gamma_effectif": round(gamma_effectif, 4),
        "details": {
            "distance_km": round(d_ab, 2),
            "search_costs": c_info,
            "transport_costs": round(cout_distance, 2),
            "quality_loss": c_qualite,
            "fuel_price_used": prix_carburant,
            "gamma_route_base": gamma_route_base,
            "gamma_effectif": round(gamma_effectif, 4),
            "prob_pluie": round(prob_pluie, 3),
            "crop": crop or "_default",
        },
    }

    return resultat

Logistique (kadi.market.logistics)¶

Formule de coût de transfert¶

Ajustement météo (Phase 4)¶

Initialisation¶

Méthodes¶

get_distance(origine, destination)¶

calculate_transfer_cost(origine, destination, prix_carburant, crop)¶

Facteurs de perte de qualité par culture¶

Prix du carburant¶

Exemple complet avec météo¶

MarketLogistics ¶

__init__ ¶

get_distance ¶

calculate_transfer_cost ¶

Logistique (`kadi.market.logistics`)¶

`get_distance(origine, destination)`¶

`calculate_transfer_cost(origine, destination, prix_carburant, crop)`¶

init ¶