# Author: Niels Nuyttens <niels@nannyml.com>
#
# License: Apache Software License 2.0
"""Implementation of the CBPE estimator."""
import copy
from abc import abstractmethod
from typing import Dict, List, Tuple, Union
import pandas as pd
from pandas import MultiIndex
from nannyml._typing import ModelOutputsType, ProblemType
from nannyml.base import AbstractEstimator
from nannyml.calibration import Calibrator, CalibratorFactory
from nannyml.chunk import Chunker
from nannyml.exceptions import InvalidArgumentsException
from nannyml.performance_estimation.confidence_based.metrics import MetricFactory
from nannyml.performance_estimation.confidence_based.results import SUPPORTED_METRIC_VALUES, Result
[docs]class CBPE(AbstractEstimator):
"""Performance estimator using the Confidence Based Performance Estimation (CBPE) technique."""
[docs] def __new__(cls, y_pred_proba: ModelOutputsType, problem_type: Union[str, ProblemType], *args, **kwargs):
"""Creates a new CBPE subclass instance based on the type of the provided ``model_metadata``."""
from ._cbpe_binary_classification import _BinaryClassificationCBPE
from ._cbpe_multiclass_classification import _MulticlassClassificationCBPE
if isinstance(problem_type, str):
problem_type = ProblemType.parse(problem_type)
if problem_type is ProblemType.CLASSIFICATION_BINARY:
return super(CBPE, cls).__new__(_BinaryClassificationCBPE)
elif problem_type is ProblemType.CLASSIFICATION_MULTICLASS:
return super(CBPE, cls).__new__(_MulticlassClassificationCBPE)
else:
raise NotImplementedError
def __init__(
self,
metrics: List[str],
y_pred: str,
y_pred_proba: ModelOutputsType,
y_true: str,
problem_type: Union[str, ProblemType],
timestamp_column_name: str = None,
chunk_size: int = None,
chunk_number: int = None,
chunk_period: str = None,
chunker: Chunker = None,
calibration: str = None,
calibrator: Calibrator = None,
):
"""Initializes a new CBPE performance estimator.
Parameters
----------
y_true: str
The name of the column containing target values (that are provided in reference data during fitting).
y_pred_proba: ModelOutputsType
Name(s) of the column(s) containing your model output.
Pass a single string when there is only a single model output column, e.g. in binary classification cases.
Pass a dictionary when working with multiple output columns, e.g. in multiclass classification cases.
The dictionary maps a class/label string to the column name containing model outputs for that class/label.
y_pred: str
The name of the column containing your model predictions.
timestamp_column_name: str, default=None
The name of the column containing the timestamp of the model prediction.
metrics: List[str]
A list of metrics to calculate.
chunk_size: int, default=None
Splits the data into chunks containing `chunks_size` observations.
Only one of `chunk_size`, `chunk_number` or `chunk_period` should be given.
chunk_number: int, default=None
Splits the data into `chunk_number` pieces.
Only one of `chunk_size`, `chunk_number` or `chunk_period` should be given.
chunk_period: str, default=None
Splits the data according to the given period.
Only one of `chunk_size`, `chunk_number` or `chunk_period` should be given.
chunker : Chunker, default=None
The `Chunker` used to split the data sets into a lists of chunks.
calibration: str, default='isotonic'
Determines which calibration will be applied to the model predictions. Defaults to ``isotonic``, currently
the only supported value.
calibrator: Calibrator, default=None
A specific instance of a Calibrator to be applied to the model predictions.
If not set NannyML will use the value of the ``calibration`` variable instead.
problem_type: Union[str, ProblemType]
Determines which CBPE implementation to use. Allowed problem type values are 'classification_binary' and
'classification_multiclass'.
Examples
--------
>>> import nannyml as nml
>>> from IPython.display import display
>>> reference_df = nml.load_synthetic_binary_classification_dataset()[0]
>>> analysis_df = nml.load_synthetic_binary_classification_dataset()[1]
>>> display(reference_df.head(3))
>>> estimator = nml.CBPE(
... y_pred_proba='y_pred_proba',
... y_pred='y_pred',
... y_true='work_home_actual',
... timestamp_column_name='timestamp',
... metrics=['roc_auc', 'f1'],
... chunk_size=5000,
... problem_type='classification_binary',
>>> )
>>> estimator.fit(reference_df)
>>> results = estimator.estimate(analysis_df)
>>> display(results.data)
>>> for metric in estimator.metrics:
... metric_fig = results.plot(kind='performance', metric=metric)
... metric_fig.show()
>>> for metric in estimator.metrics:
... metric_fig = results.plot(kind='performance', plot_reference=True, metric=metric)
... metric_fig.show()
"""
super().__init__(chunk_size, chunk_number, chunk_period, chunker, timestamp_column_name)
self.y_true = y_true
self.y_pred = y_pred
self.y_pred_proba = y_pred_proba
if metrics is None or len(metrics) == 0:
raise InvalidArgumentsException(
"no metrics provided. Please provide a non-empty list of metrics."
f"Supported values are {SUPPORTED_METRIC_VALUES}."
)
if isinstance(problem_type, str):
self.problem_type = ProblemType.parse(problem_type)
else:
self.problem_type = problem_type
self.metrics = [
MetricFactory.create(
metric,
self.problem_type,
y_pred_proba=self.y_pred_proba,
y_pred=self.y_pred,
y_true=self.y_true,
timestamp_column_name=self.timestamp_column_name,
chunker=self.chunker,
)
for metric in metrics
]
self._confidence_deviations: Dict[str, float] = {}
self._alert_thresholds: Dict[str, Tuple[float, float]] = {}
self.needs_calibration: bool = False
if calibrator is None:
calibrator = CalibratorFactory.create(calibration)
self.calibrator = calibrator
self.minimum_chunk_size: int = None # type: ignore
def __deepcopy__(self, memodict={}):
cls = self.__class__
result = cls.__new__(cls, y_pred_proba=self.y_pred_proba, problem_type=self.problem_type)
memodict[id(self)] = result
for k, v in self.__dict__.items():
setattr(result, k, copy.deepcopy(v, memodict))
return result
@abstractmethod
def _fit(self, reference_data: pd.DataFrame, *args, **kwargs) -> AbstractEstimator:
"""Fits the drift calculator using a set of reference data.
Parameters
----------
reference_data : pd.DataFrame
A reference data set containing predictions (labels and/or probabilities) and target values.
Returns
-------
estimator: PerformanceEstimator
The fitted estimator.
Examples
--------
>>> import nannyml as nml
>>> ref_df, ana_df, _ = nml.load_synthetic_binary_classification_dataset()
>>> metadata = nml.extract_metadata(ref_df, model_type=nml.ModelType.CLASSIFICATION_BINARY)
>>> # create a new estimator and fit it on reference data
>>> estimator = nml.CBPE(model_metadata=metadata, chunk_period='W').fit(ref_df)
"""
pass
@abstractmethod
def _estimate(self, data: pd.DataFrame, *args, **kwargs) -> Result:
"""Calculates the data reconstruction drift for a given data set.
Parameters
----------
data : pd.DataFrame
The dataset to calculate the reconstruction drift for.
Returns
-------
estimates: PerformanceEstimatorResult
A :class:`result<nannyml.performance_estimation.confidence_based.results.Result>`
object where each row represents a :class:`~nannyml.chunk.Chunk`,
containing :class:`~nannyml.chunk.Chunk` properties and the estimated metrics
for that :class:`~nannyml.chunk.Chunk`.
Examples
--------
>>> import nannyml as nml
>>> ref_df, ana_df, _ = nml.load_synthetic_binary_classification_dataset()
>>> metadata = nml.extract_metadata(ref_df, model_type=nml.ModelType.CLASSIFICATION_BINARY)
>>> # create a new estimator and fit it on reference data
>>> estimator = nml.CBPE(model_metadata=metadata, chunk_period='W').fit(ref_df)
>>> estimates = estimator.estimate(data)
"""
pass
def _create_multilevel_index(metric_names: List[str]):
chunk_column_names = [
'key',
'chunk_index',
'start_index',
'end_index',
'start_date',
'end_date',
'period',
]
method_column_names = [
'sampling_error',
'realized',
'value',
'upper_confidence_boundary',
'lower_confidence_boundary',
'upper_threshold',
'lower_threshold',
'alert',
]
chunk_tuples = [('chunk', chunk_column_name) for chunk_column_name in chunk_column_names]
reconstruction_tuples = [
(metric_name, column_name) for metric_name in metric_names for column_name in method_column_names
]
tuples = chunk_tuples + reconstruction_tuples
return MultiIndex.from_tuples(tuples)