fix specificity calculation for CBPE

nannyml/performance_calculation/metrics/binary_classification.py

@@ -1,12 +1,12 @@
 #  Author:   Niels Nuyttens  <niels@nannyml.com>
 #
 #  License: Apache Software License 2.0
+import warnings
 from typing import Any, Dict, List, Optional, Tuple, Union
 
 import numpy as np
 import pandas as pd
 from sklearn.metrics import confusion_matrix, f1_score, precision_score, recall_score, roc_auc_score
-import warnings
 
 from nannyml._typing import ProblemType
 from nannyml.base import _list_missing

nannyml/performance_calculation/metrics/multiclass_classification.py

@@ -7,24 +7,25 @@
 #  License: Apache Software License 2.0
 
 """Module containing metric utilities and implementations."""
+import warnings
 from typing import Dict, List, Optional, Tuple, Union  # noqa: TYP001
 
 import numpy as np
 import pandas as pd
-import warnings
 from sklearn.metrics import (
     accuracy_score,
+    confusion_matrix,
     f1_score,
     multilabel_confusion_matrix,
     precision_score,
     recall_score,
     roc_auc_score,
-    confusion_matrix,
 )
 from sklearn.preprocessing import LabelBinarizer, label_binarize
 
 from nannyml._typing import ProblemType, class_labels, model_output_column_names
 from nannyml.base import _list_missing
+from nannyml.chunk import Chunker
 from nannyml.exceptions import InvalidArgumentsException
 from nannyml.performance_calculation.metrics.base import Metric, MetricFactory, _common_data_cleaning
 from nannyml.sampling_error.multiclass_classification import (
@@ -44,7 +45,6 @@
     multiclass_confusion_matrix_sampling_error_components,
 )
 from nannyml.thresholds import Threshold, calculate_threshold_values
-from nannyml.chunk import Chunker
 
 
 @MetricFactory.register(metric='roc_auc', use_case=ProblemType.CLASSIFICATION_MULTICLASS)
@@ -636,7 +636,7 @@
         alert_thresholds = {}
 
         if self.classes is None:
             raise ValueError("classes must be set before calling this method")
 
         num_classes = len(self.classes)
 
@@ -674,7 +674,10 @@
         for true_class in classes:
             for pred_class in classes:
                 components.append(
-                    (f"true class: '{true_class}', predicted class: '{pred_class}'", f'true_{true_class}_pred_{pred_class}')
+                    (
+                        f"true class: '{true_class}', predicted class: '{pred_class}'",
+                        f'true_{true_class}_pred_{pred_class}',
+                    )
                 )
 
         return components
@@ -686,12 +689,12 @@
         y_pred = data[self.y_pred]
 
         if y_pred.isna().all().any():
             raise InvalidArgumentsException(
                 f"could not calculate metric {self.display_name}: prediction column contains no data"
             )
 
         if (y_true.nunique() <= 1) or (y_pred.nunique() <= 1):
             return np.nan
         else:
             cm = confusion_matrix(y_true, y_pred, labels=self.classes, normalize=self.normalize_confusion_matrix)
             return cm
@@ -699,13 +702,13 @@
     def get_chunk_record(self, chunk_data: pd.DataFrame) -> Dict[str, Union[float, bool]]:
 
         if self.classes is None:
             raise ValueError("classes must be set before calling this method")
 
         sampling_errors = multiclass_confusion_matrix_sampling_error(self.sampling_error_components, chunk_data)
         realized_cm = self._calculate(chunk_data)
 
         if isinstance(realized_cm, float):
             realized_cm = np.full((len(self.classes), len(self.classes)), np.nan)
 
         chunk_record = {}
 

nannyml/performance_estimation/confidence_based/metrics.py

@@ -11,6 +11,7 @@
 
 import abc
 import logging
+import warnings
 from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
 
 import numpy as np
@@ -402,6 +403,11 @@
         y_pred_proba, _, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized ROC-AUC.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized ROC-AUC.")
             return np.NaN
 
         return roc_auc_score(y_true, y_pred_proba)
