#!/usr/bin/env python3 """Run online zero-shot evaluation on live TS-Bench data via TSFM.ai API.""" from __future__ import annotations import argparse import csv import json import logging import math import os import re import sys from datetime import datetime, timezone from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[1] SRC_DIR = REPO_ROOT / "src" if str(SRC_DIR) not in sys.path: sys.path.insert(0, str(SRC_DIR)) import pandas as pd import yaml from dotenv import load_dotenv from tsfm_bench.data.registry import load_data_source, load_dataset_properties from tsfm_bench.data.ts_bench import TsBenchDataSource from tsfm_bench.eval.api_predictor import DEFAULT_QUANTILES, TsfmApiConfig, TsfmApiPredictor from tsfm_bench.eval.external_api_predictor import ExternalApiConfig, ExternalApiPredictor from tsfm_bench.eval.gift_eval_aggregation import write_aggregated_results from tsfm_bench.eval.live_aggregation import write_live_aggregates from tsfm_bench.eval.metrics import RESULT_COLUMNS from tsfm_bench.eval.predictors import ( ARIMAPredictor, ETSPredictor, MovingAveragePredictor, SeasonalNaivePredictor, ) from tsfm_bench.eval.online_eval import run_online_eval_for_dataset from tsfm_bench.eval.prequential import run_prequential_cycle from tsfm_bench.eval.canonical_metrics import CANONICAL_METRIC_VERSION logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger(__name__) RESULT_OUTPUT_COLUMNS = [*RESULT_COLUMNS, "prediction_length"] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--data-config", type=Path, default=Path("configs/datasets/ts_bench.yaml"), help="Dataset registry (TS-Bench, Open-Meteo, etc.)", ) parser.add_argument( "--model-config", type=Path, default=Path("configs/models/online_tsfm.yaml"), help="TSFM API model registry", ) parser.add_argument( "--output-root", type=Path, default=Path("space/results"), help="Leaderboard results root (one subdir per model)", ) parser.add_argument( "--baseline-model", default=None, help="Override baseline display name for metric normalization", ) parser.add_argument( "--dry-run", action="store_true", help="Use Seasonal Naive instead of TSFM.ai API (pipeline smoke test)", ) parser.add_argument( "--skip-existing", action="store_true", help="Skip models that already have all_results.csv in output-root", ) parser.add_argument( "--refresh-data", action="store_true", help="Force refresh TS-Bench data before evaluation", ) parser.add_argument( "--no-refresh-data", action="store_true", help="Skip TS-Bench collection even when auto_refresh is enabled in config", ) parser.add_argument( "--datasets", nargs="*", default=None, help="Evaluate only these dataset task IDs (used by daemon for per-interval scheduling)", ) parser.add_argument( "--models", nargs="*", default=None, help="Evaluate only these models, matched by display name, model_id, or output slug", ) parser.add_argument( "--evaluation-mode", choices=("prequential", "legacy"), default=os.getenv("TSFM_EVALUATION_MODE", "prequential"), help=( "prequential issues forecasts before targets exist and scores them in a " "later cycle; legacy performs the former retrospective rolling holdout" ), ) return parser.parse_args() LEGACY_MODEL_NAMES = { "TSFM1": "Chronos-Bolt-Tiny", "TSFM2": "Chronos-Bolt-Base", "TSFM3": "Chronos-2", } def model_display_name(model_spec: dict[str, Any]) -> str: return model_spec.get("display_name") or model_spec["model_id"] def model_output_slug(model_spec: dict[str, Any]) -> str: name = model_spec.get("display_name") or model_spec["model_id"] return re.sub(r"[^a-zA-Z0-9]+", "_", name).strip("_").lower() def resolve_baseline_name(model_specs: list[dict[str, Any]], override: str | None) -> str: if override: return LEGACY_MODEL_NAMES.get(override, override) for spec in model_specs: if spec.get("baseline"): return model_display_name(spec) if len(model_specs) > 1: return model_display_name(model_specs[1]) return model_display_name(model_specs[0]) def load_model_specs(path: Path) -> list[dict[str, Any]]: payload = yaml.safe_load(path.read_text()) return payload.get("models", []) def model_spec_enabled(model_spec: dict[str, Any]) -> bool: return parse_bool(model_spec.get("enabled", True)) def parse_bool(raw: Any) -> bool: if isinstance(raw, str): return raw.strip().lower() not in {"0", "false", "no", "off", "disabled"} return bool(raw) def skipped_model_payload(model_spec: dict[str, Any]) -> dict[str, str]: reason = str(model_spec.get("skip_reason", "disabled for automatic evaluation")).strip() return {"model": model_display_name(model_spec), "reason": reason} def filter_model_specs(model_specs: list[dict[str, Any]], selected: list[str] | None) -> list[dict[str, Any]]: if not selected: return [spec for spec in model_specs if model_spec_enabled(spec)] selected_keys = {re.sub(r"[^a-zA-Z0-9]+", "_", item).strip("_").lower() for item in selected} filtered = [] for spec in model_specs: candidates = { re.sub(r"[^a-zA-Z0-9]+", "_", str(spec.get("model_id", ""))).strip("_").lower(), re.sub(r"[^a-zA-Z0-9]+", "_", model_display_name(spec)).strip("_").lower(), model_output_slug(spec), } if candidates & selected_keys: filtered.append(spec) if not filtered: raise ValueError(f"No models matched --models: {', '.join(selected)}") return filtered def existing_result_rows(csv_path: Path) -> int: if not csv_path.exists(): return 0 with csv_path.open(newline="") as handle: return max(0, sum(1 for _ in handle) - 1) def read_existing_rows(csv_path: Path) -> dict[str, list[Any]]: """Read existing CSV rows keyed by dataset name (first column).""" if not csv_path.exists(): return {} rows: dict[str, list[Any]] = {} with csv_path.open(newline="") as handle: reader = csv.reader(handle) next(reader, None) # skip header for row in reader: if row: rows[row[0]] = row return rows def write_model_results( model_spec: dict[str, Any], rows: list[list[Any]], output_root: Path, meta: dict[str, Any], ) -> bool: model_name = model_display_name(model_spec) out_dir = output_root / model_output_slug(model_spec) out_dir.mkdir(parents=True, exist_ok=True) csv_path = out_dir / "all_results.csv" # Upsert: merge new rows into existing results keyed by dataset name. existing = read_existing_rows(csv_path) new_by_dataset = {row[0]: row for row in rows} merged = {**existing, **new_by_dataset} # new rows take priority if not merged: logger.warning("Skipping write for %s: no rows at all", model_name) return False if not new_by_dataset and existing: logger.warning( "Skipping write for %s: no new rows (keeping %d existing)", model_name, len(existing), ) return False with csv_path.open("w", newline="") as handle: writer = csv.writer(handle) writer.writerow(RESULT_OUTPUT_COLUMNS) writer.writerows(sorted(merged.values(), key=lambda r: r[0])) carried = len(existing) - len(new_by_dataset & existing.keys()) logger.info( "Wrote %s (%d new rows, %d carried from cache)", csv_path, len(new_by_dataset), max(0, carried), ) config = { "model": model_name, "model_type": model_spec.get("model_type", "zero-shot"), "model_dtype": "float32", "model_link": model_spec.get( "model_link", f"https://tsfm.ai/models/{model_spec['model_id']}" ), "code_link": model_spec.get( "code_link", "https://github.com/Thinkcat-Lab/LiveHouse-TS/blob/main/scripts/run_online_eval.py", ), "org": model_spec.get("org", "TSFM.ai"), "testdata_leakage": "No", "replication_code_available": "Yes", "api_model_id": model_spec["model_id"], "admitted_at": model_spec.get("admitted_at"), "evaluation_protocol": model_spec.get( "evaluation_protocol", "issue-now-score-later" ), "metric_version": model_spec.get( "metric_version", CANONICAL_METRIC_VERSION ), } (out_dir / "config.json").write_text(json.dumps(config, indent=4) + "\n") (out_dir / "online_meta.json").write_text(json.dumps(meta, indent=4) + "\n") return True def write_dataset_properties(properties: dict[str, dict[str, Any]], output_root: Path) -> None: rows = [ { "dataset": key, "domain": values["domain"], "frequency": values["frequency"], "num_variates": values["num_variates"], } for key, values in sorted(properties.items()) ] csv_path = output_root / "dataset_properties.csv" with csv_path.open("w", newline="") as handle: writer = csv.DictWriter( handle, fieldnames=["dataset", "domain", "frequency", "num_variates"], ) writer.writeheader() writer.writerows(rows) def write_run_metadata(output_root: Path, payload: dict[str, Any]) -> None: (output_root / "online_status.json").write_text(json.dumps(payload, indent=4) + "\n") def metric_release_id(result: Any) -> str: source_time = str(result.data_fetched_at or "").strip() if not source_time: source_time = datetime.now(timezone.utc).isoformat() return f"{result.dataset}|source:{source_time}|h:{result.prediction_length}" def append_metric_release(result: Any, output_root: Path) -> None: """Append one metrics-only evaluation row. This intentionally stores no context, forecast samples, or ground truth. """ parts = str(result.dataset).rsplit("/", 2) frequency = parts[1] if len(parts) == 3 else "unknown" evaluated_at = datetime.now(timezone.utc).isoformat() release_time = str(result.data_fetched_at or evaluated_at) metrics = result.metrics try: mape_eligible = math.isfinite(float(metrics.get("MAPE[0.5]"))) except (TypeError, ValueError): mape_eligible = False payload = { "release_id": metric_release_id(result), "release_time": release_time, "evaluated_at": evaluated_at, "dataset": result.dataset, "domain": result.domain, "frequency": frequency, "prediction_length": result.prediction_length, "model": result.model, "MSE": metrics.get("MSE[mean]"), "RMSE": metrics.get("RMSE[mean]"), "MAPE": metrics.get("MAPE[0.5]"), "CRPS": metrics.get("mean_weighted_sum_quantile_loss"), "MAPE_eligible": mape_eligible, } log_path = output_root / "evaluation_metrics.jsonl" with log_path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(payload) + "\n") def save_forecast_snapshot(snapshot: dict | None, out_dir: Path) -> None: """Write a forecast snapshot JSON to {out_dir}/forecasts/{dataset_key}.json.""" if snapshot is None: return dataset_key = snapshot["dataset"].replace("/", "__").replace(" ", "_") forecasts_dir = out_dir / "forecasts" forecasts_dir.mkdir(parents=True, exist_ok=True) (forecasts_dir / f"{dataset_key}.json").write_text( json.dumps(snapshot, indent=2) + "\n" ) def save_eval_detail(result: Any, output_root: Path) -> None: """Save context (history), prediction (forecast), and actuals (ground truth) locally. Organized by domain, dataset, and model. """ snapshot = result.forecast_snapshot if snapshot is None: logger.warning("No forecast snapshot for %s on %s; skipping detail logging", result.model, result.dataset) return # Slugify directories to be safe and cross-platform domain_slug = re.sub(r"[^a-zA-Z0-9]+", "_", result.domain).strip("_").lower() dataset_slug = re.sub(r"[^a-zA-Z0-9]+", "_", result.dataset).strip("_").lower() model_slug = re.sub(r"[^a-zA-Z0-9]+", "_", result.model).strip("_").lower() # Base details directory: e.g. space/results/saved_evals/ details_dir = output_root / "saved_evals" / domain_slug / dataset_slug / model_slug details_dir.mkdir(parents=True, exist_ok=True) evaluated_at = snapshot.get("evaluated_at") or datetime.now(timezone.utc).isoformat() # Create a clean, sorted safe timestamp string for the file name: YYYYMMDD_HHMMSS # e.g., 2026-06-16T12:00:00.000000Z -> 20260616_120000 safe_ts = re.sub(r"[^0-9a-zA-Z]", "", evaluated_at)[:15].replace("T", "_") if not safe_ts: safe_ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") filepath = details_dir / f"{safe_ts}.json" payload = { "dataset": result.dataset, "domain": result.domain, "model": result.model, "evaluated_at": evaluated_at, "metrics": result.metrics, "context_length": result.context_length, "prediction_length": result.prediction_length, "timestamps": snapshot.get("timestamps", []), "context": snapshot.get("context", []), "actual": snapshot.get("actuals", []), "predictions": { "p50": snapshot.get("p50", []), "p10": snapshot.get("p10", []), "p90": snapshot.get("p90", []), "mean": snapshot.get("mean", snapshot.get("p50", [])), "quantiles": snapshot.get("quantiles", {}), } } try: filepath.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") logger.info("Saved evaluation detail: %s", filepath) except Exception as e: logger.error("Failed to save evaluation detail to %s: %s", filepath, e) BASELINE_RANK_HISTORY_COLUMNS = [ "date", "model", "MSE_Rank", "CRPS_Rank", "MASE_Rank", "MAE_Rank", "RMSE_Rank", "sMAPE_Rank", "MSIS_Rank", "ND_Rank", "NRMSE_Rank", "MAPE_Rank", ] _RANK_METRIC_MAP = { "MSE_Rank": "eval_metrics/MSE[mean]", "CRPS_Rank": "eval_metrics/mean_weighted_sum_quantile_loss", "MASE_Rank": "eval_metrics/MASE[0.5]", "MAE_Rank": "eval_metrics/MAE[0.5]", "RMSE_Rank": "eval_metrics/RMSE[mean]", "sMAPE_Rank": "eval_metrics/sMAPE[0.5]", "MSIS_Rank": "eval_metrics/MSIS", "ND_Rank": "eval_metrics/ND[0.5]", "NRMSE_Rank": "eval_metrics/NRMSE[mean]", "MAPE_Rank": "eval_metrics/MAPE[0.5]", } _RANK_COLS = [c for c in BASELINE_RANK_HISTORY_COLUMNS if c.endswith("_Rank")] def _compute_baseline_ranks(output_root: Path, baseline_name: str) -> dict[str, float] | None: """Re-read all_results CSVs, compute per-dataset ranks, return avg rank dict for baseline.""" frames = [] for subdir in output_root.iterdir(): csv_path = subdir / "all_results.csv" if csv_path.exists(): try: frames.append(pd.read_csv(csv_path)) except Exception: pass if not frames: return None df = pd.concat(frames, ignore_index=True) df = df.dropna(subset=["dataset"]) rank_means: dict[str, float] = {} for rank_col, metric_col in _RANK_METRIC_MAP.items(): if metric_col not in df.columns: continue df[metric_col] = pd.to_numeric(df[metric_col], errors="coerce") df[rank_col] = df.groupby("dataset")[metric_col].rank(method="first", ascending=True) baseline_rows = df[df["model"] == baseline_name][rank_col] if baseline_rows.empty: continue rank_means[rank_col] = float(baseline_rows.mean(skipna=True)) return rank_means if rank_means else None def _upsert_daily_history(daily_path: Path, row: dict) -> None: """Upsert a row keyed by date (same-day entry is overwritten), keep sorted by date.""" existing: dict[str, dict] = {} if daily_path.exists(): try: with daily_path.open(newline="") as fh: for r in csv.DictReader(fh): existing[r["date"]] = r except Exception: pass existing[row["date"]] = row with daily_path.open("w", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=BASELINE_RANK_HISTORY_COLUMNS, extrasaction="ignore") writer.writeheader() for r in sorted(existing.values(), key=lambda x: x["date"]): writer.writerow(r) def _rebuild_weekly_history(daily_path: Path, weekly_path: Path) -> None: """Aggregate daily CSV into weekly CSV (ISO year-week, mean of daily rank values).""" if not daily_path.exists(): return try: df = pd.read_csv(daily_path) except Exception: return df["date"] = pd.to_datetime(df["date"], errors="coerce") df = df.dropna(subset=["date"]) df["week"] = df["date"].dt.strftime("%G-W%V") # ISO year-week, e.g. "2026-W23" for col in _RANK_COLS: if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") group_cols = ["week"] if "model" in df.columns: group_cols.append("model") agg = df.groupby(group_cols, dropna=False)[_RANK_COLS].mean(numeric_only=True).round(3).reset_index() if "model" not in agg.columns: agg.insert(1, "model", "") weekly_cols = ["week", "model"] + _RANK_COLS present = [c for c in weekly_cols if c in agg.columns] agg[present].sort_values([c for c in ["week", "model"] if c in agg.columns]).to_csv(weekly_path, index=False) def append_baseline_rank_history( output_root: Path, baseline_name: str, eval_date: str, ) -> None: """Upsert baseline daily ranks and rebuild weekly aggregation.""" ranks = _compute_baseline_ranks(output_root, baseline_name) if ranks is None: logger.warning("Could not compute baseline ranks for %s; skipping history append", baseline_name) return daily_path = output_root / "baseline_rank_history_daily.csv" weekly_path = output_root / "baseline_rank_history_weekly.csv" row: dict = {"date": eval_date, "model": baseline_name} row.update({k: f"{v:.3f}" for k, v in ranks.items()}) _upsert_daily_history(daily_path, row) _rebuild_weekly_history(daily_path, weekly_path) logger.info( "Updated baseline rank history (MASE_Rank=%.3f) → %s, %s", ranks.get("MASE_Rank", float("nan")), daily_path.name, weekly_path.name, ) def make_predictor(model_spec: dict[str, Any], pred_len: int, dry_run: bool): if dry_run: predictor = SeasonalNaivePredictor( prediction_length=pred_len, season_length=24, quantile_levels=DEFAULT_QUANTILES, ) predictor.leaderboard_name = model_display_name(model_spec) return predictor model_type = model_spec.get("model_type", "zero-shot") if model_type == "statistical": algorithm = str(model_spec.get("algorithm", model_spec["model_id"])).lower() common_kwargs = { "prediction_length": pred_len, "quantile_levels": DEFAULT_QUANTILES, "num_samples": int(model_spec.get("num_samples", 200)), "random_seed": int(model_spec.get("random_seed", 0)), } if "arima" in algorithm: order = tuple(model_spec.get("order", [1, 1, 1])) predictor = ARIMAPredictor(order=order, **common_kwargs) elif algorithm in {"ets", "exponential_smoothing", "exponential-smoothing"}: predictor = ETSPredictor(**common_kwargs) elif algorithm in {"moving_average", "moving-average", "ma"}: predictor = MovingAveragePredictor( window=int(model_spec.get("window", 24)), **common_kwargs, ) elif algorithm in {"seasonal_naive", "seasonal-naive", "naive"}: predictor = SeasonalNaivePredictor( prediction_length=pred_len, season_length=int(model_spec.get("season_length", 24)), quantile_levels=DEFAULT_QUANTILES, ) else: raise ValueError(f"Unknown statistical baseline algorithm: {algorithm}") predictor.leaderboard_name = model_display_name(model_spec) return predictor if model_type == "local": import importlib import sys # Dynamic import of user's local model class_path = model_spec["model_class"] module_name, class_name = class_path.rsplit(".", 1) # Add space to sys.path to resolve user_models space_path = str(Path(__file__).resolve().parents[1] / "space") if space_path not in sys.path: sys.path.insert(0, space_path) module = importlib.import_module(module_name) model_class = getattr(module, class_name) checkpoint_path = model_spec.get("model_path") model_kwargs = model_spec.get("model_kwargs", {}) predictor = model_class( prediction_length=pred_len, checkpoint_path=checkpoint_path, quantile_levels=DEFAULT_QUANTILES, **model_kwargs, ) predictor.leaderboard_name = model_display_name(model_spec) return predictor if model_type == "external_api": endpoint_url = model_spec.get("endpoint_url") or model_spec.get("api_url") if not endpoint_url: raise ValueError(f"external_api model {model_display_name(model_spec)} is missing endpoint_url") external_config = ExternalApiConfig( endpoint_url=str(endpoint_url), model_id=str(model_spec.get("model_id", model_display_name(model_spec))), auth_token_env=model_spec.get("auth_token_env"), auth_header=str(model_spec.get("auth_header", "Authorization")), timeout=float(model_spec.get("timeout", 90.0)), max_retries=int(model_spec.get("max_retries", 2)), max_context_points=int(model_spec.get("max_context_points", 4096)), max_response_bytes=int(model_spec.get("max_response_bytes", 5 * 1024 * 1024)), require_https=parse_bool(model_spec.get("require_https", True)), send_item_metadata=parse_bool(model_spec.get("send_item_metadata", False)), ) predictor = ExternalApiPredictor( config=external_config, prediction_length=pred_len, quantile_levels=DEFAULT_QUANTILES, ) predictor.leaderboard_name = model_display_name(model_spec) return predictor api_config = TsfmApiConfig(model_id=model_spec["model_id"], min_prediction_length=int(model_spec.get("min_api_prediction_length", 1))) predictor = TsfmApiPredictor( config=api_config, prediction_length=pred_len, quantile_levels=DEFAULT_QUANTILES, ) predictor.leaderboard_name = model_display_name(model_spec) return predictor def resolve_prediction_length(source, ds_name: str) -> int: settings = getattr(source, "_settings", None) if isinstance(source, TsBenchDataSource): return source.get_prediction_length(ds_name) if settings is not None and hasattr(settings, "prediction_length"): return int(settings.prediction_length) return 24 def resolve_data_source_name(config_path: Path) -> str: payload = yaml.safe_load(config_path.read_text()) return str(payload.get("source_type", "unknown")) def run_prequential_mode( *, args: argparse.Namespace, source: Any, dataset_names: list[str], model_specs: list[dict[str, Any]], skipped_models: list[dict[str, str]], data_source_type: str, run_started: str, ) -> None: cycle = run_prequential_cycle( source=source, dataset_names=dataset_names, model_specs=model_specs, output_root=args.output_root, predictor_factory=lambda spec, horizon: make_predictor( spec, horizon, args.dry_run ), model_name=model_display_name, model_slug=model_output_slug, ) model_meta: dict[str, Any] = {} for spec in model_specs: name = model_display_name(spec) rows = cycle.resolved_rows.get(name, []) meta = { "model": name, "api_model_id": spec["model_id"], "admitted_at": spec.get("admitted_at"), "evaluation_protocol": "issue-now-score-later", "metric_version": CANONICAL_METRIC_VERSION, "evaluated_at": datetime.now(timezone.utc).isoformat(), "resolved_tasks": cycle.resolved_meta.get(name, []), } if rows: write_model_results(spec, rows, args.output_root, meta) model_meta[name] = meta aggregate_status = "ok" aggregate_files: dict[str, str] = {} try: gift_paths = write_aggregated_results(args.output_root) live_paths = write_live_aggregates(args.output_root) paths = { **{f"gift_{name}": path for name, path in gift_paths.items()}, **{f"live_{name}": path for name, path in live_paths.items()}, } aggregate_files = { name: str(path.relative_to(args.output_root)) for name, path in paths.items() } except Exception: aggregate_status = "failed" logger.exception("Failed to rebuild aggregate tables after prequential cycle") resolved_count = sum(len(rows) for rows in cycle.resolved_rows.values()) status = "partial" if cycle.failed_forecasts else "ok" if aggregate_status != "ok" or ( cycle.failed_forecasts and cycle.issued_forecasts == 0 and resolved_count == 0 and cycle.pending_tasks == 0 ): status = "failed" write_run_metadata( args.output_root, { "status": status, "evaluation_protocol": "issue-now-score-later", "metric_version": CANONICAL_METRIC_VERSION, "started_at": run_started, "finished_at": datetime.now(timezone.utc).isoformat(), "data_source": data_source_type, "data_config": str(args.data_config), "task_count": len(dataset_names), "issued_tasks": cycle.issued_tasks, "issued_forecasts": cycle.issued_forecasts, "resolved_forecasts": sum(len(rows) for rows in cycle.resolved_rows.values()), "pending_tasks": cycle.pending_tasks, "failed_forecasts": cycle.failed_forecasts, "models": model_meta, "skipped_models": skipped_models, "aggregate_status": aggregate_status, "aggregate_files": aggregate_files, "push_status": "pending", }, ) logger.info( "Prequential cycle complete: issued_tasks=%d issued_forecasts=%d " "resolved_forecasts=%d pending_tasks=%d", cycle.issued_tasks, cycle.issued_forecasts, sum(len(rows) for rows in cycle.resolved_rows.values()), cycle.pending_tasks, ) def main() -> None: load_dotenv() args = parse_args() if args.dry_run and args.output_root.resolve() == (REPO_ROOT / "space/results").resolve(): raise SystemExit("--dry-run requires a separate --output-root (for example outputs/smoke).") source = load_data_source(args.data_config) if isinstance(source, TsBenchDataSource): if args.refresh_data or ( source._settings.auto_refresh and not args.no_refresh_data ): source.refresh() properties = load_dataset_properties(args.data_config) all_model_specs = load_model_specs(args.model_config) skipped_models = [ skipped_model_payload(spec) for spec in all_model_specs if not model_spec_enabled(spec) and not args.models ] for skipped in skipped_models: logger.info("Skipping disabled model %s: %s", skipped["model"], skipped["reason"]) model_specs = filter_model_specs(all_model_specs, args.models) data_source_type = resolve_data_source_name(args.data_config) args.output_root.mkdir(parents=True, exist_ok=True) write_dataset_properties(properties, args.output_root) run_started = datetime.now(timezone.utc).isoformat() all_model_meta: dict[str, Any] = {} evaluated_datasets_in_run: dict[str, int] = {} dataset_names = list(source.list_datasets()) if args.datasets: def clean_name(n: str) -> str: import re return re.sub(r'_\d{8}t\d{6}z$', '', n, flags=re.IGNORECASE) allowed = {clean_name(d) for d in args.datasets} dataset_names = [d for d in dataset_names if clean_name(d) in allowed or d in allowed] logger.info("Filtered to %d datasets: %s", len(dataset_names), dataset_names) task_count = len(dataset_names) if args.evaluation_mode == "prequential": run_prequential_mode( args=args, source=source, dataset_names=dataset_names, model_specs=model_specs, skipped_models=skipped_models, data_source_type=data_source_type, run_started=run_started, ) return failed_models: list[str] = [] if task_count == 0: logger.warning("No TS-Bench tasks available; skipping model evaluation") for model_spec in model_specs: model_name = model_display_name(model_spec) out_dir = args.output_root / model_output_slug(model_spec) if args.skip_existing and (out_dir / "all_results.csv").exists(): logger.info("Skipping %s (results exist at %s)", model_name, out_dir) continue rows: list[list[Any]] = [] dataset_meta: list[dict[str, Any]] = [] failed_datasets: list[str] = [] try: if task_count == 0: out_csv = out_dir / "all_results.csv" if existing_result_rows(out_csv) > 0: logger.info("Keeping existing results for %s", model_name) continue for ds_name in dataset_names: pred_len = resolve_prediction_length(source, ds_name) predictor = make_predictor(model_spec, pred_len, args.dry_run) logger.info("Evaluating %s on %s (live TS-Bench)", model_name, ds_name) try: result = run_online_eval_for_dataset(source, ds_name, predictor) except Exception: logger.exception(" Failed on dataset %s; skipping", ds_name) failed_datasets.append(ds_name) continue base_name = result.dataset.split('/')[0] evaluated_datasets_in_run[base_name] = result.prediction_length rows.append( [ result.dataset, model_name, 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, result.prediction_length, ] ) append_metric_release(result, args.output_root) save_forecast_snapshot(result.forecast_snapshot, out_dir) save_eval_detail(result, args.output_root) dataset_meta.append( { "dataset": ds_name, "data_fetched_at": result.data_fetched_at, "context_length": result.context_length, "prediction_length": result.prediction_length, } ) logger.info( " MASE=%.3f CRPS=%.3f", result.metrics["MASE[0.5]"], result.metrics["mean_weighted_sum_quantile_loss"], ) model_meta = { "model": model_name, "api_model_id": model_spec["model_id"], "evaluated_at": datetime.now(timezone.utc).isoformat(), "datasets": dataset_meta, "failed_datasets": failed_datasets, } if write_model_results(model_spec, rows, args.output_root, model_meta): all_model_meta[model_name] = model_meta elif rows: all_model_meta[model_name] = model_meta except Exception: logger.exception("Failed evaluating %s", model_name) failed_models.append(model_name) aggregate_status = "skipped" aggregate_files: dict[str, str] = {} if task_count > 0: try: gift_paths = write_aggregated_results(args.output_root) live_paths = write_live_aggregates(args.output_root) aggregate_paths = { **{f"gift_{name}": path for name, path in gift_paths.items()}, **{f"live_{name}": path for name, path in live_paths.items()}, } aggregate_files = { name: str(path.relative_to(args.output_root)) for name, path in aggregate_paths.items() } aggregate_status = "ok" logger.info( "Updated GIFT-style aggregates: %s", ", ".join(sorted(aggregate_files.values())), ) except Exception: aggregate_status = "failed" logger.exception("Failed to update GIFT-style aggregate tables") if task_count == 0: status = "no_tasks" elif failed_models: status = "partial" else: status = "ok" all_failed_datasets: list[str] = sorted( {ds for m in all_model_meta.values() for ds in m.get("failed_datasets", [])} ) write_run_metadata( args.output_root, { "status": status, "started_at": run_started, "finished_at": datetime.now(timezone.utc).isoformat(), "task_count": task_count, "data_source": data_source_type, "data_config": str(args.data_config), "ts_bench_root": str(getattr(getattr(source, "_settings", None), "ts_bench_root", "")), "models": all_model_meta, "failed_models": failed_models, "skipped_models": skipped_models, "failed_datasets": all_failed_datasets, "aggregate_status": aggregate_status, "aggregate_files": aggregate_files, "push_status": "pending", }, ) # Append to eval_history.jsonl if evaluated_datasets_in_run: history_path = args.output_root / "eval_history.jsonl" try: with open(history_path, "a") as f: now_str = datetime.now(timezone.utc).isoformat() for base_name, pred_len in sorted(evaluated_datasets_in_run.items()): entry = { "evaluated_at": now_str, "dataset": base_name, "prediction_length": pred_len } f.write(json.dumps(entry) + "\n") logger.info("Wrote %d datasets to eval_history.jsonl", len(evaluated_datasets_in_run)) except Exception as e: logger.exception("Failed to write to eval_history.jsonl: %s", e) baseline_name = resolve_baseline_name(model_specs, args.baseline_model) eval_date = datetime.now(timezone.utc).strftime("%Y-%m-%d") if status != "no_tasks": append_baseline_rank_history(args.output_root, baseline_name, eval_date) logger.info("Online evaluation complete → %s", args.output_root) if __name__ == "__main__": main()