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