Source code for etna.metrics.utils

from typing import Callable
from typing import Dict
from typing import List
from typing import Union

import numpy as np
import pandas as pd
from typing_extensions import Literal

from etna.datasets import TSDataset
from etna.metrics import Metric

[docs]def compute_metrics( metrics: List[Metric], y_true: TSDataset, y_pred: TSDataset ) -> Dict[str, Union[float, Dict[str, float]]]: """ Compute metrics for given y_true, y_pred. Parameters ---------- metrics: list of metrics to compute y_true: dataset of true values of time series y_pred: dataset of time series forecast Returns ------- : dict of metrics in format {"metric_name": metric_value} """ metrics_values = {} for metric in metrics: metrics_values[metric.__repr__()] = metric(y_true=y_true, y_pred=y_pred) return metrics_values
[docs]def percentile(n: int): """Percentile for pandas agg.""" def percentile_(x): return np.nanpercentile(a=x.values, q=n) percentile_.__name__ = f"percentile_{n}" return percentile_
MetricAggregationStatistics = Literal[ "median", "mean", "std", "percentile_5", "percentile_25", "percentile_75", "percentile_95" ] METRICS_AGGREGATION_MAP: Dict[MetricAggregationStatistics, Union[str, Callable]] = { "median": "median", "mean": "mean", "std": "std", "percentile_5": percentile(5), "percentile_25": percentile(25), "percentile_75": percentile(75), "percentile_95": percentile(95), }
[docs]def aggregate_metrics_df(metrics_df: pd.DataFrame) -> Dict[str, float]: """Aggregate metrics in :py:meth:`log_backtest_metrics` method. Parameters ---------- metrics_df: Dataframe produced with :py:meth:`etna.pipeline.Pipeline._get_backtest_metrics` """ # case for aggregate_metrics=False if "fold_number" in metrics_df.columns: metrics_dict = ( metrics_df.groupby("segment") .mean() .reset_index() .drop(["segment", "fold_number"], axis=1) .apply(list(METRICS_AGGREGATION_MAP.values())) .to_dict() ) # case for aggregate_metrics=True else: metrics_dict = metrics_df.drop(["segment"], axis=1).apply(list(METRICS_AGGREGATION_MAP.values())).to_dict() metrics_dict_wide = { f"{metrics_key}_{statistics_key}": value for metrics_key, values in metrics_dict.items() for statistics_key, value in values.items() } return metrics_dict_wide