Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from typing import Any, Iterable | |
| import numpy as np | |
| import pandas as pd | |
| from gluonts.dataset.common import ListDataset | |
| from gluonts.dataset.split import split | |
| from gluonts.model import Predictor, evaluate_model | |
| from gluonts.time_feature import get_seasonality | |
| from tsfm_bench.data.interface import TimeSeriesDataSource | |
| from tsfm_bench.data.open_meteo import OpenMeteoDataSource | |
| from tsfm_bench.data.ts_bench import TsBenchDataSource | |
| from tsfm_bench.eval.metrics import RESULT_COLUMNS, get_eval_metrics | |
| class OnlineEvalResult: | |
| dataset: str | |
| model: str | |
| metrics: dict[str, float] | |
| domain: str | |
| num_variates: int | |
| data_fetched_at: str | |
| context_length: int | |
| prediction_length: int | |
| forecast_snapshot: dict | None = None | |
| def _records_to_entries( | |
| record, | |
| to_univariate: bool, | |
| ) -> list[dict[str, Any]]: | |
| target = np.asarray(record.target, dtype=np.float32) | |
| start = pd.Period(record.start, freq=record.freq) | |
| entries: list[dict[str, Any]] = [] | |
| if target.ndim == 1 or not to_univariate: | |
| entries.append( | |
| { | |
| "item_id": record.item_id, | |
| "start": start, | |
| "target": target if target.ndim == 1 else target, | |
| "freq": record.freq, | |
| } | |
| ) | |
| return entries | |
| for idx in range(target.shape[0]): | |
| entries.append( | |
| { | |
| "item_id": f"{record.item_id}_v{idx}", | |
| "start": start, | |
| "target": target[idx], | |
| "freq": record.freq, | |
| } | |
| ) | |
| return entries | |
| def resolve_seasonality(freq: str, series_length: int, prediction_length: int) -> int: | |
| """Cap seasonality so MASE/MSIS scaling is defined on short live windows.""" | |
| nominal = int(get_seasonality(freq)) | |
| train_length = series_length - prediction_length | |
| # Need at least two seasonal periods in the training window. | |
| capped = min(nominal, max(1, train_length // 2)) | |
| return max(1, capped) | |
| def evaluate_online_window( | |
| predictor: Predictor, | |
| entries: list[dict[str, Any]], | |
| prediction_length: int, | |
| freq: str, | |
| ) -> dict[str, float]: | |
| dataset = ListDataset(entries, freq=freq) | |
| _, test_template = split(dataset, offset=-prediction_length) | |
| test_data = test_template.generate_instances( | |
| prediction_length=prediction_length, | |
| windows=1, | |
| distance=prediction_length, | |
| ) | |
| series_length = int(np.asarray(entries[0]["target"]).shape[-1]) | |
| seasonality = resolve_seasonality(freq, series_length, prediction_length) | |
| evaluated = evaluate_model( | |
| predictor, | |
| test_data=test_data, | |
| metrics=get_eval_metrics(), | |
| batch_size=64, | |
| axis=None, | |
| mask_invalid_label=True, | |
| allow_nan_forecast=True, | |
| seasonality=seasonality, | |
| ) | |
| result = {key: float(value.iloc[0]) for key, value in evaluated.items()} | |
| return result | |
| def _capture_forecast_snapshot( | |
| predictor: Predictor, | |
| entries: list[dict], | |
| pred_len: int, | |
| freq: str, | |
| evaluated_at: str, | |
| dataset: str, | |
| model: str, | |
| ) -> dict | None: | |
| """Call predictor once to capture p10/p50/p90 and actuals for visualization. | |
| Uses Option A: one extra API call per dataset, minimal code invasiveness. | |
| Returns None silently if anything fails (snapshot is best-effort). | |
| """ | |
| try: | |
| ds = ListDataset(entries, freq=freq) | |
| _, test_template = split(ds, offset=-pred_len) | |
| test_data = list(test_template.generate_instances( | |
| prediction_length=pred_len, windows=1, distance=pred_len, | |
| )) | |
| if not test_data: | |
| return None | |
| input_entry, label_entry = test_data[0] | |
| context_vals = np.asarray(input_entry["target"], dtype=np.float64).flatten().tolist() | |
| actuals = np.asarray(label_entry["target"], dtype=np.float64).flatten().tolist() | |
| input_ds = ListDataset([input_entry], freq=freq) | |
| forecasts = list(predictor.predict(input_ds)) | |
| if not forecasts: | |
| return None | |
| fc = forecasts[0] | |
| fallback = fc.mean.tolist() | |
| quantile_levels = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9] | |
| quantiles = { | |
| f"{level:g}": ( | |
| fc.quantile(level).tolist() if hasattr(fc, "quantile") else fallback | |
| ) | |
| for level in quantile_levels | |
| } | |
| p10 = quantiles["0.1"] | |
| p50 = quantiles["0.5"] | |
| p90 = quantiles["0.9"] | |
| n_context = len(context_vals) | |
| n_future = len(actuals) | |
| timestamps = [] | |
| try: | |
| import pandas as pd | |
| start_period = input_entry["start"] | |
| dr = pd.period_range(start=start_period, periods=n_context + n_future, freq=freq) | |
| timestamps = [str(p.to_timestamp()) for p in dr] | |
| except Exception: | |
| pass | |
| return { | |
| "dataset": dataset, | |
| "model": model, | |
| "evaluated_at": evaluated_at, | |
| "freq": freq, | |
| "context_length": n_context, | |
| "prediction_length": n_future, | |
| "context": context_vals, | |
| "actuals": actuals, | |
| "p50": p50, | |
| "p10": p10, | |
| "p90": p90, | |
| "mean": fallback, | |
| "quantiles": quantiles, | |
| "timestamps": timestamps, | |
| } | |
| except Exception: | |
| return None | |
| def run_online_eval_for_dataset( | |
| source: TimeSeriesDataSource, | |
| ds_name: str, | |
| predictor: Predictor, | |
| prediction_length: int | None = None, | |
| ) -> OnlineEvalResult: | |
| meta = source.get_metadata(ds_name) | |
| record = next(iter(source.stream(ds_name))) | |
| if isinstance(source, OpenMeteoDataSource): | |
| pred_len = prediction_length or source._settings.prediction_length | |
| fetched_at = source.fetched_at() | |
| elif isinstance(source, TsBenchDataSource): | |
| pred_len = prediction_length or source.get_prediction_length(ds_name) | |
| fetched_at = source.fetched_at() | |
| else: | |
| pred_len = prediction_length or 24 | |
| fetched_at = pd.Timestamp.utcnow().isoformat() | |
| to_univariate = meta.num_variates > 1 | |
| entries = _records_to_entries(record, to_univariate=to_univariate) | |
| metrics = evaluate_online_window(predictor, entries, pred_len, meta.frequency) | |
| ds_key = ds_name.split("/")[0].lower() | |
| if isinstance(source, TsBenchDataSource): | |
| config_name = source.leaderboard_dataset_name(ds_name) | |
| else: | |
| config_name = f"{ds_key}/{meta.frequency}/short" | |
| model_name = getattr(predictor, "leaderboard_name", predictor.__class__.__name__) | |
| context_length = int(np.asarray(entries[0]["target"]).shape[-1]) - pred_len | |
| evaluated_at = pd.Timestamp.utcnow().isoformat() | |
| forecast_snapshot = _capture_forecast_snapshot( | |
| predictor, | |
| entries, | |
| pred_len, | |
| meta.frequency, | |
| evaluated_at, | |
| config_name, | |
| model_name, | |
| ) | |
| return OnlineEvalResult( | |
| dataset=config_name, | |
| model=model_name, | |
| metrics=metrics, | |
| domain=meta.domain, | |
| num_variates=1 if to_univariate else meta.num_variates, | |
| data_fetched_at=fetched_at, | |
| context_length=context_length, | |
| prediction_length=pred_len, | |
| forecast_snapshot=forecast_snapshot, | |
| ) | |
| def result_to_row(result: OnlineEvalResult) -> list[Any]: | |
| return [ | |
| result.dataset, | |
| result.model, | |
| result.metrics["MSE[mean]"], | |
| result.metrics["MSE[0.5]"], | |
| result.metrics["MAE[0.5]"], | |
| result.metrics["MASE[0.5]"], | |
| result.metrics["MAPE[0.5]"], | |
| result.metrics["sMAPE[0.5]"], | |
| result.metrics["MSIS"], | |
| result.metrics["RMSE[mean]"], | |
| result.metrics["NRMSE[mean]"], | |
| result.metrics["ND[0.5]"], | |
| result.metrics["mean_weighted_sum_quantile_loss"], | |
| result.domain, | |
| result.num_variates, | |
| ] | |
| def results_to_dataframe(results: Iterable[OnlineEvalResult]) -> pd.DataFrame: | |
| rows = [result_to_row(result) for result in results] | |
| return pd.DataFrame(rows, columns=RESULT_COLUMNS) | |