from __future__ import annotations import csv import json import logging import subprocess import sys import time from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Iterator import numpy as np import pandas as pd from tsfm_bench.data.interface import DatasetMetadata, TimeSeriesRecord logger = logging.getLogger(__name__) def preserve_missing_datasets(normalized_dir: Path, backup_dir: Path) -> list[str]: """Keep the last observations/tasks of sources absent from a partial refresh. Original timestamps remain unchanged, so stale contexts are still rejected by the prequential gate. This does not present cached data as freshly collected. """ def read_rows(path): if not path.exists(): return [], [] with path.open(newline="", encoding="utf-8") as handle: reader = csv.DictReader(handle) return list(reader.fieldnames or []), list(reader) _, old = read_rows(backup_dir / "observations.csv") _, new = read_rows(normalized_dir / "observations.csv") missing = {row["dataset_id"] for row in old} - {row["dataset_id"] for row in new} if not missing: return [] for name in ("observations", "entities", "variables", "forecast_tasks", "quality_reports", "raw_snapshots"): fields, current = read_rows(normalized_dir / f"{name}.csv") old_fields, cached = read_rows(backup_dir / f"{name}.csv") fields = fields or old_fields retained = [row for row in cached if row.get("dataset_id") in missing] if not retained: continue current = [row for row in current if row.get("dataset_id") not in missing] path = normalized_dir / f"{name}.csv" temporary = path.with_suffix(".tmp") with temporary.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows([*current, *retained]) temporary.replace(path) logger.warning("Retained cached datasets absent from refresh: %s", sorted(missing)) return sorted(missing) DEFAULT_SOURCE_DOMAINS: dict[str, str] = { "open_meteo_weather": "Weather", "open_meteo_air_quality": "Air Quality", "nasa_power": "Weather", "usgs_water": "Hydrology", "noaa_coops": "Ocean", "noaa_ndbc": "Ocean", "gbfs_citibike": "Transport", "binance": "Finance", "coingecko": "Finance", "wikimedia_pageviews": "Web", "nws_observations": "Weather", "usgs_earthquake": "Hazards", "noaa_ncei": "Weather", "data_gov_sg_rainfall": "Climate", "data_gov_sg_taxi_availability": "Transportation", "data_gov_sg_sgx_turnover": "Finance", } # Explicit dataset_id → domain mapping derived from all known TS-Bench datasets. # This is the primary lookup and never relies on entities.csv being present. DATASET_ID_DOMAINS: dict[str, str] = { "binance_btcusdt_1h": "Finance", "coingecko_bitcoin_7d": "Finance", "gbfs_citibike_current": "Transport", "nasa_power_shanghai_hourly": "Weather", "noaa_coops_sf_water_level": "Ocean", "noaa_ndbc_46013_realtime": "Ocean", "nws_ksfo_observations": "Weather", "open_meteo_shanghai_air_quality_hourly": "Air Quality", "open_meteo_shanghai_hourly": "Weather", "usgs_earthquake_all_week_hourly": "Hazards", "usgs_potomac_iv": "Hydrology", "wikimedia_time_series_hourly": "Web", "data_gov_sg_singapore_rainfall_5min": "Climate", "data_gov_sg_singapore_taxi_availability": "Transportation", "data_gov_sg_sgx_equities_turnover_monthly": "Finance", } def resolve_domain( dataset_id: str, source_id: str, source_domains: dict[str, str], ) -> str: """Determine the leaderboard domain for a dataset. Resolution order (most-specific first): 1. Exact match in DATASET_ID_DOMAINS. 2. Longest-prefix match in DATASET_ID_DOMAINS. 3. Exact match of source_id in source_domains. 4. Longest-prefix match of dataset_id against source_domains keys. 5. Fallback to "Other". """ if dataset_id in DATASET_ID_DOMAINS: return DATASET_ID_DOMAINS[dataset_id] # Longest prefix match against DATASET_ID_DOMAINS keys. best = max( (k for k in DATASET_ID_DOMAINS if dataset_id.startswith(k)), key=len, default=None, ) if best: return DATASET_ID_DOMAINS[best] if source_id in source_domains: return source_domains[source_id] # Longest prefix match against source_domains keys. best_src = max( (k for k in source_domains if dataset_id.startswith(k)), key=len, default=None, ) if best_src: return source_domains[best_src] return "Other" FREQ_ALIASES = { "1h": "1h", "6min": "6min", "10min": "10min", "15min": "15min", "5min": "5min", "30s": "30s", "1mo": "M", "1d": "1D", "snapshot": "1h", } def _parse_duration(value: str | int | float | None) -> timedelta | None: if value in (None, ""): return None if isinstance(value, (int, float)): return timedelta(seconds=float(value)) text = str(value).strip().lower() units = { "s": "seconds", "sec": "seconds", "secs": "seconds", "second": "seconds", "seconds": "seconds", "m": "minutes", "min": "minutes", "mins": "minutes", "minute": "minutes", "minutes": "minutes", "h": "hours", "hr": "hours", "hrs": "hours", "hour": "hours", "hours": "hours", "d": "days", "day": "days", "days": "days", "w": "weeks", "week": "weeks", "weeks": "weeks", } import re match = re.fullmatch(r"\s*(\d+(?:\.\d+)?)\s*([a-z]+)\s*", text) if not match: raise ValueError(f"Invalid duration: {value!r}") amount = float(match.group(1)) unit = units.get(match.group(2)) if unit is None: raise ValueError(f"Unsupported duration unit in {value!r}") return timedelta(**{unit: amount}) def _duration_to_points(value: str | int | float | None, freq: str) -> int | None: duration = _parse_duration(value) if duration is None: return None offset = pd.tseries.frequencies.to_offset(FREQ_ALIASES.get(freq, freq)) delta = pd.Timedelta(offset.nanos, unit="ns") points = int(duration / delta) return max(1, points) @dataclass(frozen=True) class _TaskWindowPolicy: context_window: str | None = None prediction_window: str | None = None context_length_steps: int | None = None prediction_length_steps: int | None = None min_context_points: int = 32 max_context_points: int | None = None max_prediction_points: int | None = None @classmethod def from_dict(cls, payload: dict[str, Any] | None) -> "_TaskWindowPolicy": payload = payload or {} max_context = payload.get("max_context_points") max_prediction = payload.get("max_prediction_points") context_steps = payload.get("context_length_steps", payload.get("history_length_steps")) prediction_steps = payload.get("prediction_length_steps", payload.get("forecast_horizon_steps")) return cls( context_window=payload.get("context_window"), prediction_window=payload.get("prediction_window"), context_length_steps=int(context_steps) if context_steps is not None else None, prediction_length_steps=int(prediction_steps) if prediction_steps is not None else None, min_context_points=int(payload.get("min_context_points", 32)), max_context_points=int(max_context) if max_context is not None else None, max_prediction_points=int(max_prediction) if max_prediction is not None else None, ) def context_points(self, freq: str, available: int) -> int: desired = self.context_length_steps or _duration_to_points(self.context_window, freq) or available if self.max_context_points is not None: desired = min(desired, self.max_context_points) return max(1, min(int(desired), available)) def prediction_points(self, freq: str, available: int) -> int: desired = self.prediction_length_steps or _duration_to_points(self.prediction_window, freq) or available if self.max_prediction_points is not None: desired = min(desired, self.max_prediction_points) return max(1, min(int(desired), available)) def _resolve_dataset_registry_path(raw_settings: dict[str, Any], payload: dict[str, Any]) -> Path | None: raw_path = raw_settings.get("dataset_registry_path") or payload.get("dataset_registry_path") if not raw_path: return None path = Path(raw_path).expanduser() if path.is_absolute() or path.exists(): return path.resolve() config_dir = payload.get("__config_dir") if config_dir: candidate = Path(config_dir) / path if candidate.exists(): return candidate.resolve() return path.resolve() def _load_dataset_registry_task_windows(path: Path | None) -> dict[str, _TaskWindowPolicy]: if path is None: return {} if not path.exists(): raise FileNotFoundError(f"Dataset registry not found: {path}") registry = json.loads(path.read_text(encoding="utf-8")) windows: dict[str, _TaskWindowPolicy] = {} for entry in registry.get("datasets", []): dataset_id = str(entry.get("dataset_id", "")) if not dataset_id: continue history_steps = entry.get("history_length_steps") horizon_steps = entry.get("forecast_horizon_steps") if history_steps is None and horizon_steps is None: continue windows[dataset_id] = _TaskWindowPolicy.from_dict( { "context_length_steps": history_steps, "prediction_length_steps": horizon_steps, "min_context_points": min(int(history_steps), 32) if history_steps is not None else 32, } ) return windows @dataclass(frozen=True) class _TsBenchSettings: ts_bench_root: Path dataset_registry_path: Path | None = None output_subdir: str = "data/sample_week" collect_days: int = 7 collect_timeout: int = 45 limit_entities: int = 10 task_limit: int = 20 max_tasks_per_dataset: int = 1 min_future_steps: int = 1 skip_frequencies: tuple[str, ...] = ("snapshot",) auto_refresh: bool = True collect_retries: int = 3 collect_retry_delay: int = 15 source_intervals: dict[str, int] = None # type: ignore[assignment] default_task_window: _TaskWindowPolicy = None # type: ignore[assignment] source_task_windows: dict[str, _TaskWindowPolicy] = None # type: ignore[assignment] dataset_task_windows: dict[str, _TaskWindowPolicy] = None # type: ignore[assignment] def __post_init__(self): if self.source_intervals is None: object.__setattr__(self, "source_intervals", {}) if self.default_task_window is None: object.__setattr__(self, "default_task_window", _TaskWindowPolicy()) if self.source_task_windows is None: object.__setattr__(self, "source_task_windows", {}) if self.dataset_task_windows is None: object.__setattr__(self, "dataset_task_windows", {}) @dataclass(frozen=True) class _TaskSpec: task_id: str dataset_id: str source_id: str domain: str target_variable_id: str frequency: str gluonts_freq: str prediction_length: int context_length: int | None task_dir: Path leaderboard_name: str class TsBenchDataSource: """Live TS-Bench data: refresh via collect_weekly_data.py, read exported tasks.""" def __init__( self, settings: _TsBenchSettings, source_domains: dict[str, str], tasks: tuple[_TaskSpec, ...], fetched_at: str = "", ): self._settings = settings self._source_domains = source_domains self._tasks = {task.task_id: task for task in tasks} self._fetched_at = fetched_at @classmethod def from_config(cls, payload: dict[str, Any]) -> TsBenchDataSource: raw = payload.get("settings", {}) ts_bench_root = Path(raw.get("ts_bench_root", "../TS-Bench")).expanduser().resolve() raw_intervals = payload.get("source_intervals", {}) source_intervals = {k: int(v) for k, v in raw_intervals.items()} raw_windows = payload.get("task_windows", {}) default_task_window = _TaskWindowPolicy.from_dict(raw_windows.get("default", {})) source_task_windows = { str(k): _TaskWindowPolicy.from_dict(v) for k, v in raw_windows.get("sources", {}).items() } registry_path = _resolve_dataset_registry_path(raw, payload) registry_task_windows = _load_dataset_registry_task_windows(registry_path) dataset_task_windows = { str(k): _TaskWindowPolicy.from_dict(v) for k, v in raw_windows.get("datasets", {}).items() } dataset_task_windows = {**registry_task_windows, **dataset_task_windows} settings = _TsBenchSettings( ts_bench_root=ts_bench_root, dataset_registry_path=registry_path, output_subdir=str(raw.get("output_subdir", "data/sample_week")), collect_days=int(raw.get("collect_days", 7)), collect_timeout=int(raw.get("collect_timeout", 45)), limit_entities=int(raw.get("limit_entities", 10)), task_limit=int(raw.get("task_limit", 20)), max_tasks_per_dataset=int(raw.get("max_tasks_per_dataset", 1)), min_future_steps=int(raw.get("min_future_steps", 1)), skip_frequencies=tuple(raw.get("skip_frequencies", ["snapshot"])), auto_refresh=bool(raw.get("auto_refresh", True)), collect_retries=int(raw.get("collect_retries", 3)), collect_retry_delay=int(raw.get("collect_retry_delay", 15)), source_intervals=source_intervals, default_task_window=default_task_window, source_task_windows=source_task_windows, dataset_task_windows=dataset_task_windows, ) source_domains = {**DEFAULT_SOURCE_DOMAINS, **payload.get("source_domains", {})} source = cls(settings=settings, source_domains=source_domains, tasks=()) source._load_tasks() return source @property def data_root(self) -> Path: return self._settings.ts_bench_root / self._settings.output_subdir def _task_window_policy(self, dataset_id: str, source_id: str) -> _TaskWindowPolicy: if dataset_id in self._settings.dataset_task_windows: return self._settings.dataset_task_windows[dataset_id] best_dataset = max( (k for k in self._settings.dataset_task_windows if dataset_id.startswith(k)), key=len, default=None, ) if best_dataset: return self._settings.dataset_task_windows[best_dataset] if source_id in self._settings.source_task_windows: return self._settings.source_task_windows[source_id] best_source = max( (k for k in self._settings.source_task_windows if dataset_id.startswith(k)), key=len, default=None, ) if best_source: return self._settings.source_task_windows[best_source] return self._settings.default_task_window def _task_policy_payload(self) -> dict[str, Any]: def to_dict(policy: _TaskWindowPolicy) -> dict[str, Any]: payload: dict[str, Any] = { "min_context_points": policy.min_context_points, } if policy.context_window: payload["context_window"] = policy.context_window if policy.prediction_window: payload["prediction_window"] = policy.prediction_window if policy.context_length_steps is not None: payload["context_length_steps"] = policy.context_length_steps if policy.prediction_length_steps is not None: payload["prediction_length_steps"] = policy.prediction_length_steps if policy.max_context_points is not None: payload["max_context_points"] = policy.max_context_points if policy.max_prediction_points is not None: payload["max_prediction_points"] = policy.max_prediction_points return payload return { "default": to_dict(self._settings.default_task_window), "sources": { key: to_dict(value) for key, value in sorted(self._settings.source_task_windows.items()) }, "datasets": { key: to_dict(value) for key, value in sorted(self._settings.dataset_task_windows.items()) }, } def refresh(self) -> None: import shutil root = self._settings.ts_bench_root output_dir = self.data_root collect_script = root / "collect_weekly_data.py" export_script = root / "export_model_inputs.py" if not collect_script.exists(): raise FileNotFoundError(f"TS-Bench collect script not found: {collect_script}") # Create backups of existing data if they exist to prevent partial or empty runs from breaking the cache normalized_dir = self.data_root / "normalized" model_inputs_dir = self.data_root / "model_inputs_latest" backup_normalized = self.data_root / "normalized.bak" backup_model_inputs = self.data_root / "model_inputs_latest.bak" if normalized_dir.exists(): shutil.rmtree(backup_normalized, ignore_errors=True) shutil.copytree(normalized_dir, backup_normalized) if model_inputs_dir.exists(): shutil.rmtree(backup_model_inputs, ignore_errors=True) shutil.copytree(model_inputs_dir, backup_model_inputs) collect_cmd = [ sys.executable, str(collect_script), "--days", str(self._settings.collect_days), "--output", self._settings.output_subdir, "--timeout", str(self._settings.collect_timeout), "--limit-entities", str(self._settings.limit_entities), "--max-tasks-per-dataset", str(self._settings.max_tasks_per_dataset), "--task-window-policy", json.dumps(self._task_policy_payload()), ] retries = max(1, self._settings.collect_retries) delay = self._settings.collect_retry_delay collect_ok = False for attempt in range(1, retries + 1): logger.info( "Collecting TS-Bench data → %s (attempt %d/%d)", output_dir, attempt, retries, ) result = subprocess.run(collect_cmd, cwd=root, check=False) if result.returncode == 0: collect_ok = True break if attempt < retries: wait = delay * attempt logger.warning( "collect_weekly_data.py failed (exit %d); retrying in %ds", result.returncode, wait, ) time.sleep(wait) else: logger.warning( "collect_weekly_data.py failed after %d attempts (exit %d); " "will use existing cached data if available", retries, result.returncode, ) export_ok = False if collect_ok: preserve_missing_datasets(normalized_dir, backup_normalized) normalized_rel = self._settings.output_subdir + "/normalized" model_inputs_rel = self._settings.output_subdir + "/model_inputs_latest" if export_script.exists(): logger.info("Exporting TS-Bench model inputs → %s", self.data_root / "model_inputs_latest") export_cmd = [ sys.executable, str(export_script), "--input", normalized_rel, "--output", model_inputs_rel, "--limit", str(self._settings.task_limit), ] export_result = subprocess.run(export_cmd, cwd=root, check=False) if export_result.returncode == 0: export_ok = True else: logger.warning( "export_model_inputs.py failed (exit %d); using last exported manifest", export_result.returncode, ) # Verify that we ended up with a valid non-empty task manifest has_tasks = False manifest_path = model_inputs_dir / "manifest.json" if not manifest_path.exists(): manifest_path = model_inputs_dir / "manifest.csv" if manifest_path.exists(): try: if manifest_path.suffix == ".json": entries = json.loads(manifest_path.read_text()) else: with manifest_path.open(newline="", encoding="utf-8") as handle: entries = list(csv.DictReader(handle)) if len(entries) > 0: has_tasks = True except Exception as e: logger.warning("Failed to parse manifest to verify tasks: %s", e) if not collect_ok or not export_ok or not has_tasks: logger.warning( "Realtime collection or export failed/returned 0 tasks (collect_ok=%s, export_ok=%s, has_tasks=%s). " "Restoring previous cached data from backup to prevent empty leaderboard.", collect_ok, export_ok, has_tasks ) if backup_normalized.exists(): shutil.rmtree(normalized_dir, ignore_errors=True) shutil.copytree(backup_normalized, normalized_dir) if backup_model_inputs.exists(): shutil.rmtree(model_inputs_dir, ignore_errors=True) shutil.copytree(backup_model_inputs, model_inputs_dir) # Clean up backups shutil.rmtree(backup_normalized, ignore_errors=True) shutil.rmtree(backup_model_inputs, ignore_errors=True) self._load_tasks() def _load_tasks(self) -> None: summary_path = self.data_root / "summary.json" if summary_path.exists(): summary = json.loads(summary_path.read_text()) self._fetched_at = summary.get("generated_at", "") else: self._fetched_at = datetime.now(timezone.utc).isoformat() source_by_dataset = self._load_dataset_sources() tasks = self._load_task_specs(source_by_dataset) self._tasks = {task.task_id: task for task in tasks} logger.info("Loaded %d TS-Bench tasks from %s", len(tasks), self.data_root) def _load_dataset_sources(self) -> dict[str, str]: entities_path = self.data_root / "normalized" / "entities.csv" mapping: dict[str, str] = {} if not entities_path.exists(): return mapping with entities_path.open(newline="", encoding="utf-8") as handle: for row in csv.DictReader(handle): mapping.setdefault(row["dataset_id"], row["source_id"]) return mapping def _load_task_specs(self, source_by_dataset: dict[str, str]) -> tuple[_TaskSpec, ...]: manifest_path = self.data_root / "model_inputs_latest" / "manifest.json" if not manifest_path.exists(): manifest_path = self.data_root / "model_inputs_latest" / "manifest.csv" if not manifest_path.exists(): raise FileNotFoundError( f"No TS-Bench manifest under {self.data_root / 'model_inputs_latest'}. " "Run collect_weekly_data.py and export_model_inputs.py first." ) entries: list[dict[str, Any]] if manifest_path.suffix == ".json": entries = json.loads(manifest_path.read_text()) else: with manifest_path.open(newline="", encoding="utf-8") as handle: entries = list(csv.DictReader(handle)) specs: list[_TaskSpec] = [] for entry in entries: future_rows = int(entry.get("future_target_rows", 0)) if future_rows < self._settings.min_future_steps: continue task_id = entry["task_id"] task_dir = self.data_root / "model_inputs_latest" / task_id task_json = task_dir / "task.json" if not task_json.exists(): continue task = json.loads(task_json.read_text()) raw_freq = str(task.get("frequency", "1h")) if raw_freq in self._settings.skip_frequencies: continue dataset_id = task["dataset_id"] source_id = source_by_dataset.get(dataset_id, dataset_id.split("_")[0]) domain = resolve_domain(dataset_id, source_id, self._source_domains) gluonts_freq = FREQ_ALIASES.get(raw_freq, raw_freq) policy = self._task_window_policy(dataset_id, source_id) pred_len = policy.prediction_points(raw_freq, future_rows) raw_context_rows = int(entry.get("context_rows", task.get("input_length", 0)) or 0) context_len = ( policy.context_points(raw_freq, raw_context_rows) if raw_context_rows > 0 else None ) variable = task["target_variable_id"] leaderboard_name = f"{dataset_id}_{variable}" specs.append( _TaskSpec( task_id=task_id, dataset_id=dataset_id, source_id=source_id, domain=domain, target_variable_id=variable, frequency=raw_freq, gluonts_freq=gluonts_freq, prediction_length=pred_len, context_length=context_len, task_dir=task_dir, leaderboard_name=leaderboard_name, ) ) return tuple(specs) def list_datasets(self) -> list[str]: return sorted(self._tasks.keys()) def get_metadata(self, name: str) -> DatasetMetadata: task = self._require_task(name) return DatasetMetadata( name=name, domain=task.domain, frequency=task.gluonts_freq, num_variates=1, terms=("short",), source_type="ts_bench", ) def get_prediction_length(self, name: str) -> int: return self._require_task(name).prediction_length def stream(self, name: str) -> Iterator[TimeSeriesRecord]: task = self._require_task(name) context_path = task.task_dir / "context.csv" future_path = task.task_dir / "future_target.csv" if not context_path.exists() or not future_path.exists(): raise FileNotFoundError(f"Missing task CSVs for {name}") context = pd.read_csv(context_path) future = pd.read_csv(future_path) target_col = task.target_variable_id if target_col not in context.columns: raise RuntimeError(f"Target column {target_col} missing in {context_path}") context_times = list(context["timestamp"]) future_times = list(future["timestamp"]) context_vals = pd.to_numeric(context[target_col], errors="coerce").to_numpy(dtype=np.float32) future_vals = pd.to_numeric(future[target_col], errors="coerce").to_numpy(dtype=np.float32) if task.context_length is not None and len(context_vals) > task.context_length: context_vals = context_vals[-task.context_length:] context_times = context_times[-task.context_length:] if len(future_vals) > task.prediction_length: future_vals = future_vals[:task.prediction_length] future_times = future_times[:task.prediction_length] values = np.concatenate([context_vals, future_vals]) mask = ~np.isnan(values) values = values[mask] if len(values) < task.prediction_length + 2: raise RuntimeError( f"Insufficient data for {name}: {len(values)} points, " f"need > {task.prediction_length + 2}" ) timestamps = context_times + future_times timestamps = [timestamps[i] for i, keep in enumerate(mask) if keep] start_ts = pd.Timestamp(timestamps[0]) period_freq = ( "M" if task.gluonts_freq == "M" else pd.tseries.frequencies.to_offset(task.gluonts_freq) ) yield TimeSeriesRecord( item_id=task.leaderboard_name.replace("/", "_"), target=values, start=str(pd.Period(start_ts, freq=period_freq)), freq=task.gluonts_freq, ) def leaderboard_dataset_name(self, name: str) -> str: task = self._require_task(name) display_frequency = "1mo" if task.gluonts_freq == "M" else task.gluonts_freq return f"{task.leaderboard_name}/{display_frequency}/short" def fetched_at(self) -> str: return self._fetched_at or datetime.now(timezone.utc).isoformat() def get_eval_interval(self, name: str, default: int = 300) -> int: """Return the eval interval in seconds for a dataset task. Looks up source_intervals by source_id with longest-prefix fallback. Returns *default* when no match is configured. """ task = self._require_task(name) intervals = self._settings.source_intervals if not intervals: return default # Exact match on source_id if task.source_id in intervals: return intervals[task.source_id] # Longest-prefix match on dataset_id against interval keys best = max( (k for k in intervals if task.dataset_id.startswith(k)), key=len, default=None, ) if best: return intervals[best] return default def _require_task(self, name: str) -> _TaskSpec: if name not in self._tasks: raise KeyError(f"Unknown TS-Bench task: {name}") return self._tasks[name]