bikes.core.metrics

  1"""Evaluate model performances with metrics."""
  2
  3# %% IMPORTS
  4
  5from __future__ import annotations
  6
  7import abc
  8import typing as T
  9
 10import mlflow
 11import pandas as pd
 12import pydantic as pdt
 13from sklearn import metrics
 14
 15from bikes.core import models, schemas
 16
 17# %% TYPINGS
 18
 19MlflowMetric: T.TypeAlias = mlflow.metrics.MetricValue
 20MlflowThreshold: T.TypeAlias = mlflow.models.MetricThreshold
 21MlflowModelValidationFailedException: T.TypeAlias = (
 22    mlflow.models.evaluation.validation.ModelValidationFailedException
 23)
 24
 25# %% METRICS
 26
 27
 28class Metric(abc.ABC, pdt.BaseModel, strict=True, frozen=True, extra="forbid"):
 29    """Base class for a project metric.
 30
 31    Use metrics to evaluate model performance.
 32    e.g., accuracy, precision, recall, MAE, F1, ...
 33
 34    Parameters:
 35        name (str): name of the metric for the reporting.
 36        greater_is_better (bool): maximize or minimize result.
 37    """
 38
 39    KIND: str
 40
 41    name: str
 42    greater_is_better: bool
 43
 44    @abc.abstractmethod
 45    def score(self, targets: schemas.Targets, outputs: schemas.Outputs) -> float:
 46        """Score the outputs against the targets.
 47
 48        Args:
 49            targets (schemas.Targets): expected values.
 50            outputs (schemas.Outputs): predicted values.
 51
 52        Returns:
 53            float: single result from the metric computation.
 54        """
 55
 56    def scorer(
 57        self, model: models.Model, inputs: schemas.Inputs, targets: schemas.Targets
 58    ) -> float:
 59        """Score model outputs against targets.
 60
 61        Args:
 62            model (models.Model): model to evaluate.
 63            inputs (schemas.Inputs): model inputs values.
 64            targets (schemas.Targets): model expected values.
 65
 66        Returns:
 67            float: single result from the metric computation.
 68        """
 69        outputs = model.predict(inputs=inputs)
 70        score = self.score(targets=targets, outputs=outputs)
 71        return score
 72
 73    def to_mlflow(self) -> MlflowMetric:
 74        """Convert the metric to an Mlflow metric.
 75
 76        Returns:
 77            MlflowMetric: the Mlflow metric.
 78        """
 79
 80        def eval_fn(predictions: pd.Series[int], targets: pd.Series[int]) -> MlflowMetric:
 81            """Evaluation function associated with the mlflow metric.
 82
 83            Args:
 84                predictions (pd.Series): model predictions.
 85                targets (pd.Series | None): model targets.
 86
 87            Returns:
 88                MlflowMetric: the mlflow metric.
 89            """
 90            score_targets = schemas.Targets(
 91                {schemas.TargetsSchema.cnt: targets}, index=targets.index
 92            )
 93            score_outputs = schemas.Outputs(
 94                {schemas.OutputsSchema.prediction: predictions}, index=predictions.index
 95            )
 96            sign = 1 if self.greater_is_better else -1  # reverse the effect
 97            score = self.score(targets=score_targets, outputs=score_outputs)
 98            return MlflowMetric(aggregate_results={self.name: score * sign})
 99
100        return mlflow.metrics.make_metric(
101            eval_fn=eval_fn, name=self.name, greater_is_better=self.greater_is_better
102        )
103
104
105class SklearnMetric(Metric):
106    """Compute metrics with sklearn.
107
108    Parameters:
109        name (str): name of the sklearn metric.
110        greater_is_better (bool): maximize or minimize.
111    """
112
113    KIND: T.Literal["SklearnMetric"] = "SklearnMetric"
114
115    name: str = "mean_squared_error"
116    greater_is_better: bool = False
117
118    @T.override
119    def score(self, targets: schemas.Targets, outputs: schemas.Outputs) -> float:
120        metric = getattr(metrics, self.name)
121        sign = 1 if self.greater_is_better else -1
122        y_true = targets[schemas.TargetsSchema.cnt]
123        y_pred = outputs[schemas.OutputsSchema.prediction]
124        score = metric(y_pred=y_pred, y_true=y_true) * sign
125        return float(score)
126
127
128MetricKind = SklearnMetric
129
130# %% THRESHOLDS
131
132
133class Threshold(abc.ABC, pdt.BaseModel, strict=True, frozen=True, extra="forbid"):
134    """A project threshold for a metric.
135
136    Use thresholds to monitor model performances.
137    e.g., to trigger an alert when a threshold is met.
138
139    Parameters:
140        threshold (int | float): absolute threshold value.
141        greater_is_better (bool): maximize or minimize result.
142    """
143
144    threshold: int | float
145    greater_is_better: bool
146
147    def to_mlflow(self) -> MlflowThreshold:
148        """Convert the threshold to an mlflow threshold.
149
150        Returns:
151            MlflowThreshold: the mlflow threshold.
152        """
153        return MlflowThreshold(threshold=self.threshold, greater_is_better=self.greater_is_better)