diff --git a/nannyml/performance_estimation/confidence_based/cbpe.py b/nannyml/performance_estimation/confidence_based/cbpe.py index 2bfa7287..add5e577 100644 --- a/nannyml/performance_estimation/confidence_based/cbpe.py +++ b/nannyml/performance_estimation/confidence_based/cbpe.py @@ -298,14 +298,6 @@ def __init__( self.result: Optional[Result] = None - 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 - @log_usage(UsageEvent.CBPE_ESTIMATOR_FIT, metadata_from_self=['metrics', 'problem_type']) def _fit(self, reference_data: pd.DataFrame, *args, **kwargs) -> CBPE: """Fits the drift calculator using a set of reference data. diff --git a/nannyml/performance_estimation/confidence_based/metrics.py b/nannyml/performance_estimation/confidence_based/metrics.py index b92e0d3a..f6663edc 100644 --- a/nannyml/performance_estimation/confidence_based/metrics.py +++ b/nannyml/performance_estimation/confidence_based/metrics.py @@ -94,7 +94,6 @@ def __init__( Notes ----- - The `components` approach taken here is a quick fix to deal with metrics that return multiple values. Look at the `confusion_matrix` for example: a single metric produces 4 different result sets (containing values, thresholds, alerts, etc.). @@ -123,25 +122,25 @@ def _logger(self) -> logging.Logger: return logging.getLogger(__name__) @property - def display_name(self) -> str: + def display_name(self) -> str: # noqa: D102 return self.name @property - def column_name(self) -> str: + def column_name(self) -> str: # noqa: D102 return self.components[0][1] @property - def display_names(self): + def display_names(self): # noqa: D102 return [c[0] for c in self.components] @property - def column_names(self): + def column_names(self): # noqa: D102 return [c[1] for c in self.components] - def __str__(self): + def __str__(self): # noqa: D105 return self.display_name - def __repr__(self): + def __repr__(self): # noqa: D105 return self.column_name def fit(self, reference_data: pd.DataFrame): @@ -211,6 +210,19 @@ def alert(self, value: float) -> bool: ) def __eq__(self, other): + """Compares two Metric instances. + + They are considered equal when their components are equal. + + Parameters + ---------- + other: Metric + The other Metric instance you're comparing to. + + Returns + ------- + is_equal: bool + """ return self.components == other.components def _common_cleaning( @@ -239,6 +251,7 @@ def get_chunk_record(self, chunk_data: pd.DataFrame) -> Dict: ---------- chunk_data : pd.DataFrame A pandas dataframe containing the data for a given chunk. + Raises ------ NotImplementedError: occurs when a metric has multiple componets @@ -296,6 +309,7 @@ def _logger(cls) -> logging.Logger: @classmethod def create(cls, key: str, use_case: ProblemType, **kwargs) -> Metric: + """Create new Metric.""" if kwargs is None: kwargs = {} @@ -321,6 +335,7 @@ def create(cls, key: str, use_case: ProblemType, **kwargs) -> Metric: @classmethod def register(cls, metric: str, use_case: ProblemType) -> Callable: + """Register a Metric in the MetricFactory registry.""" def inner_wrapper(wrapped_class: Type[Metric]) -> Type[Metric]: if metric in cls.registry: if use_case in cls.registry[metric]: @@ -335,6 +350,7 @@ def inner_wrapper(wrapped_class: Type[Metric]) -> Type[Metric]: @MetricFactory.register('roc_auc', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationAUROC(Metric): + """CBPE binary classification AUROC Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -345,6 +361,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification AUROC Metric Class.""" super().__init__( name='roc_auc', y_pred_proba=y_pred_proba, @@ -428,6 +445,7 @@ def estimate_roc_auc(y_pred_proba: pd.Series) -> float: @MetricFactory.register('f1', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationF1(Metric): + """CBPE binary classification f1 Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -438,6 +456,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification f1 Metric Class.""" super().__init__( name='f1', y_pred_proba=y_pred_proba, @@ -512,6 +531,7 @@ def estimate_f1(y_pred: pd.DataFrame, y_pred_proba: pd.DataFrame) -> float: @MetricFactory.register('precision', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationPrecision(Metric): + """CBPE binary classification precision Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -522,6 +542,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification precision Metric Class.""" super().__init__( name='precision', y_pred_proba=y_pred_proba, @@ -596,6 +617,7 @@ def estimate_precision(y_pred: pd.DataFrame, y_pred_proba: pd.DataFrame) -> floa @MetricFactory.register('recall', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationRecall(Metric): + """CBPE binary classification recall Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -606,6 +628,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification recall Metric Class.""" super().__init__( name='recall', y_pred_proba=y_pred_proba, @@ -679,6 +702,7 @@ def estimate_recall(y_pred: pd.DataFrame, y_pred_proba: pd.DataFrame) -> float: @MetricFactory.register('specificity', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationSpecificity(Metric): + """CBPE binary classification specificity Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -689,6 +713,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification specificity Metric Class.""" super().__init__( name='specificity', y_pred_proba=y_pred_proba, @@ -763,6 +788,7 @@ def estimate_specificity(y_pred: pd.DataFrame, y_pred_proba: pd.DataFrame) -> fl @MetricFactory.register('accuracy', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationAccuracy(Metric): + """CBPE binary classification accuracy Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -773,6 +799,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification accuracy Metric Class.""" super().__init__( name='accuracy', y_pred_proba=y_pred_proba, @@ -828,6 +855,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('confusion_matrix', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationConfusionMatrix(Metric): + """CBPE binary classification confusion matrix Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -839,6 +867,7 @@ def __init__( normalize_confusion_matrix: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification confusion matrix Metric Class.""" super().__init__( name='confusion_matrix', y_pred_proba=y_pred_proba, @@ -870,6 +899,7 @@ def __init__( def fit(self, reference_data: pd.DataFrame): # override the superclass fit method """Fits a Metric on reference data. + Parameters ---------- reference_data: pd.DataFrame @@ -1283,18 +1313,14 @@ def get_true_pos_info(self, chunk_data: pd.DataFrame) -> Dict: true_pos_info['sampling_error_true_positive'] = sampling_error_true_positives true_pos_info['realized_true_positive'] = self._true_positive_realized_performance(chunk_data) - if self.normalize_confusion_matrix is None: - true_pos_info['upper_confidence_boundary_true_positive'] = ( - estimated_true_positives + SAMPLING_ERROR_RANGE * sampling_error_true_positives - ) - else: - true_pos_info['upper_confidence_boundary_true_positive'] = np.minimum( - self.upper_threshold_value_limit, - estimated_true_positives + SAMPLING_ERROR_RANGE * sampling_error_true_positives, - ) + true_pos_info['upper_confidence_boundary_true_positive'] = np.minimum( + self.upper_threshold_value_limit or np.inf, + estimated_true_positives + SAMPLING_ERROR_RANGE * sampling_error_true_positives, + ) true_pos_info['lower_confidence_boundary_true_positive'] = np.maximum( - self.lower_threshold_value_limit, estimated_true_positives - SAMPLING_ERROR_RANGE * sampling_error_true_positives + self.lower_threshold_value_limit or -np.inf, + estimated_true_positives - SAMPLING_ERROR_RANGE * sampling_error_true_positives ) true_pos_info['upper_threshold_true_positive'] = self.true_positive_upper_threshold @@ -1335,18 +1361,14 @@ def get_true_neg_info(self, chunk_data: pd.DataFrame) -> Dict: true_neg_info['sampling_error_true_negative'] = sampling_error_true_negatives true_neg_info['realized_true_negative'] = self._true_negative_realized_performance(chunk_data) - if self.normalize_confusion_matrix is None: - true_neg_info['upper_confidence_boundary_true_negative'] = ( - estimated_true_negatives + SAMPLING_ERROR_RANGE * sampling_error_true_negatives - ) - else: - true_neg_info['upper_confidence_boundary_true_negative'] = np.minimum( - self.upper_threshold_value_limit, - estimated_true_negatives + SAMPLING_ERROR_RANGE * sampling_error_true_negatives, - ) + true_neg_info['upper_confidence_boundary_true_negative'] = np.minimum( + self.upper_threshold_value_limit or np.inf, + estimated_true_negatives + SAMPLING_ERROR_RANGE * sampling_error_true_negatives, + ) true_neg_info['lower_confidence_boundary_true_negative'] = np.maximum( - self.lower_threshold_value_limit, estimated_true_negatives - SAMPLING_ERROR_RANGE * sampling_error_true_negatives + self.lower_threshold_value_limit or -np.inf, + estimated_true_negatives - SAMPLING_ERROR_RANGE * sampling_error_true_negatives ) true_neg_info['upper_threshold_true_negative'] = self.true_negative_upper_threshold @@ -1387,18 +1409,13 @@ def get_false_pos_info(self, chunk_data: pd.DataFrame) -> Dict: false_pos_info['sampling_error_false_positive'] = sampling_error_false_positives false_pos_info['realized_false_positive'] = self._false_positive_realized_performance(chunk_data) - if self.normalize_confusion_matrix is None: - false_pos_info['upper_confidence_boundary_false_positive'] = ( - estimated_false_positives + SAMPLING_ERROR_RANGE * sampling_error_false_positives - ) - else: - false_pos_info['upper_confidence_boundary_false_positive'] = np.minimum( - self.upper_threshold_value_limit, - estimated_false_positives + SAMPLING_ERROR_RANGE * sampling_error_false_positives, - ) + false_pos_info['upper_confidence_boundary_false_positive'] = np.minimum( + self.upper_threshold_value_limit or np.inf, + estimated_false_positives + SAMPLING_ERROR_RANGE * sampling_error_false_positives, + ) false_pos_info['lower_confidence_boundary_false_positive'] = np.maximum( - self.lower_threshold_value_limit, + self.lower_threshold_value_limit or -np.inf, estimated_false_positives - SAMPLING_ERROR_RANGE * sampling_error_false_positives, ) @@ -1440,18 +1457,13 @@ def get_false_neg_info(self, chunk_data: pd.DataFrame) -> Dict: false_neg_info['sampling_error_false_negative'] = sampling_error_false_negatives false_neg_info['realized_false_negative'] = self._false_negative_realized_performance(chunk_data) - if self.normalize_confusion_matrix is None: - false_neg_info['upper_confidence_boundary_false_negative'] = ( - estimated_false_negatives + SAMPLING_ERROR_RANGE * sampling_error_false_negatives - ) - else: - false_neg_info['upper_confidence_boundary_false_negative'] = np.minimum( - self.upper_threshold_value_limit, - estimated_false_negatives + SAMPLING_ERROR_RANGE * sampling_error_false_negatives, - ) + false_neg_info['upper_confidence_boundary_false_negative'] = np.minimum( + self.upper_threshold_value_limit or np.inf, + estimated_false_negatives + SAMPLING_ERROR_RANGE * sampling_error_false_negatives, + ) false_neg_info['lower_confidence_boundary_false_negative'] = np.maximum( - self.lower_threshold_value_limit, + self.lower_threshold_value_limit or -np.inf, estimated_false_negatives - SAMPLING_ERROR_RANGE * sampling_error_false_negatives, ) @@ -1469,6 +1481,18 @@ def get_false_neg_info(self, chunk_data: pd.DataFrame) -> Dict: return false_neg_info def get_chunk_record(self, chunk_data: pd.DataFrame) -> Dict: + """Returns a dictionary containing the performance metrics for a given chunk. + + Parameters + ---------- + chunk_data : pd.DataFrame + A pandas dataframe containing the data for a given chunk. + + Returns + ------- + chunk_record : Dict + A dictionary of perfomance metric, value pairs. + """ chunk_record = {} true_pos_info = self.get_true_pos_info(chunk_data) @@ -1497,6 +1521,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('business_value', ProblemType.CLASSIFICATION_BINARY) class BinaryClassificationBusinessValue(Metric): + """CBPE binary classification business value Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -1509,6 +1534,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE binary classification business value Metric Class.""" super().__init__( name='business_value', y_pred_proba=y_pred_proba, @@ -1606,15 +1632,19 @@ def estimate_business_value( normalize_business_value: str, default=None Determines how the business value will be normalized. Allowed values are None and 'per_prediction'. - - None - the business value will not be normalized and the value returned will be the total value per chunk. - - 'per_prediction' - the value will be normalized by the number of predictions in the chunk. + - None - the business value will not be normalized and the value returned will be the total value per chunk. + - 'per_prediction' - the value will be normalized by the number of predictions in the chunk. + + business_value_matrix: np.ndarray + A 2x2 matrix that specifies the value of each cell in the confusion matrix. + The format of the business value matrix must be specified as [[value_of_TN, value_of_FP], \ + [value_of_FN, value_of_TP]]. Returns ------- business_value: float Estimated Business Value score. """ - est_tn_ratio = np.mean(np.where(y_pred == 0, 1 - y_pred_proba, 0)) est_tp_ratio = np.mean(np.where(y_pred == 1, y_pred_proba, 0)) est_fp_ratio = np.mean(np.where(y_pred == 1, 1 - y_pred_proba, 0)) @@ -1682,6 +1712,7 @@ def _ensure_targets(self, data: pd.DataFrame) -> Optional[pd.DataFrame]: @MetricFactory.register('roc_auc', ProblemType.CLASSIFICATION_MULTICLASS) class MulticlassClassificationAUROC(_MulticlassClassificationMetric): + """CBPE multiclass classification AUROC Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -1692,6 +1723,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE multiclass classification AUROC Metric Class.""" super().__init__( name='roc_auc', y_pred_proba=y_pred_proba, @@ -1746,6 +1778,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('f1', ProblemType.CLASSIFICATION_MULTICLASS) class MulticlassClassificationF1(_MulticlassClassificationMetric): + """CBPE multiclass classification f1 Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -1756,6 +1789,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE multiclass classification f1 Metric Class.""" super().__init__( name='f1', y_pred_proba=y_pred_proba, @@ -1813,6 +1847,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('precision', ProblemType.CLASSIFICATION_MULTICLASS) class MulticlassClassificationPrecision(_MulticlassClassificationMetric): + """CBPE multiclass classification precision Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -1823,6 +1858,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE multiclass classification precision Metric Class.""" super().__init__( name='precision', y_pred_proba=y_pred_proba, @@ -1880,6 +1916,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('recall', ProblemType.CLASSIFICATION_MULTICLASS) class MulticlassClassificationRecall(_MulticlassClassificationMetric): + """CBPE multiclass classification recall Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -1890,6 +1927,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE multiclass classification recall Metric Class.""" super().__init__( name='recall', y_pred_proba=y_pred_proba, @@ -1947,6 +1985,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('specificity', ProblemType.CLASSIFICATION_MULTICLASS) class MulticlassClassificationSpecificity(_MulticlassClassificationMetric): + """CBPE multiclass classification specificity Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -1957,6 +1996,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE multiclass classification specificity Metric Class.""" super().__init__( name='specificity', y_pred_proba=y_pred_proba, @@ -2018,6 +2058,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('accuracy', ProblemType.CLASSIFICATION_MULTICLASS) class MulticlassClassificationAccuracy(_MulticlassClassificationMetric): + """CBPE multiclass classification accuracy Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -2028,6 +2069,7 @@ def __init__( timestamp_column_name: Optional[str] = None, **kwargs, ): + """Initialize CBPE multiclass classification accuracy Metric Class.""" super().__init__( name='accuracy', y_pred_proba=y_pred_proba, @@ -2082,6 +2124,7 @@ def _realized_performance(self, data: pd.DataFrame) -> float: @MetricFactory.register('confusion_matrix', ProblemType.CLASSIFICATION_MULTICLASS) class MulticlassClassificationConfusionMatrix(Metric): + """CBPE multiclass classification confusion matrix Metric Class.""" def __init__( self, y_pred_proba: ModelOutputsType, @@ -2093,6 +2136,7 @@ def __init__( normalize_confusion_matrix: Optional[str] = None, **kwargs, ): + """Initialize CBPE multiclass classification confusion matrix Metric Class.""" 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" @@ -2135,6 +2179,7 @@ def _get_components(self, classes: List[str]) -> List[Tuple[str, str]]: def fit(self, reference_data: pd.DataFrame): # override the superclass fit method """Fits a Metric on reference data. + Parameters ---------- reference_data: pd.DataFrame @@ -2247,6 +2292,18 @@ def _get_multiclass_confusion_matrix_estimate(self, chunk_data: pd.DataFrame) -> return normalized_est_confusion_matrix def get_chunk_record(self, chunk_data: pd.DataFrame) -> Dict: + """Returns a dictionary containing the performance metrics for a given chunk. + + Parameters + ---------- + chunk_data : pd.DataFrame + A pandas dataframe containing the data for a given chunk. + + Returns + ------- + chunk_record : Dict + A dictionary of perfomance metric, value pairs. + """ chunk_record = {} estimated_cm = self._get_multiclass_confusion_matrix_estimate(chunk_data) diff --git a/nannyml/performance_estimation/confidence_based/results.py b/nannyml/performance_estimation/confidence_based/results.py index 8de0ffd7..308b897d 100644 --- a/nannyml/performance_estimation/confidence_based/results.py +++ b/nannyml/performance_estimation/confidence_based/results.py @@ -37,7 +37,8 @@ def __init__( problem_type: ProblemType, timestamp_column_name: Optional[str] = None, ): - """ + """Initialize CBPE results class. + Parameters ---------- results_data: pd.DataFrame @@ -124,9 +125,7 @@ def _get_metric_by_name(self, name: str) -> Optional[Metric]: return None def keys(self) -> List[Key]: - """ - Creates a list of keys where each Key is a `namedtuple('Key', 'properties display_names')` - """ + """Creates a list of keys where each Key is a `namedtuple('Key', 'properties display_names')`.""" return [ Key( properties=(component[1],), @@ -155,6 +154,8 @@ def plot( Parameters ---------- kind: str, default='performance' + What kind of plot to create. Only performance type is available. + Raises ------