from __future__ import annotations import hashlib import json import re from datetime import datetime, timezone from itertools import combinations from pathlib import Path from typing import Final, Iterable import numpy as np import pandas as pd from scipy import stats from tsfm_bench.eval.canonical_metrics import CANONICAL_METRIC_VERSION COMMON_QUANTILES: Final = (0.1, 0.5, 0.9) DEFAULT_SEASON_LENGTH: Final = 24 DEFAULT_BASELINE_MODEL: Final = "Seasonal-Naive" MODEL_NAME_ALIASES: Final = { "Sundial (Base 128M)": "Sundial", } SNAPSHOT_METRIC_COLUMNS: Final = { "MSE": "eval_metrics/MSE[mean]", "RMSE": "eval_metrics/RMSE[mean]", "MAPE": "eval_metrics/MAPE[0.5]", "CRPS": "eval_metrics/mean_weighted_sum_quantile_loss", } MAPE_RELATIVE_EPSILON: Final = 1e-3 METRIC_COLUMNS: Final = ["MSE", "RMSE", "MAPE", "CRPS", "RTG"] RELEASE_COLUMNS: Final = [ "release_id", "release_time", "evaluated_at", "issued_at", "protocol", "dataset", "domain", "frequency", "prediction_length", "model", "metric_version", *METRIC_COLUMNS, "MAPE_eligible", ] DATASET_COLUMNS: Final = [ "model", "dataset", "domain", "frequency", "prediction_lengths", *METRIC_COLUMNS, "Stability", "Improvement", "Improvement_pvalue", "n_releases", "MAPE_coverage", "first_release", "last_release", ] RANK_COLUMNS: Final = [ "Rank", "Status", "model", "WinRate", "eligible_opponents", "shared_releases", "shared_datasets", "covered_span_days", ] MIN_PAIR_SHARED_RELEASES: Final = 30 MIN_PAIR_SHARED_DATASETS: Final = 5 MIN_PAIR_SPAN_DAYS: Final = 7.0 MIN_OFFICIAL_OPPONENTS: Final = 3 def _dataset_key(value: object) -> str: return str(value).split("/")[0].strip().lower() def _canonical_model_name(value: object) -> str: name = str(value).strip() return MODEL_NAME_ALIASES.get(name, name) def _model_key(value: object) -> str: return re.sub(r"[^a-z0-9]+", "", _canonical_model_name(value).lower()) def _load_model_allowlist(root: Path) -> set[str]: path = root / "model_allowlist.json" if not path.exists(): return set() try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return set() models = payload.get("models", payload) if isinstance(payload, dict) else payload if not isinstance(models, list): return set() return {_model_key(model) for model in models if str(model).strip()} def _load_live_dataset_keys(root: Path) -> set[str]: path = root / "dataset_properties.csv" if not path.exists(): return set() try: props = pd.read_csv(path) except (OSError, pd.errors.ParserError, pd.errors.EmptyDataError): return set() if "dataset" not in props.columns: return set() return {_dataset_key(value) for value in props["dataset"].dropna()} def _numeric_array(value: object) -> np.ndarray: try: array = np.asarray(value, dtype=np.float64).reshape(-1) except (TypeError, ValueError): return np.array([], dtype=np.float64) return array def _quantile_key(level: float) -> str: return f"{level:g}" def _extract_quantile(payload: dict, level: float) -> np.ndarray: predictions = payload.get("predictions", {}) quantiles = predictions.get("quantiles", {}) for key in (_quantile_key(level), str(level), f"q{level:g}"): if key in quantiles: return _numeric_array(quantiles[key]) percentile_key = f"p{int(round(100 * level))}" if percentile_key in predictions: return _numeric_array(predictions[percentile_key]) return np.array([], dtype=np.float64) def _release_identity( payload: dict, context: np.ndarray, actual: np.ndarray, ) -> tuple[str, str]: dataset = str(payload.get("dataset", "")).strip() evaluated_at = str(payload.get("evaluated_at", "")) fingerprint_values = np.concatenate([context[-min(24, context.size) :], actual]) digest = hashlib.sha256( fingerprint_values.astype(" tuple[float, float]: finite = context[np.isfinite(context)] if finite.size == 0: return 0.0, 1.0 center = float(np.mean(finite)) scale = float(np.std(finite, ddof=0)) if not np.isfinite(scale) or scale < 1e-8: scale = 1.0 return center, scale def _seasonal_naive(context: np.ndarray, horizon: int, season_length: int) -> np.ndarray: finite_context = context[np.isfinite(context)] if finite_context.size == 0: return np.zeros(horizon, dtype=np.float64) period = max(1, min(int(season_length), finite_context.size)) history = finite_context[-period:] repeats = int(np.ceil(horizon / period)) return np.tile(history, repeats)[:horizon] def quantile_crps( actual: np.ndarray, forecasts: Iterable[np.ndarray], levels: Iterable[float] = COMMON_QUANTILES, ) -> float: """Estimate CRPS as twice the mean pinball loss over fixed quantiles.""" losses: list[float] = [] for level, forecast in zip(levels, forecasts): forecast = _numeric_array(forecast) if forecast.size != actual.size: return np.nan valid = np.isfinite(actual) & np.isfinite(forecast) if not valid.any(): return np.nan residual = actual[valid] - forecast[valid] pinball = np.maximum(level * residual, (level - 1.0) * residual) losses.append(float(np.mean(pinball))) return float(2.0 * np.mean(losses)) if losses else np.nan def compute_release_metrics( payload: dict, *, quantile_levels: Iterable[float] = COMMON_QUANTILES, season_length: int = DEFAULT_SEASON_LENGTH, ) -> dict | None: """Compute one causal, z-normalized release score from a saved evaluation.""" actual = _numeric_array(payload.get("actual", [])) context = _numeric_array(payload.get("context", [])) point_forecast = _extract_quantile(payload, 0.5) horizon = int(payload.get("prediction_length", actual.size) or actual.size) if actual.size == 0 or context.size == 0 or point_forecast.size != actual.size: return None if horizon != actual.size: horizon = actual.size center, scale = _z_scale(context) actual_z = (actual - center) / scale point_z = (point_forecast - center) / scale valid = np.isfinite(actual_z) & np.isfinite(point_z) if not valid.any(): return None stored_mse = payload.get("metrics", {}).get("MSE[mean]") try: stored_mse = float(stored_mse) except (TypeError, ValueError): stored_mse = np.nan if np.isfinite(stored_mse) and stored_mse >= 0.0: mse = float(stored_mse / np.square(scale)) else: squared_error = np.square(actual_z[valid] - point_z[valid]) mse = float(np.mean(squared_error)) rmse = float(np.sqrt(mse)) context_magnitude = float(np.nanmedian(np.abs(context))) if context.size else 0.0 mape_threshold = max(1e-8, MAPE_RELATIVE_EPSILON * context_magnitude) mape_eligible = bool(np.all(np.abs(actual[valid]) > mape_threshold)) mape = ( float(np.mean(np.abs((actual[valid] - point_forecast[valid]) / actual[valid]))) if mape_eligible else np.nan ) levels = tuple(float(level) for level in quantile_levels) quantile_forecasts = [_extract_quantile(payload, level) for level in levels] quantile_forecasts_z = [(forecast - center) / scale for forecast in quantile_forecasts] crps = quantile_crps(actual_z, quantile_forecasts_z, levels) baseline = _seasonal_naive(context, horizon, season_length) baseline_z = (baseline - center) / scale baseline_valid = np.isfinite(actual_z) & np.isfinite(baseline_z) baseline_mse = float(np.mean(np.square(actual_z[baseline_valid] - baseline_z[baseline_valid]))) denominator = baseline_mse + mse rtg = 0.0 if denominator <= 1e-12 else float(100.0 * (baseline_mse - mse) / denominator) dataset = str(payload.get("dataset", "")).strip() parts = dataset.rsplit("/", 2) frequency = parts[1] if len(parts) == 3 else "unknown" release_id, release_time = _release_identity(payload, context, actual) return { "release_id": release_id, "release_time": release_time, "evaluated_at": str(payload.get("evaluated_at", "")), "issued_at": payload.get("issued_at"), "protocol": str(payload.get("protocol", "legacy-snapshot")), "dataset": dataset, "domain": str(payload.get("domain", "Unknown")).strip(), "frequency": frequency, "prediction_length": horizon, "model": _canonical_model_name(payload.get("model", "")), "metric_version": str(payload.get("metric_version", "legacy-z-v0")), "MSE": mse, "RMSE": rmse, "MAPE": mape, "CRPS": crps, "RTG": rtg, "MAPE_eligible": mape_eligible, } def _with_baseline_rtg( frame: pd.DataFrame, baseline_model: str = DEFAULT_BASELINE_MODEL, ) -> pd.DataFrame: if frame.empty: return frame frame = frame.copy() baseline_rows = frame[frame["model"].map(_model_key) == _model_key(baseline_model)] if baseline_rows.empty: frame["RTG"] = np.nan return frame baseline_mse = baseline_rows.groupby("release_id")["MSE"].last() frame["__baseline_mse"] = frame["release_id"].map(baseline_mse) denominator = frame["__baseline_mse"] + frame["MSE"] valid = ( np.isfinite(frame["__baseline_mse"]) & np.isfinite(frame["MSE"]) & denominator.gt(1e-12) ) frame["RTG"] = np.where( valid, 100.0 * (frame["__baseline_mse"] - frame["MSE"]) / denominator, np.nan, ) zero_denom = ( np.isfinite(frame["__baseline_mse"]) & np.isfinite(frame["MSE"]) & denominator.le(1e-12) ) frame.loc[zero_denom, "RTG"] = 0.0 return frame.drop(columns="__baseline_mse") def load_metric_release_metrics(results_root: str | Path) -> pd.DataFrame: """Load metrics-only release rows; stores no forecasts, context, or actuals.""" root = Path(results_root) canonical_path = root / "evaluation_metrics_canonical.jsonl" legacy_path = root / "evaluation_metrics.jsonl" log_path = canonical_path if canonical_path.exists() else legacy_path if not log_path.exists(): return pd.DataFrame(columns=RELEASE_COLUMNS) rows: list[dict] = [] try: with log_path.open(encoding="utf-8") as handle: for line in handle: if not line.strip(): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue except OSError: return pd.DataFrame(columns=RELEASE_COLUMNS) if not rows: return pd.DataFrame(columns=RELEASE_COLUMNS) frame = pd.DataFrame(rows) if "model" in frame.columns: frame["model"] = frame["model"].map(_canonical_model_name) if "metric_version" not in frame.columns: frame["metric_version"] = "legacy-unknown" else: frame["metric_version"] = frame["metric_version"].fillna("legacy-unknown") if "issued_at" not in frame.columns: frame["issued_at"] = None if "protocol" not in frame.columns: frame["protocol"] = "legacy-snapshot" else: frame["protocol"] = frame["protocol"].fillna("legacy-snapshot") required = { "release_id", "release_time", "evaluated_at", "dataset", "domain", "frequency", "prediction_length", "model", "MSE", "RMSE", "MAPE", "CRPS", } if not required.issubset(frame.columns): return pd.DataFrame(columns=RELEASE_COLUMNS) for metric in ("MSE", "RMSE", "MAPE", "CRPS"): frame[metric] = pd.to_numeric(frame[metric], errors="coerce") frame.loc[~np.isfinite(frame[metric]), metric] = np.nan frame["prediction_length"] = pd.to_numeric(frame["prediction_length"], errors="coerce") if "MAPE_eligible" not in frame.columns: frame["MAPE_eligible"] = frame["MAPE"].notna() else: frame["MAPE_eligible"] = frame["MAPE_eligible"].fillna(frame["MAPE"].notna()) frame = frame.dropna(subset=["release_id", "dataset", "model", "prediction_length"]) if frame.empty: return pd.DataFrame(columns=RELEASE_COLUMNS) canonical = frame["metric_version"].eq(CANONICAL_METRIC_VERSION) if canonical.any(): frame = frame[canonical].copy() frame = _with_baseline_rtg(frame) frame["__evaluated_order"] = pd.to_datetime(frame["evaluated_at"], errors="coerce", utc=True) frame = frame.sort_values(["model", "release_id", "__evaluated_order"]) frame = frame.drop_duplicates(["model", "release_id"], keep="last") return frame.drop(columns="__evaluated_order").loc[:, RELEASE_COLUMNS].reset_index(drop=True) def load_release_metrics(results_root: str | Path) -> pd.DataFrame: """Load and deduplicate all release-level metrics saved by the live evaluator.""" root = Path(results_root) metric_releases = load_metric_release_metrics(root) if not metric_releases.empty: return metric_releases rows: list[dict] = [] for path in sorted((root / "saved_evals").glob("*/*/*/*.json")): try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): continue row = compute_release_metrics(payload) if row is not None and row["dataset"] and row["model"]: rows.append(row) if not rows: return pd.DataFrame(columns=RELEASE_COLUMNS) frame = pd.DataFrame(rows) frame["__evaluated_order"] = pd.to_datetime(frame["evaluated_at"], errors="coerce", utc=True) frame = frame.sort_values(["model", "release_id", "__evaluated_order"]) frame = frame.drop_duplicates(["model", "release_id"], keep="last") return frame.drop(columns="__evaluated_order").loc[:, RELEASE_COLUMNS].reset_index(drop=True) def _trend_statistics(group: pd.DataFrame) -> tuple[float, float]: ordered = group.sort_values(["release_time", "evaluated_at"]) values = pd.to_numeric(ordered["MSE"], errors="coerce").dropna().to_numpy(dtype=float) if values.size < 2: return np.nan, np.nan if np.allclose(values, values[0]): return 0.0, 1.0 result = stats.kendalltau(np.arange(values.size), values, nan_policy="omit") return float(result.statistic), float(result.pvalue) def aggregate_by_dataset(releases: pd.DataFrame) -> pd.DataFrame: if releases.empty: return pd.DataFrame(columns=DATASET_COLUMNS) keys = ["model", "dataset", "domain", "frequency"] rows: list[dict] = [] for key, group in releases.groupby(keys, dropna=False, sort=True): tau, pvalue = _trend_statistics(group) mse_values = pd.to_numeric(group["MSE"], errors="coerce") row = dict(zip(keys, key)) prediction_lengths = sorted( int(value) for value in pd.to_numeric(group["prediction_length"], errors="coerce").dropna().unique() ) row["prediction_lengths"] = ";".join(str(value) for value in prediction_lengths) for metric in METRIC_COLUMNS: row[metric] = float(pd.to_numeric(group[metric], errors="coerce").mean()) row.update( { "Stability": float(mse_values.std(ddof=1)) if mse_values.count() > 1 else np.nan, "Improvement": tau, "Improvement_pvalue": pvalue, "n_releases": int(group["release_id"].nunique()), "MAPE_coverage": float(group["MAPE"].notna().mean()), "first_release": str(group["release_time"].min()), "last_release": str(group["release_time"].max()), } ) rows.append(row) return pd.DataFrame(rows).loc[:, DATASET_COLUMNS] def _main_connected_component( models: list[str], edges: set[tuple[str, str]], ) -> set[str]: adjacency = {model: set() for model in models} for model_a, model_b in edges: adjacency[model_a].add(model_b) adjacency[model_b].add(model_a) components: list[set[str]] = [] unseen = set(models) while unseen: start = min(unseen) component = {start} frontier = [start] unseen.remove(start) while frontier: node = frontier.pop() neighbours = adjacency[node] & unseen unseen.difference_update(neighbours) component.update(neighbours) frontier.extend(sorted(neighbours)) components.append(component) components_with_edges = [ component for component in components if any(model_a in component and model_b in component for model_a, model_b in edges) ] if not components_with_edges: return set() return max( components_with_edges, key=lambda component: (len(component), tuple(sorted(component))), ) def build_rank_table(releases: pd.DataFrame) -> pd.DataFrame: """Compute the dataset-balanced pairwise historical ranking.""" required = {"release_id", "release_time", "dataset", "model", "MSE", "CRPS"} if releases.empty or not required.issubset(releases.columns): return pd.DataFrame(columns=RANK_COLUMNS) work = releases.loc[:, sorted(required)].copy() work["model"] = work["model"].map(_canonical_model_name) work["MSE"] = pd.to_numeric(work["MSE"], errors="coerce") work["CRPS"] = pd.to_numeric(work["CRPS"], errors="coerce") work["__release_time"] = pd.to_datetime( work["release_time"], errors="coerce", utc=True ) work = work.dropna(subset=["release_id", "dataset", "model"]) work = work.drop_duplicates(["model", "dataset", "release_id"], keep="last") models = sorted(str(model) for model in work["model"].unique()) if not models: return pd.DataFrame(columns=RANK_COLUMNS) pair_scores: dict[str, list[float]] = {model: [] for model in models} opponents: dict[str, set[str]] = {model: set() for model in models} evidence_releases: dict[str, set[tuple[str, str]]] = {model: set() for model in models} evidence_datasets: dict[str, set[str]] = {model: set() for model in models} evidence_times: dict[str, list[pd.Timestamp]] = {model: [] for model in models} eligible_edges: set[tuple[str, str]] = set() model_frames = { model: work[work["model"] == model].drop(columns="model") for model in models } for model_a, model_b in combinations(models, 2): paired = model_frames[model_a].merge( model_frames[model_b], on=["release_id", "dataset"], how="inner", suffixes=("_a", "_b"), ) if paired.empty: continue valid = ( np.isfinite(paired["MSE_a"]) & np.isfinite(paired["MSE_b"]) & np.isfinite(paired["CRPS_a"]) & np.isfinite(paired["CRPS_b"]) ) paired = paired.loc[valid].copy() if paired.empty: continue release_times = paired["__release_time_a"].combine_first( paired["__release_time_b"] ) valid_times = release_times.dropna() span_days = ( float((valid_times.max() - valid_times.min()).total_seconds() / 86400.0) if not valid_times.empty else np.nan ) shared_release_count = int(len(paired)) shared_dataset_count = int(paired["dataset"].nunique()) shared_keys = set(zip(paired["dataset"].astype(str), paired["release_id"].astype(str))) shared_datasets = set(paired["dataset"].astype(str)) for model in (model_a, model_b): evidence_releases[model].update(shared_keys) evidence_datasets[model].update(shared_datasets) evidence_times[model].extend(valid_times.tolist()) eligible = ( shared_release_count >= MIN_PAIR_SHARED_RELEASES and shared_dataset_count >= MIN_PAIR_SHARED_DATASETS and np.isfinite(span_days) and span_days >= MIN_PAIR_SPAN_DAYS ) if not eligible: continue mse_score = np.where( paired["MSE_a"] < paired["MSE_b"], 1.0, np.where(paired["MSE_a"] > paired["MSE_b"], 0.0, 0.5), ) crps_score = np.where( paired["CRPS_a"] < paired["CRPS_b"], 1.0, np.where(paired["CRPS_a"] > paired["CRPS_b"], 0.0, 0.5), ) paired["__score_a"] = 0.5 * (mse_score + crps_score) win_rate_a = float(paired.groupby("dataset")["__score_a"].mean().mean()) win_rate_b = 1.0 - win_rate_a eligible_edges.add((model_a, model_b)) opponents[model_a].add(model_b) opponents[model_b].add(model_a) pair_scores[model_a].append(win_rate_a) pair_scores[model_b].append(win_rate_b) main_component = _main_connected_component(models, eligible_edges) rows: list[dict] = [] for model in models: times = evidence_times[model] covered_span_days = ( float((max(times) - min(times)).total_seconds() / 86400.0) if times else np.nan ) win_rate = float(np.mean(pair_scores[model])) if pair_scores[model] else np.nan official = ( model in main_component and len(opponents[model]) >= MIN_OFFICIAL_OPPONENTS ) rows.append( { "Rank": pd.NA, "Status": "Official" if official else "Provisional", "model": model, "WinRate": win_rate, "eligible_opponents": int(len(opponents[model])), "shared_releases": int(len(evidence_releases[model])), "shared_datasets": int(len(evidence_datasets[model])), "covered_span_days": covered_span_days, } ) table = pd.DataFrame(rows) official_mask = table["Status"].eq("Official") official = table[official_mask].sort_values( ["WinRate", "model"], ascending=[False, True] ) official["Rank"] = np.arange(1, len(official) + 1) provisional = table[~official_mask].sort_values( ["eligible_opponents", "WinRate", "model"], ascending=[False, False, True], na_position="last", ) table = pd.concat([official, provisional], ignore_index=True) table["Rank"] = pd.array(table["Rank"], dtype="Int64") return table.loc[:, RANK_COLUMNS] def aggregate_overall(dataset_scores: pd.DataFrame, rank_table: pd.DataFrame) -> pd.DataFrame: if dataset_scores.empty: return pd.DataFrame( columns=[ "Rank", "Status", "model", *METRIC_COLUMNS, "WinRate", "eligible_opponents", "shared_releases", "shared_datasets", "covered_span_days", "Stability", "Improvement", "SignificantImprovementRate", "n_datasets", "n_releases", "MAPE_dataset_coverage", "first_release", "last_release", ] ) rows = [] for model, group in dataset_scores.groupby("model", sort=True): row = {"model": model} for metric in METRIC_COLUMNS: row[metric] = float(pd.to_numeric(group[metric], errors="coerce").mean()) significant_improvement = ( group["Improvement"].lt(0) & group["Improvement_pvalue"].lt(0.05) ) row.update( { "Stability": float(group["Stability"].mean()), "Improvement": float(group["Improvement"].mean()), "SignificantImprovementRate": float(significant_improvement.mean()), "n_datasets": int(group["dataset"].nunique()), "n_releases": int(group["n_releases"].sum()), "MAPE_dataset_coverage": float(group["MAPE"].notna().mean()), "first_release": str(group["first_release"].min()), "last_release": str(group["last_release"].max()), } ) rows.append(row) overall = pd.DataFrame(rows) overall = overall.merge(rank_table, on="model", how="left", validate="one_to_one") ordered = [ "Rank", "Status", "model", *METRIC_COLUMNS, "WinRate", "eligible_opponents", "shared_releases", "shared_datasets", "covered_span_days", "Stability", "Improvement", "SignificantImprovementRate", "n_datasets", "n_releases", "MAPE_dataset_coverage", "first_release", "last_release", ] return overall.sort_values(["Rank", "MSE"], na_position="last").loc[:, ordered].reset_index(drop=True) def build_live_aggregates(releases: pd.DataFrame) -> dict[str, pd.DataFrame]: dataset_scores = aggregate_by_dataset(releases) rank_table = build_rank_table(releases) overall = aggregate_overall(dataset_scores, rank_table) return { "releases": releases, "by_dataset": dataset_scores, "rank": rank_table, "overall": overall, } def build_fixed_window_aggregates( releases: pd.DataFrame, *, reference_time: pd.Timestamp, days: int, ) -> dict[str, pd.DataFrame]: """Use cumulative history for ranking and a trailing window for metrics.""" release_times = pd.to_datetime(releases["release_time"], errors="coerce", utc=True) historical_releases = releases[release_times.le(reference_time)].copy() cutoff = reference_time - pd.Timedelta(days=days) window_releases = releases[ release_times.gt(cutoff) & release_times.le(reference_time) ].copy() rank_table = build_rank_table(historical_releases) overall = aggregate_overall(aggregate_by_dataset(window_releases), rank_table) return { "releases": window_releases, "rank": rank_table, "overall": overall, } def load_snapshot_release_metrics( results_root: str | Path, *, model_allowlist: set[str], existing_model_dataset_keys: set[tuple[str, str]], ) -> pd.DataFrame: """Load one latest snapshot release for allowlisted model-datasets without history.""" root = Path(results_root) if not model_allowlist: return pd.DataFrame(columns=RELEASE_COLUMNS) rows: list[dict] = [] for model_dir in sorted(root.iterdir()): if not model_dir.is_dir(): continue result_path = model_dir / "all_results.csv" if not result_path.exists(): continue try: result = pd.read_csv(result_path) except (OSError, pd.errors.ParserError, pd.errors.EmptyDataError): continue if result.empty or not {"dataset", "model", *SNAPSHOT_METRIC_COLUMNS.values()}.issubset(result.columns): continue result = result.copy() result["model"] = result["model"].map(_canonical_model_name) model_name = _canonical_model_name(result["model"].dropna().iloc[0]) if result["model"].notna().any() else "" model_key = _model_key(model_name) if model_key not in model_allowlist: continue meta = {} meta_path = model_dir / "online_meta.json" if meta_path.exists(): try: meta = json.loads(meta_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): meta = {} evaluated_at = str( meta.get("evaluated_at") or datetime.fromtimestamp(result_path.stat().st_mtime, timezone.utc).isoformat() ) for _, row in result.iterrows(): dataset = str(row.get("dataset", "")).strip() if not dataset: continue dataset_key = _dataset_key(dataset) if (model_key, dataset_key) in existing_model_dataset_keys: continue parts = dataset.rsplit("/", 2) frequency = parts[1] if len(parts) == 3 else "unknown" prediction_length = pd.to_numeric(row.get("prediction_length"), errors="coerce") if pd.isna(prediction_length): prediction_length = 0 metrics = { metric: pd.to_numeric(row.get(column), errors="coerce") for metric, column in SNAPSHOT_METRIC_COLUMNS.items() } metrics = { metric: float(value) if pd.notna(value) and np.isfinite(value) else np.nan for metric, value in metrics.items() } rows.append( { "release_id": f"snapshot|{dataset}", "release_time": evaluated_at, "evaluated_at": evaluated_at, "issued_at": None, "protocol": "legacy-latest-snapshot", "dataset": dataset, "domain": str(row.get("domain", "Unknown")).strip(), "frequency": frequency, "prediction_length": int(prediction_length), "model": model_name, "metric_version": "legacy-snapshot-v0", "MSE": metrics["MSE"], "RMSE": metrics["RMSE"], "MAPE": metrics["MAPE"], "CRPS": metrics["CRPS"], "RTG": np.nan, "MAPE_eligible": bool(np.isfinite(metrics["MAPE"])), } ) if not rows: return pd.DataFrame(columns=RELEASE_COLUMNS) return pd.DataFrame(rows).loc[:, RELEASE_COLUMNS].reset_index(drop=True) def _write_csv_atomic(frame: pd.DataFrame, path: Path) -> None: temporary = path.with_suffix(path.suffix + ".tmp") frame.to_csv(temporary, index=False) temporary.replace(path) def write_live_aggregates(results_root: str | Path) -> dict[str, Path]: root = Path(results_root) aggregate_dir = root / "aggregates" aggregate_dir.mkdir(parents=True, exist_ok=True) releases = load_release_metrics(root) if not releases.empty: releases = releases.copy() releases["model"] = releases["model"].map(_canonical_model_name) model_allowlist = _load_model_allowlist(root) existing_model_dataset_keys = ( set(zip(releases["model"].map(_model_key), releases["dataset"].map(_dataset_key))) if not releases.empty else set() ) snapshot_releases = load_snapshot_release_metrics( root, model_allowlist=model_allowlist, existing_model_dataset_keys=existing_model_dataset_keys, ) if ( not releases.empty and "metric_version" in releases.columns and releases["metric_version"].eq(CANONICAL_METRIC_VERSION).any() ): # Never mix legacy snapshot metrics with the canonical release history. snapshot_releases = pd.DataFrame(columns=RELEASE_COLUMNS) live_dataset_keys = _load_live_dataset_keys(root) if live_dataset_keys and not snapshot_releases.empty: snapshot_releases = snapshot_releases[ snapshot_releases["dataset"].map(_dataset_key).isin(live_dataset_keys) ].copy() if not snapshot_releases.empty: releases = pd.concat([releases, snapshot_releases], ignore_index=True) if model_allowlist and not releases.empty: releases = releases[releases["model"].map(_model_key).isin(model_allowlist)].copy() if not releases.empty: releases = _with_baseline_rtg(releases) tables = build_live_aggregates(releases) filenames = { "releases": "live_release_metrics.csv", "by_dataset": "live_results_by_dataset.csv", "rank": "live_rank_up_to_now.csv", "overall": "live_overall_up_to_now.csv", } paths: dict[str, Path] = {} for name, filename in filenames.items(): path = aggregate_dir / filename _write_csv_atomic(tables[name], path) paths[name] = path # "Last 24 hours / 7 days / 30 days" must advance during an outage too. # Anchoring to the last available score incorrectly labels old scores fresh. reference_time = pd.Timestamp.now(tz="UTC") fixed_windows: dict[str, dict[str, object]] = {} if pd.notna(reference_time): for label, days in (("1d", 1), ("7d", 7), ("30d", 30)): window_tables = build_fixed_window_aggregates( tables["releases"], reference_time=reference_time, days=days, ) for table_name in ("rank", "overall"): filename = f"live_{table_name}_{label}.csv" path = aggregate_dir / filename _write_csv_atomic(window_tables[table_name], path) paths[f"{table_name}_{label}"] = path window_releases = window_tables["releases"] fixed_windows[label] = { "days": days, "num_model_releases": int(len(window_releases)), "num_unique_releases": int(window_releases["release_id"].nunique()), "ranking_scope": "Cumulative release history through reference_time", "metrics_scope": f"Trailing {days * 24} hours through reference_time", } releases = tables["releases"] metadata = { "generated_at": datetime.now(timezone.utc).isoformat(), "scope": ( "Canonical metrics-only release history when available; otherwise legacy " "release history plus latest allowlisted snapshots" ), "release_deduplication": ( "One row per model, dataset, data source timestamp, and prediction length" ), "dataset_aggregation": "Arithmetic mean over releases, then equal-weight mean over datasets", "persistence": ( "Public aggregates contain metric rows only. Forecast/target snapshots may be " "retained privately for audit and are excluded from Space uploads." ), "metric_version": CANONICAL_METRIC_VERSION, "normalization": ( "Context-only z-normalization for MSE/RMSE and quantile-CRPS; " "original-scale MAPE" ), "mape": ( "Original-scale p50 MAPE; omitted when any target is too close to zero" ), "crps": "Twice the mean pinball loss over q=0.1,...,0.9 after context-only z-normalization", "rtg": ( "100 * (MSE_seasonal_naive - MSE_model) / " "(MSE_seasonal_naive + MSE_model) on metrics-only matched releases" ), "ranking": ( "Dataset-balanced pairwise MSE/CRPS win rate on shared releases; eligible " "pairs require >=30 releases, >=5 datasets, and >=7 days of coverage; " "official models require >=3 opponents in the main comparison component" ), "temporal_metrics": "Per-dataset MSE stability and Kendall tau, then equal-weight mean over datasets", "num_models": int(releases["model"].nunique()) if not releases.empty else 0, "num_datasets": int(releases["dataset"].nunique()) if not releases.empty else 0, "num_model_releases": int(len(releases)), "num_unique_releases": int(releases["release_id"].nunique()) if not releases.empty else 0, "fixed_windows": fixed_windows, "window_reference_time": reference_time.isoformat(), "protocol_counts": ( releases["protocol"].fillna("unknown").value_counts().to_dict() if not releases.empty and "protocol" in releases.columns else {} ), } metadata_path = aggregate_dir / "live_metadata.json" temporary = metadata_path.with_suffix(".json.tmp") temporary.write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8") temporary.replace(metadata_path) paths["metadata"] = metadata_path return paths