src/api/monitoring/drift.py

"""
Ce module contient les fonctions de calcul du PSI (Population Stability Index) pour les variables d'entrée utilisées dans le modèle de prédiction de rendement agricole.
Il inclut :
- get_input_drift_summary() : Calcule un résumé des statistiques descriptives pour les variables d'entrée à partir des données de production.
- get_numeric_psi() : Calcule le PSI pour les variables d'entrée numériques en comparant la distribution de référence d'entraînement avec la distribution des données reçues en production.
- get_distribution_comparison(feature_name) : Compare la distribution d'une variable d'entrée spécifique entre
la référence d'entraînement et les données reçues en production, et retourne les valeurs pour les deux distributions.
"""

import math
from src.api.monitoring.db import get_connection



def get_input_drift_summary():
    """Calcule un résumé des statistiques descriptives pour les variables d'entrée
    à partir des données de production.
    Returns:
        dict: Un dictionnaire contenant les statistiques descriptives pour chaque variable d'entrée.
    """
    query = """
    SELECT
        input_json
    FROM api_requests
    WHERE input_json IS NOT NULL;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query)
            rows = cursor.fetchall()

    inputs = [row[0] for row in rows]

    numeric_fields = ["temp", "rainfall_mm", "pesticides", "trend"]

    summary = {}

    for field in numeric_fields:
        values = [
            float(input_data[field])
            for input_data in inputs
            if field in input_data and input_data[field] is not None
        ]

        if values:
            summary[field] = {
                "count": len(values),
                "mean": round(sum(values) / len(values), 2),
                "min": round(min(values), 2),
                "max": round(max(values), 2),
            }
        else:
            summary[field] = {
                "count": 0,
                "mean": None,
                "min": None,
                "max": None,
            }

    return summary

import math

def _calculate_psi(reference_values, production_values, bins=10):
    """
    Calcule le PSI entre une distribution de référence
    et une distribution de production.
    """

    if not reference_values or not production_values:
        return None

    min_value = min(reference_values)
    max_value = max(reference_values)

    if min_value == max_value:
        return 0

    bin_width = (max_value - min_value) / bins

    psi = 0

    for i in range(bins):
        lower = min_value + i * bin_width
        upper = min_value + (i + 1) * bin_width

        if i == bins - 1:
            ref_count = sum(lower <= value <= upper for value in reference_values)
            prod_count = sum(lower <= value <= upper for value in production_values)
        else:
            ref_count = sum(lower <= value < upper for value in reference_values)
            prod_count = sum(lower <= value < upper for value in production_values)

        ref_pct = ref_count / len(reference_values)
        prod_pct = prod_count / len(production_values)

        ref_pct = max(ref_pct, 0.0001)
        prod_pct = max(prod_pct, 0.0001)

        psi += (prod_pct - ref_pct) * math.log(prod_pct / ref_pct)

    return round(psi, 4)


def _get_psi_status(psi):
    if psi is None:
        return "Données insuffisantes"

    if psi < 0.1:
        return "Stable"

    if psi < 0.2:
        return "Légère dérive"

    return "Drift important"


def get_numeric_psi():
    features = ["temp", "rainfall_mm", "pesticides", "trend"]

    results = []

    with get_connection() as conn:
        with conn.cursor() as cursor:

            for feature in features:
                cursor.execute(
                    """
                    SELECT feature_value
                    FROM reference_distributions
                    WHERE feature_name = %s
                      AND source = 'train';
                    """,
                    (feature,)
                )

                reference_values = [
                    float(row[0])
                    for row in cursor.fetchall()
                    if row[0] is not None
                ]

                cursor.execute(
                    """
                    SELECT input_json
                    FROM api_requests
                    WHERE input_json IS NOT NULL;
                    """
                )

                production_inputs = cursor.fetchall()

                production_values = []

                for row in production_inputs:
                    input_data = row[0]

                    if feature in input_data and input_data[feature] is not None:
                        production_values.append(float(input_data[feature]))

                psi = _calculate_psi(reference_values, production_values)

                results.append(
                    {
                        "feature": feature,
                        "psi": psi,
                        "status": _get_psi_status(psi),
                        "reference_count": len(reference_values),
                        "production_count": len(production_values),
                    }
                )

    return results

def get_distribution_comparison(feature_name):
    """
    Compare la distribution d'une variable entre :
    - la référence d'entraînement
    - les données reçues en production
    """

    allowed_features = [
        "temp",
        "rainfall_mm",
        "pesticides",
        "trend",
    ]

    if feature_name not in allowed_features:
        raise ValueError(
            f"Variable non autorisée : {feature_name}"
        )

    with get_connection() as conn:
        with conn.cursor() as cursor:

            cursor.execute(
                """
                SELECT feature_value
                FROM reference_distributions
                WHERE feature_name = %s
                  AND source = 'train';
                """,
                (feature_name,)
            )

            reference_values = [
                float(row[0])
                for row in cursor.fetchall()
                if row[0] is not None
            ]

            cursor.execute(
                """
                SELECT input_json
                FROM api_requests
                WHERE input_json IS NOT NULL;
                """
            )

            production_inputs = cursor.fetchall()

    production_values = []

    for row in production_inputs:
        input_data = row[0]

        if (
            feature_name in input_data
            and input_data[feature_name] is not None
        ):
            production_values.append(
                float(input_data[feature_name])
            )

    return {
        "feature": feature_name,
        "reference": reference_values,
        "production": production_values,
    }