@@ -494,6 +500,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized F1 score.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized F1 score.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized F1 score.")
             return np.NaN
 
         return f1_score(y_true=y_true, y_pred=y_pred)
@@ -570,6 +585,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized precision.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized precision.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized precision.")
             return np.NaN
 
         return precision_score(y_true=y_true, y_pred=y_pred)
@@ -644,6 +668,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized recall.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as recall precision.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as recall precision.")
             return np.NaN
 
         return recall_score(y_true=y_true, y_pred=y_pred)
@@ -718,10 +751,19 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized specificity.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized specificity.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized specificity.")
             return np.NaN
 
-        conf_matrix = confusion_matrix(y_true=y_true, y_pred=y_pred)
-        return conf_matrix[1, 1] / (conf_matrix[1, 0] + conf_matrix[1, 1])
+        tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel()
+        return tn / (tn + fp)
 
 
 def estimate_specificity(y_pred: pd.DataFrame, y_pred_proba: pd.DataFrame) -> float:
@@ -797,6 +839,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized accuracy.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized accuracy.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized accuracy.")
             return np.NaN
 
         return accuracy_score(y_true=y_true, y_pred=y_pred)
@@ -961,6 +1012,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized confusion matrix.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized confusion matrix.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized confusion matrix.")
             return np.NaN
 
         num_tp = np.sum(np.logical_and(y_pred, y_true))
@@ -980,6 +1040,7 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized confusion matrix.")
             return np.NaN
 
         num_tn = np.sum(np.logical_and(np.logical_not(y_pred), np.logical_not(y_true)))
@@ -999,6 +1060,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized confusion matrix.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized confusion matrix.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized confusion matrix.")
             return np.NaN
 
         num_tp = np.sum(np.logical_and(y_pred, y_true))
@@ -1018,6 +1088,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized confusion matrix.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized confusion matrix.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized confusion matrix.")
             return np.NaN
 
         num_tp = np.sum(np.logical_and(y_pred, y_true))
@@ -1500,6 +1579,15 @@
         _, y_pred, y_true = self._common_cleaning(data, y_pred_proba_column_name=self.uncalibrated_y_pred_proba)
 
         if y_true is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized business value.")
+            return np.NaN
+
+        if y_true.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized business value.")
+            return np.NaN
+
+        if y_pred.nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized business value.")
             return np.NaN
 
         tp_value = self.business_value_matrix[1, 1]
@@ -1677,7 +1765,13 @@
 
     def _realized_performance(self, data: pd.DataFrame) -> float:
         data = self._ensure_targets(data)
+
         if data is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized ROC-AUC.")
+            return np.NaN
+
+        if data[self.y_true].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized ROC-AUC.")
             return np.NaN
 
         _, y_pred_probas, labels = _get_multiclass_uncalibrated_predictions(data, self.y_pred, self.y_pred_proba)
@@ -1734,7 +1828,17 @@
 
     def _realized_performance(self, data: pd.DataFrame) -> float:
         data = self._ensure_targets(data)
+
         if data is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized F1 score.")
+            return np.NaN
+
+        if data[self.y_true].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized F1 score.")
+            return np.NaN
+
+        if data[self.y_pred].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized F1 score.")
             return np.NaN
 
         y_pred, _, labels = _get_multiclass_uncalibrated_predictions(data, self.y_pred, self.y_pred_proba)
@@ -1791,7 +1895,17 @@
 
     def _realized_performance(self, data: pd.DataFrame) -> float:
         data = self._ensure_targets(data)
+
         if data is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized precision.")
+            return np.NaN
+
+        if data[self.y_true].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized precision.")
+            return np.NaN
+
+        if data[self.y_pred].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized precision.")
             return np.NaN
 
         y_pred, _, labels = _get_multiclass_uncalibrated_predictions(data, self.y_pred, self.y_pred_proba)
@@ -1848,7 +1962,17 @@
 
     def _realized_performance(self, data: pd.DataFrame) -> float:
         data = self._ensure_targets(data)
