Source code for nannyml.performance_estimation.confidence_based.results

#  Author:   Niels Nuyttens  <>
#  License: Apache Software License 2.0

"""Module containing CBPE estimation results and plotting implementations."""
from __future__ import annotations

import copy
from typing import List, Optional, cast

import pandas as pd
from plotly import graph_objects as go

from nannyml._typing import Key, ModelOutputsType, ProblemType, Self
from nannyml.base import PerMetricResult
from nannyml.chunk import Chunker
from nannyml.exceptions import InvalidArgumentsException
from nannyml.performance_estimation.confidence_based import SUPPORTED_METRIC_FILTER_VALUES
from nannyml.performance_estimation.confidence_based.metrics import Metric
from nannyml.plots import Colors
from nannyml.plots.blueprints.comparisons import ResultCompareMixin
from nannyml.plots.blueprints.metrics import plot_metrics
from nannyml.usage_logging import UsageEvent, log_usage

[docs]class Result(PerMetricResult[Metric], ResultCompareMixin): """Contains results for CBPE estimation and adds filtering and plotting functionality.""" def __init__( self, results_data: pd.DataFrame, metrics: List[Metric], y_pred: str, y_pred_proba: ModelOutputsType, y_true: str, chunker: Chunker, problem_type: ProblemType, timestamp_column_name: Optional[str] = None, ): """Initialize CBPE results class. Parameters ---------- results_data: pd.DataFrame Results data returned by a CBPE estimator. metrics: List[nannyml.performance_estimation.confidence_based.metrics.Metric] List of metrics to evaluate. y_pred: str The name of the column containing your model predictions. y_pred_proba: Union[str, Dict[str, str]] Name(s) of the column(s) containing your model output. - For binary classification, pass a single string refering to the model output column. - For multiclass classification, pass a dictionary that maps a class string to the column name containing model outputs for that class. y_true: str The name of the column containing target values (that are provided in reference data during fitting). chunker: Chunker The `Chunker` used to split the data sets into a lists of chunks. problem_type: ProblemType Determines which CBPE implementation to use. Allowed problem type values are 'classification_binary' and 'classification_multiclass'. timestamp_column_name: str, default=None The name of the column containing the timestamp of the model prediction. If not given, plots will not use a time-based x-axis but will use the index of the chunks instead. """ super().__init__(results_data, metrics) self.y_pred = y_pred self.y_pred_proba = y_pred_proba self.y_true = y_true self.timestamp_column_name = timestamp_column_name self.problem_type = problem_type self.chunker = chunker def _filter(self, period: str, metrics: Optional[List[str]] = None, *args, **kwargs) -> Self: """Filter the results based on the specified period and metrics. This function begins by expanding the metrics to all the metrics that were specified or if no metrics were specified, all the metrics that were used to calculate the results. Since some metrics have multiple components, we expand these to their individual components. For example, the ``confusion_matrix`` metric has four components: ``true_positive``, ``true_negative``, ``false_positive``, and ``false_negative``. Specifying ``confusion_matrix`` or, for example, ``true_positive`` are both valid. We then filter the results based on the specified period and metrics. """ if metrics is None: filtered_metrics = self.metrics else: filtered_metrics = [] for name in metrics: if name not in SUPPORTED_METRIC_FILTER_VALUES: raise InvalidArgumentsException( f"invalid metric '{name}'. Please choose from {SUPPORTED_METRIC_FILTER_VALUES}" ) m = self._get_metric_by_name(name) if m: filtered_metrics = filtered_metrics + [m] else: raise InvalidArgumentsException(f"no '{name}' in result, did you calculate it?") metric_column_names = [name for metric in filtered_metrics for name in metric.column_names] res = super()._filter(period, metric_column_names, *args, **kwargs) return res def _get_metric_by_name(self, name: str) -> Optional[Metric]: for metric in self.metrics: # If we match the metric by name, return the metric # E.g. matching the name 'confusion_matrix' if name == return metric # If we match one of the metric component names # E.g. matching the name 'true_positive' with the confusion matrix metric elif name in metric.column_names: # Only retain the component whose column name was given to filter on res = copy.deepcopy(metric) res.components = list(filter(lambda c: c[1] == name, metric.components)) return res else: continue return None
[docs] def keys(self) -> List[Key]: """Creates a list of keys where each Key is a `namedtuple('Key', 'properties display_names')`.""" return [ Key( properties=(component[1],), display_names=( f'estimated {component[0]}', component[0],, ), ) for metric in self.metrics for component in cast(Metric, metric).components ]
[docs] @log_usage(UsageEvent.CBPE_PLOT, metadata_from_kwargs=['kind']) def plot( self, kind: str = 'performance', *args, **kwargs, ) -> go.Figure: """Render plots based on CBPE estimation results. This function will return a :class:`plotly.graph_objects.Figure` object. The following kinds of plots are available: Parameters ---------- kind: str, default='performance' What kind of plot to create. Only performance type is available. Raises ------ InvalidArgumentsException: when an unknown plot ``kind`` is provided. Returns ------- fig: :class:`plotly.graph_objs._figure.Figure` A :class:`~plotly.graph_objs._figure.Figure` object containing the requested drift plot. Can be saved to disk using the :meth:`~plotly.graph_objs._figure.Figure.write_image` method or shown rendered on screen using the :meth:`` method. Examples -------- >>> import nannyml as nml >>> from IPython.display import display >>> reference_df = nml.load_synthetic_car_loan_dataset()[0] >>> analysis_df = nml.load_synthetic_car_loan_dataset()[1] >>> display(reference_df.head(3)) >>> estimator = nml.CBPE( ... y_pred_proba='y_pred_proba', ... y_pred='y_pred', ... y_true='repaid', ... timestamp_column_name='timestamp', ... metrics=['roc_auc', 'accuracy', 'f1'], ... chunk_size=5000, ... problem_type='classification_binary', >>> ) >>> >>> results = estimator.estimate(analysis_df) >>> display(results.filter(period='analysis').to_df()) >>> metric_fig = results.plot() >>> """ if kind == 'performance': return plot_metrics( self, title='Estimated performance <b>(CBPE)</b>', subplot_title_format='Estimated <b>{display_names[1]}</b>', subplot_y_axis_title_format='{display_names[1]}', color=Colors.INDIGO_PERSIAN, line_dash='dash', ) else: raise InvalidArgumentsException(f"unknown plot kind '{kind}'. " f"Please provide on of: ['performance'].")