Fix: regression metric plots should have a lower limit of 0 for confi…

…dence bounds and thresholds (#127)

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   multiclass use cases [(#98)](https://github.com/NannyML/nannyml/issues/98)
 - Fix an issue where reference data was rendered incorrectly on joy plots
 - Updated the 'California Housing' example docs, thanks for the help [@NeoKish](https://github.com/NeoKish)
+- Fix lower confidence bounds and thresholds under zero for regression cases. When the lower limit is set to 0,
+  the lower threshold will not be plotted. [(#127)](https://github.com/NannyML/nannyml/issues/127)
 
 ## [0.6.2] - 2022-09-16
 

nannyml/performance_calculation/calculator.py

@@ -193,7 +193,7 @@ def _calculate_metrics_for_chunk(self, chunk: Chunk) -> Dict:
             metrics_results[f'{metric.column_name}_upper_threshold'] = metric.upper_threshold
             metrics_results[f'{metric.column_name}_sampling_error'] = metric.sampling_error(chunk.data)
             metrics_results[f'{metric.column_name}_alert'] = (
-                metric.lower_threshold > chunk_metric or chunk_metric > metric.upper_threshold
-            )
+                metric.lower_threshold > chunk_metric if metric.lower_threshold else False
+            ) or (chunk_metric > metric.upper_threshold if metric.upper_threshold else False)
 
         return metrics_results

nannyml/performance_calculation/metrics/base.py

@@ -132,15 +132,15 @@ def _calculate_alert_thresholds(
         std_num: int = 3,
         lower_limit: Optional[float] = None,
         upper_limit: Optional[float] = None,
-    ) -> Tuple[float, float]:
+    ) -> Tuple[Optional[float], Optional[float]]:
         chunked_reference_metric = [self.calculate(chunk.data) for chunk in reference_chunks]
         deviation = np.std(chunked_reference_metric) * std_num
         mean_reference_metric = np.mean(chunked_reference_metric)
         lower_threshold = mean_reference_metric - deviation
-        if lower_limit:
+        if lower_limit is not None:
             lower_threshold = np.maximum(lower_threshold, lower_limit)
         upper_threshold = mean_reference_metric + deviation
-        if upper_limit:
+        if upper_limit is not None:
             upper_threshold = np.minimum(upper_threshold, upper_limit)
         return lower_threshold, upper_threshold
 

nannyml/performance_calculation/metrics/regression.py

@@ -1,7 +1,8 @@
 #  Author:   Niels Nuyttens  <niels@nannyml.com>
 #
 #  License: Apache Software License 2.0
-from typing import Tuple
+from abc import ABC
+from typing import List, Optional, Tuple
 
 import pandas as pd
 from sklearn.metrics import (
@@ -13,6 +14,7 @@
 
 from nannyml._typing import ProblemType
 from nannyml.base import _list_missing, _raise_exception_for_negative_values
+from nannyml.chunk import Chunk
 from nannyml.performance_calculation.metrics.base import Metric, MetricFactory, _common_data_cleaning
 from nannyml.sampling_error.regression import (
     mae_sampling_error,
@@ -30,8 +32,28 @@
 )
 
 
+class RegressionMetric(Metric, ABC):
+    def __init__(self, *args, **kwargs):
+        super().__init__(lower_threshold_limit=0, *args, **kwargs)
+
+    def _calculate_alert_thresholds(
+        self,
+        reference_chunks: List[Chunk],
+        std_num: int = 3,
+        lower_limit: Optional[float] = None,
+        upper_limit: Optional[float] = None,
+    ) -> Tuple[Optional[float], Optional[float]]:
+        lower_threshold, upper_threshold = super()._calculate_alert_thresholds(
+            reference_chunks, std_num, lower_limit, upper_limit
+        )
+        if lower_threshold == 0.0:
+            return None, upper_threshold
+        else:
+            return lower_threshold, upper_threshold
+
+
 @MetricFactory.register(metric='mae', use_case=ProblemType.REGRESSION)
-class MAE(Metric):
+class MAE(RegressionMetric):
     """Mean Absolute Error metric."""
 
     def __init__(self, calculator):
@@ -67,7 +89,7 @@ def _sampling_error(self, data: pd.DataFrame) -> float:
 
 
 @MetricFactory.register(metric='mape', use_case=ProblemType.REGRESSION)
-class MAPE(Metric):
+class MAPE(RegressionMetric):
     """Mean Absolute Percentage Error metric."""
 
     def __init__(self, calculator):
@@ -103,7 +125,7 @@ def _sampling_error(self, data: pd.DataFrame) -> float:
 
 
 @MetricFactory.register(metric='mse', use_case=ProblemType.REGRESSION)
-class MSE(Metric):
+class MSE(RegressionMetric):
     """Mean Squared Error metric."""
 
     def __init__(self, calculator):
@@ -139,7 +161,7 @@ def _sampling_error(self, data: pd.DataFrame) -> float:
 
 
 @MetricFactory.register(metric='msle', use_case=ProblemType.REGRESSION)
-class MSLE(Metric):
+class MSLE(RegressionMetric):
     """Mean Squared Logarithmic Error metric."""
 
     def __init__(self, calculator):
@@ -180,7 +202,7 @@ def _sampling_error(self, data: pd.DataFrame) -> float:
 
 
 @MetricFactory.register(metric='rmse', use_case=ProblemType.REGRESSION)
-class RMSE(Metric):
+class RMSE(RegressionMetric):
     """Root Mean Squared Error metric."""
 
     def __init__(self, calculator):
@@ -216,7 +238,7 @@ def _sampling_error(self, data: pd.DataFrame) -> float:
 
 
 @MetricFactory.register(metric='rmsle', use_case=ProblemType.REGRESSION)
-class RMSLE(Metric):
+class RMSLE(RegressionMetric):
     """Root Mean Squared Logarithmic Error metric."""
 
     def __init__(self, calculator):

nannyml/performance_estimation/direct_loss_estimation/dle.py

@@ -257,14 +257,23 @@ def _estimate_chunk(self, chunk: Chunk) -> Dict:
         for metric in self.metrics:
             estimated_metric = metric.estimate(chunk.data)
             sampling_error = metric.sampling_error(chunk.data)
+
+            upper_confidence_boundary = estimated_metric + 3 * sampling_error
+            if metric.upper_value_limit is not None:
+                upper_confidence_boundary = min(metric.upper_value_limit, upper_confidence_boundary)
+
+            lower_confidence_boundary = estimated_metric - 3 * sampling_error
+            if metric.lower_value_limit is not None:
+                lower_confidence_boundary = max(metric.lower_value_limit, lower_confidence_boundary)
+
             estimates[f'realized_{metric.column_name}'] = metric.realized_performance(chunk.data)
             estimates[f'estimated_{metric.column_name}'] = estimated_metric
-            estimates[f'upper_confidence_{metric.column_name}'] = estimated_metric + 3 * sampling_error
-            estimates[f'lower_confidence_{metric.column_name}'] = estimated_metric - 3 * sampling_error
+            estimates[f'upper_confidence_{metric.column_name}'] = upper_confidence_boundary
+            estimates[f'lower_confidence_{metric.column_name}'] = lower_confidence_boundary
             estimates[f'sampling_error_{metric.column_name}'] = sampling_error
             estimates[f'upper_threshold_{metric.column_name}'] = metric.upper_threshold
             estimates[f'lower_threshold_{metric.column_name}'] = metric.lower_threshold
             estimates[f'alert_{metric.column_name}'] = (
-                estimated_metric > metric.upper_threshold or estimated_metric < metric.lower_threshold
-            )
+                estimated_metric > metric.upper_threshold if metric.upper_threshold else False
+            ) or (estimated_metric < metric.lower_threshold if metric.lower_threshold else False)
         return estimates

nannyml/performance_estimation/direct_loss_estimation/metrics.py

@@ -45,8 +45,8 @@ def __init__(
         display_name: str,
         column_name: str,
         estimator: AbstractEstimator,
-        upper_threshold_limit: float = None,
-        lower_threshold_limit: float = None,
+        upper_value_limit: float = None,
+        lower_value_limit: float = 0.0,
     ):
         """Creates a new Metric instance.
 
@@ -70,8 +70,8 @@ def __init__(
 
         self.upper_threshold: Optional[float] = None
         self.lower_threshold: Optional[float] = None
-        self.upper_threshold_limit: Optional[float] = upper_threshold_limit
-        self.lower_threshold_limit: Optional[float] = lower_threshold_limit
+        self.upper_value_limit: Optional[float] = upper_value_limit
+        self.lower_value_limit: Optional[float] = lower_value_limit
 
         self._dee_model: LGBMRegressor
 
@@ -99,7 +99,9 @@ def fit(self, reference_data: pd.DataFrame):
         reference_chunks = self.estimator.chunker.split(
             reference_data,
         )
-        self.lower_threshold, self.upper_threshold = self._alert_thresholds(reference_chunks)
+        self.lower_threshold, self.upper_threshold = self._alert_thresholds(
+            reference_chunks, lower_limit=self.lower_value_limit, upper_limit=self.upper_value_limit
+        )
 
         # Delegate to subclass
         self._fit(reference_data)
@@ -156,15 +158,20 @@ def _sampling_error(self, data: pd.DataFrame) -> float:
 
     def _alert_thresholds(
         self, reference_chunks: List[Chunk], std_num: int = 3, lower_limit: float = None, upper_limit: float = None
-    ) -> Tuple[float, float]:
+    ) -> Tuple[Optional[float], Optional[float]]:
         realized_chunk_performance = [self.realized_performance(chunk.data) for chunk in reference_chunks]
         deviation = np.std(realized_chunk_performance) * std_num
         mean_realised_performance = np.mean(realized_chunk_performance)
         lower_threshold = mean_realised_performance - deviation
-        if lower_limit:
+        if lower_limit is not None:
             lower_threshold = np.maximum(lower_threshold, lower_limit)
+
+        # Special case... in case lower threshold equals 0, it should not be shown at all
+        if lower_threshold == 0.0:
+            lower_threshold = None
+
         upper_threshold = mean_realised_performance + deviation
-        if upper_limit:
+        if upper_limit is not None:
             upper_threshold = np.minimum(upper_threshold, upper_limit)
 
         return lower_threshold, upper_threshold

nannyml/plots/_step_plot.py

@@ -622,25 +622,27 @@ def _plot_thresholds(
 ):
     if lower_threshold_column_name and lower_threshold_column_name in data.columns:
         threshold_value = data[lower_threshold_column_name].values[0]
-        fig.add_hline(
-            threshold_value,
-            annotation_text=threshold_value_format.format(threshold_value),
-            annotation_position="top right",
-            annotation=dict(font_color=colors[-1]),
-            line=dict(color=colors[-1], width=1, dash='dash'),
-            layer='below',
-        )
+        if threshold_value is not None:
+            fig.add_hline(
+                threshold_value,
+                annotation_text=threshold_value_format.format(threshold_value),
+                annotation_position="top right",
+                annotation=dict(font_color=colors[-1]),
+                line=dict(color=colors[-1], width=1, dash='dash'),
+                layer='below',
+            )
 
     if upper_threshold_column_name and upper_threshold_column_name in data.columns:
         threshold_value = data[upper_threshold_column_name].values[0]
-        fig.add_hline(
-            threshold_value,
-            annotation_text=threshold_value_format.format(threshold_value),
-            annotation_position="top right",
-            annotation=dict(font_color=colors[-1]),
-            line=dict(color=colors[-1], width=1, dash='dash'),
-            layer='below',
-        )
+        if threshold_value is not None:
+            fig.add_hline(
+                threshold_value,
+                annotation_text=threshold_value_format.format(threshold_value),
+                annotation_position="top right",
+                annotation=dict(font_color=colors[-1]),
+                line=dict(color=colors[-1], width=1, dash='dash'),
+                layer='below',
+            )
 
 
 def _plot_reference_analysis_separator(