src/interfaces/metrics_calculator.py

import cupy as cp
from typing import Dict, Protocol, Tuple
import warnings
# Utiliser cuml.metrics pour les calculs accélérés par GPU
from cuml.metrics import accuracy_score, precision_recall_curve, roc_auc_score

# Filtrer les avertissements
warnings.filterwarnings("ignore", category=Warning)


class MetricsCalculator(Protocol):
    """
    Interface pour les calculateurs de métriques pour la classification binaire.
    """
    def calculate_and_log(
        self,
        y_true: cp.ndarray,
        y_pred: cp.ndarray,
        y_proba: cp.ndarray, # Probabilités classe positive (1D) - Supposées toujours fournies
        prefix: str
    ) -> Dict[str, float]:
        """
        Calcule les métriques pour un problème binaire.
        Assume que y_proba est toujours fourni et est un tableau 1D
        contenant les probabilités de la classe positive.
        Retourne Accuracy, AUC ROC, Precision, Recall et F1 Score.
        """
        pass


class DefaultMetricsCalculator(MetricsCalculator):
    """
    Implémentation concrète de MetricsCalculator utilisant cuML pour la classification binaire.
    Calcule l'accuracy, l'AUC-ROC, la précision, le rappel et le F1 score en utilisant les fonctions cuML.
    Utilise precision_recall_curve pour calculer les métriques de précision, rappel et F1 score optimales.
    Assume que les données d'entrée sont valides et que y_proba est toujours fourni
    en tant que tableau 1D des probabilités de la classe positive.
    """

    def calculate_and_log(
        self,
        y_true: cp.ndarray,
        y_pred: cp.ndarray,
        y_proba: cp.ndarray, # Probabilités classe positive (1D) - Supposées toujours fournies
        prefix: str
    ) -> Dict[str, float]:
        """
        Calcule l'accuracy, l'AUC ROC, la précision, le rappel et le F1 score pour un problème binaire en utilisant cuML.
        Utilise precision_recall_curve pour calculer les métriques optimales.
        Assume des entrées valides et que y_proba est un tableau 1D fourni.
        """
        # 1. Calculer l'accuracy (comme dans l'exemple accuracy_score)
        acc = accuracy_score(y_true, y_pred)

        # 2. Calculer l'AUC binaire (comme dans l'exemple roc_auc_score)
        auc = roc_auc_score(y_true.astype(cp.int32), y_proba.astype(cp.float32))

        # 3. Utiliser precision_recall_curve pour obtenir les courbes
        precision, recall, thresholds = precision_recall_curve(
            y_true.astype(cp.int32), y_proba.astype(cp.float32)
        )
        
        # 4. Calculer la précision, le rappel et le F1 score optimaux
        optimal_precision, optimal_recall, optimal_f1, optimal_threshold = self._calculate_optimal_f1(
            precision, recall, thresholds
        )

        # Construire le dictionnaire des métriques scalaires disponibles
        metrics = {
            f"{prefix}_accuracy"           : acc,
            f"{prefix}_precision"          : optimal_precision,
            f"{prefix}_recall"             : optimal_recall,
            f"{prefix}_f1"                 : optimal_f1,
            f"{prefix}_optimal_threshold"  : optimal_threshold,
            f"{prefix}_auc_roc"            : auc
        }
        # Retourner les métriques scalaires calculées
        return metrics
        
    def _calculate_optimal_f1(
        self, 
        precision: cp.ndarray, 
        recall: cp.ndarray, 
        thresholds: cp.ndarray
    ) -> Tuple[float, float, float, float]:
        """
        Calcule le F1 score optimal à partir des courbes de précision et de rappel.
        
        Args:
            precision: Tableau de précisions pour différents seuils
            recall: Tableau de rappels pour différents seuils
            thresholds: Tableau de seuils correspondants
            
        Returns:
            Tuple contenant (précision optimale, rappel optimal, F1 score optimal, seuil optimal)
        """
        # Ajouter le seuil 1.0 à thresholds (qui n'est pas inclus par défaut dans precision_recall_curve)
        thresholds_with_one = cp.append(thresholds, cp.array([1.0]))
            
        # Calculer le F1 score pour chaque point de la courbe
        # F1 = 2 * (precision * recall) / (precision + recall)
        f1_scores = 2 * (precision * recall) / (precision + recall)
        
        # Trouver l'indice du F1 score maximal
        best_idx = cp.argmax(f1_scores)
        best_precision = float(precision[best_idx])
        best_recall = float(recall[best_idx])
        best_f1 = float(f1_scores[best_idx])
        
        # Obtenir le seuil optimal
        best_threshold = float(thresholds_with_one[best_idx])
            
        return best_precision, best_recall, best_f1, best_threshold