src/tsfm_bench/eval/live_aggregation.py

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("<f8", copy=False).tobytes()
    ).hexdigest()[:20]
    return f"{dataset}|window:{digest}", evaluated_at


def _z_scale(context: np.ndarray) -> 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