scripts/run_online_eval.py

#!/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()