+
         if data is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized recall.")
+            return np.NaN
+
+        if data[self.y_true].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized recall.")
+            return np.NaN
+
+        if data[self.y_pred].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized recall.")
             return np.NaN
 
         y_pred, _, labels = _get_multiclass_uncalibrated_predictions(data, self.y_pred, self.y_pred_proba)
@@ -1905,7 +2029,17 @@
 
     def _realized_performance(self, data: pd.DataFrame) -> float:
         data = self._ensure_targets(data)
+
         if data is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized specificity.")
+            return np.NaN
+
+        if data[self.y_true].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized specificity.")
+            return np.NaN
+
+        if data[self.y_pred].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized specificity.")
             return np.NaN
 
         y_pred, _, labels = _get_multiclass_uncalibrated_predictions(data, self.y_pred, self.y_pred_proba)
@@ -1964,8 +2098,19 @@
 
     def _realized_performance(self, data: pd.DataFrame) -> float:
         data = self._ensure_targets(data)
+
         if data is None:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized accuracy.")
             return np.NaN
+
+        if data[self.y_true].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized accuracy.")
+            return np.NaN
+
+        if data[self.y_pred].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized accuracy.")
+            return np.NaN
+
         y_pred, _, _ = _get_multiclass_uncalibrated_predictions(data, self.y_pred, self.y_pred_proba)
         return accuracy_score(data[self.y_true], y_pred)
 
@@ -1985,7 +2130,7 @@
     ):
 
         if isinstance(y_pred_proba, str):
             raise ValueError(
                 "y_pred_proba must be a dictionary with class labels as keys and pred_proba column names as values"
             )
 
@@ -2011,7 +2156,10 @@
         for true_class in classes:
             for pred_class in classes:
                 components.append(
-                    (f"true class: '{true_class}', predicted class: '{pred_class}'", f'true_{true_class}_pred_{pred_class}')
+                    (
+                        f"true class: '{true_class}', predicted class: '{pred_class}'",
+                        f'true_{true_class}_pred_{pred_class}',
+                    )
                 )
 
         return components
@@ -2074,9 +2222,17 @@
         return alert_thresholds
 
     def _multi_class_confusion_matrix_realized_performance(self, data: pd.DataFrame) -> Union[np.ndarray, float]:
+        if data is None or self.y_true not in data.columns:
+            warnings.warn("No 'y_true' values given for chunk, returning NaN as realized precision.")
+            return np.NaN
+
+        if data[self.y_true].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_true', returning NaN as realized precision.")
+            return np.NaN
 
-        if self.y_true not in data.columns or data[self.y_true].isna().all():
+        if data[self.y_pred].nunique() <= 1:
+            warnings.warn("Too few unique values present in 'y_pred', returning NaN as realized precision.")
             return np.NaN
 
         cm = confusion_matrix(
             data[self.y_true], data[self.y_pred], labels=self.classes, normalize=self.normalize_confusion_matrix
@@ -2105,7 +2261,7 @@
     def _get_multiclass_confusion_matrix_estimate(self, chunk_data: pd.DataFrame) -> np.ndarray:
 
         if isinstance(self.y_pred_proba, str):
             raise ValueError(
                 "y_pred_proba must be a dictionary with class labels as keys and pred_proba column names as values"
             )
 
@@ -2138,7 +2294,7 @@
         elif self.normalize_confusion_matrix == 'all':
             normalized_est_confusion_matrix = est_confusion_matrix / np.sum(est_confusion_matrix)
         else:
             raise ValueError(
                 f'normalize_confusion_matrix should be one of None, "true", \
                     "pred", or "all", but got {self.normalize_confusion_matrix}'
             )
@@ -2232,10 +2388,10 @@
         return chunk_record
 
     def _estimate(self, data: pd.DataFrame):
         pass
 
     def _sampling_error(self, data: pd.DataFrame) -> float:
         return 0.0
 
     def _realized_performance(self, data: pd.DataFrame) -> float:
         return 0.0