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 @dataclass 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)