DualFlow/common/metrics/classification.py

"""
分类任务评估指标
"""
import numpy as np
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score, f1_score,
    confusion_matrix, classification_report, roc_auc_score
)
from typing import List, Dict, Any, Optional, Union
import logging

logger = logging.getLogger(__name__)


class ClassificationMetrics:
    """分类任务评估指标"""
    
    def __init__(self):
        """初始化分类指标"""
        pass
    
    def accuracy(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """准确率"""
        return accuracy_score(y_true, y_pred)
    
    def precision(self, y_true: np.ndarray, y_pred: np.ndarray, 
                  average: str = 'weighted') -> float:
        """精确率"""
        return precision_score(y_true, y_pred, average=average, zero_division=0)
    
    def recall(self, y_true: np.ndarray, y_pred: np.ndarray, 
               average: str = 'weighted') -> float:
        """召回率"""
        return recall_score(y_true, y_pred, average=average, zero_division=0)
    
    def f1_score(self, y_true: np.ndarray, y_pred: np.ndarray, 
                 average: str = 'weighted') -> float:
        """F1分数"""
        return f1_score(y_true, y_pred, average=average, zero_division=0)
    
    def confusion_matrix(self, y_true: np.ndarray, y_pred: np.ndarray) -> np.ndarray:
        """混淆矩阵"""
        return confusion_matrix(y_true, y_pred)
    
    def classification_report(self, y_true: np.ndarray, y_pred: np.ndarray, 
                            target_names: Optional[List[str]] = None) -> str:
        """分类报告"""
        return classification_report(y_true, y_pred, target_names=target_names)
    
    def roc_auc(self, y_true: np.ndarray, y_pred_proba: np.ndarray, 
                average: str = 'weighted') -> float:
        """ROC AUC分数"""
        try:
            return roc_auc_score(y_true, y_pred_proba, average=average)
        except ValueError as e:
            logger.warning(f"无法计算ROC AUC: {e}")
            return 0.0
    
    def compute_all_metrics(self, y_true: np.ndarray, y_pred: np.ndarray, 
                           y_pred_proba: Optional[np.ndarray] = None) -> Dict[str, float]:
        """计算所有指标"""
        metrics = {
            'accuracy': self.accuracy(y_true, y_pred),
            'precision': self.precision(y_true, y_pred),
            'recall': self.recall(y_true, y_pred),
            'f1_score': self.f1_score(y_true, y_pred)
        }
        
        if y_pred_proba is not None:
            metrics['roc_auc'] = self.roc_auc(y_true, y_pred_proba)
        
        return metrics


class MultiClassMetrics:
    """多分类任务评估指标"""
    
    def __init__(self):
        """初始化多分类指标"""
        pass
    
    def macro_precision(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """宏平均精确率"""
        return precision_score(y_true, y_pred, average='macro', zero_division=0)
    
    def macro_recall(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """宏平均召回率"""
        return recall_score(y_true, y_pred, average='macro', zero_division=0)
    
    def macro_f1(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """宏平均F1分数"""
        return f1_score(y_true, y_pred, average='macro', zero_division=0)
    
    def micro_precision(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """微平均精确率"""
        return precision_score(y_true, y_pred, average='micro', zero_division=0)
    
    def micro_recall(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """微平均召回率"""
        return recall_score(y_true, y_pred, average='micro', zero_division=0)
    
    def micro_f1(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """微平均F1分数"""
        return f1_score(y_true, y_pred, average='micro', zero_division=0)
    
    def per_class_metrics(self, y_true: np.ndarray, y_pred: np.ndarray) -> Dict[str, Dict[str, float]]:
        """每个类别的指标"""
        unique_labels = np.unique(np.concatenate([y_true, y_pred]))
        per_class = {}
        
        for label in unique_labels:
            # 二分类指标
            y_true_binary = (y_true == label).astype(int)
            y_pred_binary = (y_pred == label).astype(int)
            
            tp = np.sum((y_true_binary == 1) & (y_pred_binary == 1))
            fp = np.sum((y_true_binary == 0) & (y_pred_binary == 1))
            fn = np.sum((y_true_binary == 1) & (y_pred_binary == 0))
            
            precision = tp / (tp + fp) if (tp + fp) > 0 else 0
            recall = tp / (tp + fn) if (tp + fn) > 0 else 0
            f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
            
            per_class[str(label)] = {
                'precision': precision,
                'recall': recall,
                'f1_score': f1
            }
        
        return per_class


class BinaryClassificationMetrics:
    """二分类任务评估指标"""
    
    def __init__(self):
        """初始化二分类指标"""
        pass
    
    def true_positive(self, y_true: np.ndarray, y_pred: np.ndarray) -> int:
        """真正例"""
        return np.sum((y_true == 1) & (y_pred == 1))
    
    def false_positive(self, y_true: np.ndarray, y_pred: np.ndarray) -> int:
        """假正例"""
        return np.sum((y_true == 0) & (y_pred == 1))
    
    def true_negative(self, y_true: np.ndarray, y_pred: np.ndarray) -> int:
        """真负例"""
        return np.sum((y_true == 0) & (y_pred == 0))
    
    def false_negative(self, y_true: np.ndarray, y_pred: np.ndarray) -> int:
        """假负例"""
        return np.sum((y_true == 1) & (y_pred == 0))
    
    def sensitivity(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """敏感性 (召回率)"""
        tp = self.true_positive(y_true, y_pred)
        fn = self.false_negative(y_true, y_pred)
        return tp / (tp + fn) if (tp + fn) > 0 else 0
    
    def specificity(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """特异性"""
        tn = self.true_negative(y_true, y_pred)
        fp = self.false_positive(y_true, y_pred)
        return tn / (tn + fp) if (tn + fp) > 0 else 0
    
    def positive_predictive_value(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """阳性预测值 (精确率)"""
        tp = self.true_positive(y_true, y_pred)
        fp = self.false_positive(y_true, y_pred)
        return tp / (tp + fp) if (tp + fp) > 0 else 0
    
    def negative_predictive_value(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """阴性预测值"""
        tn = self.true_negative(y_true, y_pred)
        fn = self.false_negative(y_true, y_pred)
        return tn / (tn + fn) if (tn + fn) > 0 else 0