src/utils.py

from __future__ import annotations

import json
import os
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path

import numpy as np
import pandas as pd
from scipy import stats

from src.display.formatting import format_timestamp_utc8
from src.rank_based_metric import compute_rank_score

METRIC_COLUMNS = [
    "eval_metrics/MSE[mean]",
    "eval_metrics/MSE[0.5]",
    "eval_metrics/MAE[0.5]",
    "eval_metrics/MASE[0.5]",
    "eval_metrics/MAPE[0.5]",
    "eval_metrics/sMAPE[0.5]",
    "eval_metrics/MSIS",
    "eval_metrics/RMSE[mean]",
    "eval_metrics/NRMSE[mean]",
    "eval_metrics/ND[0.5]",
    "eval_metrics/mean_weighted_sum_quantile_loss",
]

DISPLAY_METRICS = [
    "eval_metrics/MSE[mean]",
    "eval_metrics/mean_weighted_sum_quantile_loss",
    "eval_metrics/MASE[0.5]",
    "eval_metrics/MAE[0.5]",
    "eval_metrics/RMSE[mean]",
    "eval_metrics/sMAPE[0.5]",
    "eval_metrics/MSIS",
    "eval_metrics/ND[0.5]",
    "eval_metrics/NRMSE[mean]",
    "eval_metrics/MAPE[0.5]",
]

METRIC_LABELS = {
    "eval_metrics/MSE[mean]": "MSE",
    "eval_metrics/MASE[0.5]": "MASE",
    "eval_metrics/mean_weighted_sum_quantile_loss": "CRPS",
    "eval_metrics/MAE[0.5]": "MAE",
    "eval_metrics/RMSE[mean]": "RMSE",
    "eval_metrics/sMAPE[0.5]": "sMAPE",
    "eval_metrics/MSIS": "MSIS",
    "eval_metrics/ND[0.5]": "ND",
    "eval_metrics/NRMSE[mean]": "NRMSE",
    "eval_metrics/MAPE[0.5]": "MAPE",
}

RANK_SCORE_COLUMN = "RankScore"
RANK_SCORE_RANK_COLUMN = "RankScore_Rank"
RANK_SCORE_BASE_METRICS = [
    "eval_metrics/MSE[mean]",
    "eval_metrics/mean_weighted_sum_quantile_loss",
]
GEOMEAN_EPSILON = 1e-12

VALUE_COLUMNS = ["MSE", "CRPS", RANK_SCORE_COLUMN] + [
    METRIC_LABELS[m] for m in DISPLAY_METRICS if METRIC_LABELS[m] not in {"MSE", "CRPS"}
]
RANK_COLUMNS = ["MSE_Rank", "CRPS_Rank", RANK_SCORE_RANK_COLUMN] + [
    f"{METRIC_LABELS[m]}_Rank"
    for m in DISPLAY_METRICS
    if METRIC_LABELS[m] not in {"MSE", "CRPS"}
]

LEGACY_MODEL_NAMES = {
    "TSFM1": "Chronos-Bolt-Tiny",
    "TSFM2": "Chronos-Bolt-Base",
    "TSFM3": "Chronos-2",
}

RESULT_COLUMNS = [
    "dataset",
    "model",
    *METRIC_COLUMNS,
    "domain",
    "num_variates",
]

CANONICAL_DOMAINS = [
    "Finance",
    "Climate",
    "Ocean",
    "Hydrology",
    "Transportation",
    "Web",
    "Hazards",
]

DOMAIN_DISPLAY = {
    "Finance": "Finance",
    "Econ/Fin": "Finance",
    "Sales": "Finance",
    "Climate": "Climate",
    "Weather": "Climate",
    "Air Quality": "Climate",
    "Nature": "Climate",
    "Energy": "Climate",
    "Ocean": "Ocean",
    "Hydrology": "Hydrology",
    "Transport": "Transportation",
    "Transportation": "Transportation",
    "Web": "Web",
    "Web/CloudOps": "Web",
    "Hazards": "Hazards",
    "Healthcare": "Hazards",
}


def _read_result_csv(path: str | Path) -> pd.DataFrame:
    df = pd.read_csv(
        path,
        usecols=lambda column: column in RESULT_COLUMNS,
        index_col=False,
    )
    missing = [column for column in RESULT_COLUMNS if column not in df.columns]
    if missing:
        raise ValueError(f"{path} is missing required result columns: {missing}")
    return df


def _dataset_key(value: object) -> str:
    return str(value).split("/")[0].strip().lower()


def _model_key(value: object) -> str:
    return re.sub(r"[^a-z0-9]+", "", str(value).lower())


def _load_model_allowlist(root_dir: str | Path) -> set[str]:
    path = Path(root_dir) / "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(ds_properties: str | Path | None) -> set[str]:
    if ds_properties is None:
        return set()
    path = Path(ds_properties)
    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()}


GIFT_AGGREGATE_FILES = {
    "prediction_length": "results_by_prediction_length.csv",
    "domain": "results_by_domain.csv",
    "frequency": "results_by_frequency.csv",
}

GIFT_AGGREGATE_LABELS = {
    "prediction_length": "Prediction Length",
    "domain": "Domain",
    "frequency": "Frequency",
}

LIVE_AGGREGATE_FILES = {
    "overall": "live_overall_up_to_now.csv",
    "rank": "live_rank_up_to_now.csv",
}


def format_number(value):
    """Format a displayed leaderboard metric without hiding tiny non-zero values."""
    if pd.isna(value):
        return "n/a"
    number = float(value)
    if not np.isfinite(number):
        return "n/a"
    if number != 0.0 and abs(number) < 0.005:
        return f"{number:.2e}"
    return f"{number:.2f}"


def format_rank_number(value):
    """Format rank-like columns as positive integers for leaderboard display."""
    if pd.isna(value):
        return "n/a"
    number = float(value)
    if not np.isfinite(number):
        return "n/a"
    return str(max(1, int(round(number))))


def _is_rank_display_column(column: object) -> bool:
    name = str(column)
    return "Rank" in name and name != RANK_SCORE_COLUMN


def aggregate_gmean(series: pd.Series) -> float:
    values = pd.to_numeric(series, errors="coerce").dropna()
    values = values[np.isfinite(values) & (values >= 0.0)]
    if values.empty:
        return np.nan
    values = values.clip(lower=GEOMEAN_EPSILON)
    return float(stats.gmean(values))


def aggregate_rank_mean(series: pd.Series) -> float:
    return float(series.mean(skipna=True))


def format_df(df: pd.DataFrame) -> pd.DataFrame:
    formatted = df.copy()
    for col in formatted.columns:
        if pd.api.types.is_numeric_dtype(formatted[col]):
            formatter = format_rank_number if _is_rank_display_column(col) else format_number
            formatted[col] = formatted[col].map(formatter)
    return formatted


def unify_freq(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["frequency"] = df["frequency"].str.replace(r"\d+", "", regex=True)
    df["frequency"] = df["frequency"].str.split("-").str[0]
    return df


def load_all_results(root_dir: str = "results") -> pd.DataFrame:
    frames = []
    for subdir, _, files in os.walk(root_dir):
        for file in files:
            if file != "all_results.csv":
                continue
            frame = _read_result_csv(os.path.join(subdir, file))
            if frame.empty:
                continue
            frames.append(frame)
    if not frames:
        return pd.DataFrame(columns=RESULT_COLUMNS)
    df = pd.concat(frames, ignore_index=True)
    df = df.dropna(subset=["dataset"], how="any")
    df = df[df["dataset"].astype(str).str.len() > 0]
    if df.empty:
        return pd.DataFrame(columns=RESULT_COLUMNS)
    df = df.sort_values(by=["model", "dataset"]).reset_index(drop=True)
    df["model"] = df["model"].replace(LEGACY_MODEL_NAMES)
    model_allowlist = _load_model_allowlist(root_dir)
    if model_allowlist:
        df = df[df["model"].map(_model_key).isin(model_allowlist)].copy()
        if df.empty:
            return pd.DataFrame(columns=RESULT_COLUMNS)
    
    parts = df["dataset"].astype(str).str.split("/", expand=True)
    if parts.shape[1] < 3:
        for idx in range(parts.shape[1], 3):
            parts[idx] = "short" if idx == 2 else ""
    df["dataset"] = parts[0]
    df["frequency"] = parts[1]
    df["term_length"] = parts[2]
    return df


def enrich_with_properties(df: pd.DataFrame, ds_properties_path: str) -> pd.DataFrame:
    props = pd.read_csv(ds_properties_path)
    props["dataset"] = props["dataset"].str.lower().str.replace(" ", "_")
    props["dataset"] = props["dataset"].str.replace("-", "_")
    props["dataset"] = props["dataset"].apply(lambda x: re.sub("_+", "_", x).strip("_"))
    props_map = props.set_index("dataset").T.to_dict()

    for dataset_key, values in props_map.items():
        for key, value in values.items():
            if key == "frequency" and df.loc[df["dataset"] == dataset_key, "frequency"].notna().all():
                continue
            df.loc[df["dataset"] == dataset_key, key] = value
    return unify_freq(df)


def prepare_results_df(
    root_dir: str = "results",
    ds_properties: str | None = None,
) -> pd.DataFrame:
    if ds_properties is None:
        ds_properties = str(Path(root_dir) / "dataset_properties.csv")

    df = load_all_results(root_dir)
    if df.empty:
        return df
    live_dataset_keys = _load_live_dataset_keys(ds_properties)
    if live_dataset_keys:
        df = df[df["dataset"].map(_dataset_key).isin(live_dataset_keys)].copy()
        if df.empty:
            return pd.DataFrame(columns=[*RESULT_COLUMNS, "frequency", "term_length"])
    if Path(ds_properties).exists():
        df = enrich_with_properties(df, ds_properties)
    if "domain" in df.columns:
        df["domain"] = df["domain"].map(display_domain)

    for metric in METRIC_COLUMNS:
        df[metric] = pd.to_numeric(df[metric], errors="coerce")
        df.loc[~np.isfinite(df[metric]), metric] = np.nan

    return _add_per_dataset_ranks(df)


def _add_per_dataset_ranks(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df
    df = df.copy()
    group_keys = ["dataset", "term_length", "frequency"]
    for metric in DISPLAY_METRICS:
        rank_col = f"Rank_{metric}"
        df[rank_col] = df.groupby(group_keys)[metric].rank(method="first", ascending=True)
    return df


def _empty_grouped_dfs() -> dict[str, pd.DataFrame | list[str]]:
    rank_cols = [f"Rank_{metric}" for metric in DISPLAY_METRICS]
    return {
        "overall_values": pd.DataFrame(columns=DISPLAY_METRICS),
        "overall_ranks": pd.DataFrame(columns=rank_cols),
        "dataset_values": {},
        "dataset_ranks": {},
        "datasets": [],
    }


def get_grouped_dfs(
    root_dir: str = "results",
    ds_properties: str | None = None,
) -> dict[str, pd.DataFrame | list[str]]:
    df = prepare_results_df(root_dir, ds_properties)
    if df.empty:
        return _empty_grouped_dfs()

    rank_cols = [f"Rank_{metric}" for metric in DISPLAY_METRICS]
    overall_values = df.groupby(["model"])[DISPLAY_METRICS].agg(aggregate_gmean)
    overall_ranks = df.groupby(["model"])[rank_cols].agg(aggregate_rank_mean)

    overall_values = overall_values.rename(columns=METRIC_LABELS)
    overall_ranks = overall_ranks.rename(
        columns={f"Rank_{metric}": f"{METRIC_LABELS[metric]}_Rank" for metric in DISPLAY_METRICS}
    )
    rank_score = compute_rank_score(
        df,
        metric_columns=RANK_SCORE_BASE_METRICS,
        group_columns=("dataset", "term_length", "frequency"),
    )
    overall_values = overall_values.join(rank_score, how="left")
    if RANK_SCORE_COLUMN in overall_values.columns:
        overall_ranks[RANK_SCORE_RANK_COLUMN] = overall_values[RANK_SCORE_COLUMN].rank(
            method="average",
            ascending=False,
        )

    datasets = sorted(df["dataset"].dropna().unique())
    dataset_values: dict[str, pd.DataFrame] = {}
    dataset_ranks: dict[str, pd.DataFrame] = {}
    for dataset in datasets:
        subset = df[df["dataset"] == dataset]
        if subset.empty:
            continue
        values = subset.groupby("model")[DISPLAY_METRICS].agg(aggregate_gmean).rename(
            columns=METRIC_LABELS
        )
        ranks = subset.groupby("model")[rank_cols].agg(aggregate_rank_mean).rename(
            columns={f"Rank_{metric}": f"{METRIC_LABELS[metric]}_Rank" for metric in DISPLAY_METRICS}
        )
        dataset_values[dataset] = values
        dataset_ranks[dataset] = ranks

    return {
        "overall_values": overall_values,
        "overall_ranks": overall_ranks,
        "dataset_values": dataset_values,
        "dataset_ranks": dataset_ranks,
        "datasets": datasets,
    }


def prepare_values_table(values_df: pd.DataFrame) -> pd.DataFrame:
    if values_df is None or values_df.empty:
        return pd.DataFrame(columns=["Model"] + VALUE_COLUMNS)
    table = values_df.reset_index().rename(columns={"model": "Model"})
    cols = ["Model"] + [col for col in VALUE_COLUMNS if col in table.columns]
    table = table[cols]
    if "MSE" in table.columns and not table["MSE"].isna().all():
        table = table.sort_values(by=["MSE"])
        return format_df(table)
    if "RankScore" in table.columns and not table["RankScore"].isna().all():
        table = table.sort_values(by=["RankScore"], ascending=False)
    return format_df(table)


def prepare_ranks_table(ranks_df: pd.DataFrame) -> pd.DataFrame:
    if ranks_df is None or ranks_df.empty:
        return pd.DataFrame(columns=["Model"] + RANK_COLUMNS)
    table = ranks_df.reset_index().rename(columns={"model": "Model"})
    cols = ["Model"] + [col for col in RANK_COLUMNS if col in table.columns]
    table = table[cols]
    if "MSE_Rank" in table.columns and not table["MSE_Rank"].isna().all():
        table = table.sort_values(by=["MSE_Rank"])
        return format_df(table)
    if RANK_SCORE_RANK_COLUMN in table.columns and not table[RANK_SCORE_RANK_COLUMN].isna().all():
        table = table.sort_values(by=[RANK_SCORE_RANK_COLUMN])
    return format_df(table)


def load_gift_aggregate_table(
    root_dir: str = "results",
    dimension: str = "prediction_length",
) -> pd.DataFrame:
    """Load one persisted GIFT-Eval-style grouped result table."""

    if dimension not in GIFT_AGGREGATE_FILES:
        raise ValueError(f"Unknown aggregate dimension: {dimension}")

    label = GIFT_AGGREGATE_LABELS[dimension]
    columns = [label, "Model", "MSE", "CRPS", "Rank", "Configs", "Coverage"]
    path = Path(root_dir) / "aggregates" / GIFT_AGGREGATE_FILES[dimension]
    if not path.exists():
        return pd.DataFrame(columns=columns)
    try:
        table = pd.read_csv(path)
    except (OSError, pd.errors.ParserError, pd.errors.EmptyDataError):
        return pd.DataFrame(columns=columns)

    required = {
        dimension,
        "model",
        "MSE",
        "CRPS",
        "Rank",
        "n_configs",
        "n_group_configs",
        "coverage",
    }
    if table.empty or not required.issubset(table.columns):
        return pd.DataFrame(columns=columns)

    table = table.sort_values([dimension, "Rank", "model"]).copy()
    table["Configs"] = (
        table["n_configs"].fillna(0).astype(int).astype(str)
        + "/"
        + table["n_group_configs"].fillna(0).astype(int).astype(str)
    )
    table["Coverage"] = table["coverage"].map(
        lambda value: "n/a" if pd.isna(value) else f"{100.0 * float(value):.2f}%"
    )
    for metric in ("MSE", "CRPS"):
        table[metric] = pd.to_numeric(table[metric], errors="coerce").map(format_number)
    table["Rank"] = pd.to_numeric(table["Rank"], errors="coerce").map(format_rank_number)
    table = table.rename(columns={dimension: label, "model": "Model"})
    return table.loc[:, columns].reset_index(drop=True)


def load_gift_aggregate_metadata_md(root_dir: str = "results") -> str:
    path = Path(root_dir) / "aggregates" / "metadata.json"
    if not path.exists():
        return "Aggregate metadata is not available yet."
    try:
        metadata = json.loads(path.read_text())
    except (OSError, json.JSONDecodeError):
        return "Aggregate metadata could not be loaded."
    generated = format_timestamp_utc8(metadata.get("generated_at", ""))
    return (
        f"**Protocol:** per-configuration normalization against "
        f"`{metadata.get('baseline_model', 'Seasonal-Naive')}`; geometric mean for "
        f"MSE/CRPS; mean CRPS rank. **Coverage:** "
        f"{metadata.get('num_models', 0)} models × {metadata.get('num_configs', 0)} "
        f"configurations. **Generated:** {generated}."
    )


def load_live_aggregate_table(
    root_dir: str = "results",
    table: str = "overall",
) -> pd.DataFrame:
    """Load the cumulative future-release metric or rank table."""

    if table not in LIVE_AGGREGATE_FILES:
        raise ValueError(f"Unknown live aggregate table: {table}")
    if table == "overall":
        columns = [
            "Rank",
            "Model",
            "MSE",
            "RMSE",
            "MAPE",
            "CRPS",
            "RTG",
            "Stability",
            "Improvement",
            "Datasets",
            "Releases",
        ]
    else:
        columns = [
            "Rank",
            "Status",
            "Model",
            "Win Rate",
            "Eligible Opponents",
            "Shared Releases",
            "Shared Datasets",
            "Covered Span (days)",
        ]

    path = Path(root_dir) / "aggregates" / LIVE_AGGREGATE_FILES[table]
    if not path.exists():
        return pd.DataFrame(columns=columns)
    try:
        frame = pd.read_csv(path)
    except (OSError, pd.errors.ParserError, pd.errors.EmptyDataError):
        return pd.DataFrame(columns=columns)
    if frame.empty:
        return pd.DataFrame(columns=columns)

    if table == "overall":
        source_columns = [
            "Rank",
            "model",
            "MSE",
            "RMSE",
            "MAPE",
            "CRPS",
            "RTG",
            "Stability",
            "Improvement",
            "n_datasets",
            "n_releases",
        ]
        if not set(source_columns).issubset(frame.columns):
            return pd.DataFrame(columns=columns)
        frame = frame[source_columns].rename(
            columns={"model": "Model", "n_datasets": "Datasets", "n_releases": "Releases"}
        )
        for metric in ("MSE", "RMSE", "MAPE", "CRPS", "RTG", "Stability", "Improvement"):
            frame[metric] = pd.to_numeric(frame[metric], errors="coerce").map(format_number)
        frame["Rank"] = pd.to_numeric(frame["Rank"], errors="coerce").map(format_rank_number)
    else:
        source_columns = [
            "Rank",
            "Status",
            "model",
            "WinRate",
            "eligible_opponents",
            "shared_releases",
            "shared_datasets",
            "covered_span_days",
        ]
        if not set(source_columns).issubset(frame.columns):
            return pd.DataFrame(columns=columns)
        frame = frame[source_columns].rename(
            columns={
                "model": "Model",
                "WinRate": "Win Rate",
                "eligible_opponents": "Eligible Opponents",
                "shared_releases": "Shared Releases",
                "shared_datasets": "Shared Datasets",
                "covered_span_days": "Covered Span (days)",
            }
        )
        frame["Rank"] = pd.to_numeric(frame["Rank"], errors="coerce").map(format_rank_number)
        frame["Win Rate"] = pd.to_numeric(frame["Win Rate"], errors="coerce").map(
            lambda value: "n/a" if pd.isna(value) else f"{100.0 * float(value):.2f}%"
        )
        frame["Covered Span (days)"] = pd.to_numeric(
            frame["Covered Span (days)"], errors="coerce"
        ).map(format_number)
    return frame.loc[:, columns].reset_index(drop=True)


def load_live_aggregate_metadata_md(root_dir: str = "results") -> str:
    path = Path(root_dir) / "aggregates" / "live_metadata.json"
    if not path.exists():
        return "Cumulative live metrics are not available yet."
    try:
        metadata = json.loads(path.read_text())
    except (OSError, json.JSONDecodeError):
        return "Cumulative live metric metadata could not be loaded."
    generated = format_timestamp_utc8(metadata.get("generated_at", ""))
    return (
        f"**Up to now:** {metadata.get('num_models', 0)} models, "
        f"{metadata.get('num_datasets', 0)} datasets, and "
        f"{metadata.get('num_unique_releases', 0)} distinct future releases. "
        "Metrics are stored without forecasts or ground truth; RTG is relative to "
        "matched Seasonal-Naive MSE. **Generated:** "
        f"{generated}."
    )

DATASET_SHORT_NAMES = {
    "binance_btcusdt_1h_close": "BTC/USDT",
    "coingecko_bitcoin_7d_price_usd": "Bitcoin USD",
    "nasa_power_shanghai_hourly_t2m": "Shanghai temperature (NASA POWER)",
    "nasa_power_shanghai_hourly_T2M": "Shanghai temperature (NASA POWER)",
    "open_meteo_shanghai_hourly_temperature_2m": "Shanghai temperature",
    "open_meteo_shanghai_air_quality_hourly_pm2_5": "Shanghai PM2.5",
    "noaa_coops_sf_water_level_water_level": "SF Bay water level",
    "noaa_ndbc_46013_realtime_wave_height": "Ocean wave height",
    "nws_ksfo_observations_temperature": "KSFO temperature",
    "usgs_earthquake_all_week_hourly_earthquake_count": "Earthquake count",
    "usgs_potomac_iv_usgs_00060": "Potomac river flow",
    "wikimedia_time_series_hourly_pageviews": "Wikipedia pageviews",
    "data_gov_sg_singapore_rainfall_5min_rainfall_5min_total": "Singapore rainfall",
    "data_gov_sg_singapore_taxi_availability_available_taxi_count": "Singapore available taxis",
    "data_gov_sg_sgx_equities_turnover_monthly_turnover_value_million_sgd": "SGX equities turnover",
}


def display_domain(raw_domain: str) -> str:
    domain = str(raw_domain).strip()
    if domain in DOMAIN_DISPLAY:
        return DOMAIN_DISPLAY[domain]
    lowered = domain.lower()
    if any(token in lowered for token in ("finance", "econ", "sales")):
        return "Finance"
    if any(token in lowered for token in ("climate", "weather", "air", "nature", "energy")):
        return "Climate"
    if "ocean" in lowered:
        return "Ocean"
    if "hydro" in lowered:
        return "Hydrology"
    if any(token in lowered for token in ("transport", "taxi")):
        return "Transportation"
    if any(token in lowered for token in ("web", "cloud", "wiki")):
        return "Web"
    return "Hazards"


def dataset_short_name(dataset: str) -> str:
    base = dataset.split("/")[0].strip()
    if base in DATASET_SHORT_NAMES:
        return DATASET_SHORT_NAMES[base]
    label = base.replace("_", " ")
    for prefix in ("open meteo ", "noaa coops ", "noaa ndbc ", "nws ", "usgs ", "wikimedia "):
        if label.lower().startswith(prefix):
            label = label[len(prefix) :]
    return label[:40] + ("…" if len(label) > 40 else "")


def lookup_dataset_domain(dataset: str, root_dir: str = "results") -> str:
    # Primary: dataset_properties.csv
    base = dataset.split("/")[0].strip()
    props_path = Path(root_dir) / "dataset_properties.csv"
    if props_path.exists():
        props = pd.read_csv(props_path)
        match = props[props["dataset"].map(_dataset_key) == _dataset_key(base)]
        if not match.empty and "domain" in match.columns:
            return str(match.iloc[0]["domain"])

    # Fallback: read domain column from any all_results.csv
    for subdir in Path(root_dir).iterdir():
        csv_path = subdir / "all_results.csv"
        if not csv_path.exists():
            continue
        try:
            df = _read_result_csv(csv_path)
            if "domain" not in df.columns or "dataset" not in df.columns:
                continue
            # dataset column in all_results is "dataset_id/freq/term", strip suffix
            df["_ds_key"] = df["dataset"].astype(str).str.split("/").str[0]
            match = df[df["_ds_key"].str.lower() == base.lower()]
            if not match.empty:
                return str(match.iloc[0]["domain"])
        except Exception:
            continue
    return "Other"


DOMAIN_TAB_ORDER = [
    *CANONICAL_DOMAINS,
]


def group_datasets_by_domain(
    datasets: list[str],
    root_dir: str = "results",
) -> dict[str, list[str]]:
    by_domain: dict[str, list[str]] = {}
    seen: set[str] = set()
    for dataset in datasets:
        dataset = dataset.split("/")[0].strip()
        if dataset in seen:
            continue
        seen.add(dataset)
        domain = display_domain(lookup_dataset_domain(dataset, root_dir))
        by_domain.setdefault(domain, []).append(dataset)

    ordered: dict[str, list[str]] = {}
    for domain in DOMAIN_TAB_ORDER:
        if domain in by_domain:
            ordered[domain] = by_domain.pop(domain)
    for domain in sorted(by_domain):
        ordered[domain] = by_domain[domain]
    return ordered


def _parse_timestamp(value: object) -> datetime | None:
    if not value or value == "n/a":
        return None
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except (TypeError, ValueError):
        return None


def _read_json(path: Path) -> dict:
    if not path.exists():
        return {}
    try:
        return json.loads(path.read_text())
    except (OSError, json.JSONDecodeError):
        return {}


def leaderboard_refresh_timestamp(root_dir: str = "results") -> str:
    """Return the latest timestamp that can change leaderboard-visible results."""
    root = Path(root_dir)
    candidates: list[datetime] = []

    status = _read_json(root / "online_status.json")
    for key in ("finished_at", "pushed_at"):
        parsed = _parse_timestamp(status.get(key))
        if parsed is not None:
            candidates.append(parsed)

    for metadata_name in ("metadata.json", "live_metadata.json"):
        metadata = _read_json(root / "aggregates" / metadata_name)
        parsed = _parse_timestamp(metadata.get("generated_at"))
        if parsed is not None:
            candidates.append(parsed)

    if not candidates:
        return "n/a"
    return format_timestamp_utc8(max(candidates).isoformat())


def build_leaderboard_summary_html(root_dir: str = "results") -> str:
    df = prepare_results_df(root_dir)
    if df.empty:
        return """
        <div class="summary-grid">
          <div class="summary-card"><div class="summary-value">0</div><div class="summary-label">Models</div></div>
          <div class="summary-card"><div class="summary-value">0</div><div class="summary-label">Datasets</div></div>
          <div class="summary-card"><div class="summary-value">0</div><div class="summary-label">Domains</div></div>
          <div class="summary-card"><div class="summary-value">n/a</div><div class="summary-label">Last Refresh</div></div>
        </div>
        """

    models = df["model"].nunique()
    datasets = df["dataset"].nunique()
    domains = df["domain"].nunique() if "domain" in df.columns else 0
    last_refresh = leaderboard_refresh_timestamp(root_dir)

    return f"""
    <div class="summary-grid">
      <div class="summary-card"><div class="summary-value">{models}</div><div class="summary-label">Models</div></div>
      <div class="summary-card"><div class="summary-value">{datasets}</div><div class="summary-label">Datasets</div></div>
      <div class="summary-card"><div class="summary-value">{domains}</div><div class="summary-label">Domains</div></div>
      <div class="summary-card wide"><div class="summary-value small">{last_refresh}</div><div class="summary-label">Last Refresh</div></div>
    </div>
    """


def dataset_tab_label(dataset: str, root_dir: str = "results") -> str:
    raw_domain = lookup_dataset_domain(dataset, root_dir)
    domain = display_domain(raw_domain)
    short = dataset_short_name(dataset)
    return f"{domain} · {short}"


def dataset_section_title(dataset: str, root_dir: str, *, ranks: bool = False) -> str:
    label = dataset_tab_label(dataset, root_dir)
    if ranks:
        return f"**{label} · ranks (lower is better)**"
    return f"**{label} · metric values (absolute, lower is better)**"


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",
]

LIVE_RANK_HISTORY_COLUMNS = [
    "date",
    "model",
    "MSE_Rank",
    "RMSE_Rank",
    "MAPE_Rank",
    "CRPS_Rank",
    "RTG_Rank",
]

_LIVE_RANK_METRICS = {
    "MSE_Rank": ("MSE", True),
    "RMSE_Rank": ("RMSE", True),
    "MAPE_Rank": ("MAPE", True),
    "CRPS_Rank": ("CRPS", True),
    "RTG_Rank": ("RTG", False),
}

_RANK_COLS = [c for c in BASELINE_RANK_HISTORY_COLUMNS if c.endswith("_Rank")]
_LIVE_RANK_COLS = [c for c in LIVE_RANK_HISTORY_COLUMNS if c.endswith("_Rank")]
_DAILY_DISPLAY_COLS = ["date", "model"] + _RANK_COLS
_WEEKLY_DISPLAY_COLS = ["week", "model"] + _RANK_COLS
_MONTHLY_DISPLAY_COLS = ["month", "model"] + _RANK_COLS
_LIVE_DAILY_DISPLAY_COLS = ["date", "model"] + _LIVE_RANK_COLS
_LIVE_WEEKLY_DISPLAY_COLS = ["week", "model"] + _LIVE_RANK_COLS
_LIVE_MONTHLY_DISPLAY_COLS = ["month", "model"] + _LIVE_RANK_COLS


def _read_rank_history(path: Path, date_col: str) -> pd.DataFrame:
    """Read a legacy rank history CSV, coerce rank columns to float, sort by date_col."""
    if not path.exists():
        return pd.DataFrame(columns=[date_col, "model"] + _RANK_COLS)
    try:
        df = pd.read_csv(path)
    except Exception:
        return pd.DataFrame(columns=[date_col, "model"] + _RANK_COLS)
    for col in _RANK_COLS:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce").map(format_rank_number)
    sort_cols = [c for c in (date_col, "model") if c in df.columns]
    if sort_cols:
        df = df.sort_values(sort_cols)
    present = [c for c in [date_col, "model"] + _RANK_COLS if c in df.columns]
    return df[present].reset_index(drop=True)


def _live_release_metrics_path(root_dir: str) -> Path:
    return Path(root_dir) / "aggregates" / "live_release_metrics.csv"


def _read_live_release_metrics(root_dir: str = "results") -> pd.DataFrame:
    path = _live_release_metrics_path(root_dir)
    if not path.exists():
        return pd.DataFrame()
    try:
        df = pd.read_csv(path)
    except Exception:
        return pd.DataFrame()
    required = {"release_id", "model"}
    if not required.issubset(df.columns):
        return pd.DataFrame()
    time_col = "release_time" if "release_time" in df.columns else "evaluated_at"
    if time_col not in df.columns:
        return pd.DataFrame()
    df = df.copy()
    df["__timestamp"] = pd.to_datetime(df[time_col], errors="coerce", utc=True)
    df = df.dropna(subset=["__timestamp", "model", "release_id"])
    if df.empty:
        return pd.DataFrame()
    for metric_col, _ascending in _LIVE_RANK_METRICS.values():
        if metric_col in df.columns:
            df[metric_col] = pd.to_numeric(df[metric_col], errors="coerce")
    return df


def live_rank_history_model_choices(root_dir: str = "results") -> list[str]:
    df = _read_live_release_metrics(root_dir)
    if df.empty or "model" not in df.columns:
        return []
    return sorted({str(model).strip() for model in df["model"].dropna() if str(model).strip()})


def _aggregate_live_rank_history(df: pd.DataFrame, period_col: str) -> pd.DataFrame:
    rank_cols: list[str] = []
    work = df.copy()
    work["__models_per_release"] = work.groupby("release_id")["model"].transform("nunique")
    work = work[work["__models_per_release"] >= 2]
    if work.empty:
        return pd.DataFrame(columns=[period_col, "model"] + _LIVE_RANK_COLS)

    for rank_col, (metric_col, ascending) in _LIVE_RANK_METRICS.items():
        if metric_col not in work.columns:
            continue
        if work[metric_col].notna().sum() == 0:
            continue
        work[rank_col] = work.groupby("release_id")[metric_col].rank(
            method="average",
            ascending=ascending,
            na_option="keep",
        )
        rank_cols.append(rank_col)

    if not rank_cols:
        return pd.DataFrame(columns=[period_col, "model"] + _LIVE_RANK_COLS)

    grouped = (
        work.groupby([period_col, "model"], dropna=False)[rank_cols]
        .mean(numeric_only=True)
        .round(3)
        .reset_index()
    )
    for col in rank_cols:
        grouped[col] = grouped[col].map(format_rank_number)
    present = [c for c in [period_col, "model"] + _LIVE_RANK_COLS if c in grouped.columns]
    return grouped[present].sort_values([period_col, "model"]).reset_index(drop=True)


def load_live_model_rank_history(root_dir: str = "results", period: str = "date") -> pd.DataFrame:
    """Compute all-model rank history from live release metrics for date/week/month."""
    df = _read_live_release_metrics(root_dir)
    if df.empty:
        columns = {
            "date": _LIVE_DAILY_DISPLAY_COLS,
            "week": _LIVE_WEEKLY_DISPLAY_COLS,
            "month": _LIVE_MONTHLY_DISPLAY_COLS,
        }.get(period, _LIVE_DAILY_DISPLAY_COLS)
        return pd.DataFrame(columns=columns)

    if period == "week":
        iso = df["__timestamp"].dt.isocalendar()
        df["week"] = iso["year"].astype(str) + "-W" + iso["week"].astype(str).str.zfill(2)
        return _aggregate_live_rank_history(df, "week")
    if period == "month":
        df["month"] = df["__timestamp"].dt.strftime("%Y-%m")
        return _aggregate_live_rank_history(df, "month")
    df["date"] = df["__timestamp"].dt.strftime("%Y-%m-%d")
    return _aggregate_live_rank_history(df, "date")


def load_baseline_rank_history(root_dir: str = "results") -> pd.DataFrame:
    """Load all-model daily rank history from live releases, with legacy fallback."""
    live = load_live_model_rank_history(root_dir, "date")
    if not live.empty:
        return live
    for name in ("baseline_rank_history_daily.csv", "baseline_rank_history.csv"):
        path = Path(root_dir) / name
        if path.exists():
            return _read_rank_history(path, "date")
    return pd.DataFrame(columns=_DAILY_DISPLAY_COLS)


def load_baseline_rank_history_weekly(root_dir: str = "results") -> pd.DataFrame:
    """Load all-model weekly rank history from live releases, with legacy fallback."""
    live = load_live_model_rank_history(root_dir, "week")
    if not live.empty:
        return live
    return _read_rank_history(Path(root_dir) / "baseline_rank_history_weekly.csv", "week")


def load_baseline_rank_history_monthly(root_dir: str = "results") -> pd.DataFrame:
    """Load all-model monthly rank history from live releases, with legacy fallback."""
    live = load_live_model_rank_history(root_dir, "month")
    if not live.empty:
        return live
    daily_path = Path(root_dir) / "baseline_rank_history_daily.csv"
    if not daily_path.exists():
        daily_path = Path(root_dir) / "baseline_rank_history.csv"
    if not daily_path.exists():
        return pd.DataFrame(columns=_MONTHLY_DISPLAY_COLS)
    try:
        df = pd.read_csv(daily_path)
    except Exception:
        return pd.DataFrame(columns=_MONTHLY_DISPLAY_COLS)
    if "date" not in df.columns:
        return pd.DataFrame(columns=_MONTHLY_DISPLAY_COLS)

    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    df = df.dropna(subset=["date"])
    if df.empty:
        return pd.DataFrame(columns=_MONTHLY_DISPLAY_COLS)
    if "model" not in df.columns:
        df["model"] = ""
    df["month"] = df["date"].dt.strftime("%Y-%m")
    for col in _RANK_COLS:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce")

    present_rank_cols = [col for col in _RANK_COLS if col in df.columns]
    if present_rank_cols:
        monthly = (
            df.groupby(["month", "model"], dropna=False)[present_rank_cols]
            .mean(numeric_only=True)
            .round(3)
            .reset_index()
        )
    else:
        monthly = df.loc[:, ["month", "model"]].drop_duplicates()
    for col in present_rank_cols:
        monthly[col] = monthly[col].map(format_rank_number)
    present = [c for c in _MONTHLY_DISPLAY_COLS if c in monthly.columns]
    return monthly[present].sort_values(["month", "model"]).reset_index(drop=True)


def load_fixed_window_table(
    root_dir: str = "results",
    window: str = "1d",
    table: str = "rank",
) -> pd.DataFrame:
    """Load one full trailing-window ranking or metric table."""

    if window not in {"1d", "7d", "30d"}:
        raise ValueError(f"Unsupported fixed leaderboard window: {window}")
    if table not in {"rank", "overall"}:
        raise ValueError(f"Unsupported fixed leaderboard table: {table}")
    path = Path(root_dir) / "aggregates" / f"live_{table}_{window}.csv"
    if not path.exists():
        return pd.DataFrame()
    try:
        frame = pd.read_csv(path)
    except Exception:
        return pd.DataFrame()
    if table == "rank":
        columns = [
            "Rank",
            "Status",
            "Model",
            "Win Rate",
            "Eligible Opponents",
            "Shared Releases",
            "Shared Datasets",
            "Covered Span (days)",
        ]
        required = {
            "Rank",
            "Status",
            "model",
            "WinRate",
            "eligible_opponents",
            "shared_releases",
            "shared_datasets",
            "covered_span_days",
        }
        if not required.issubset(frame.columns):
            return pd.DataFrame(columns=columns)
        frame = frame.loc[:, list(required)].rename(
            columns={
                "model": "Model",
                "WinRate": "Win Rate",
                "eligible_opponents": "Eligible Opponents",
                "shared_releases": "Shared Releases",
                "shared_datasets": "Shared Datasets",
                "covered_span_days": "Covered Span (days)",
            }
        )
        frame["Rank"] = pd.to_numeric(frame["Rank"], errors="coerce").map(format_rank_number)
        frame["Win Rate"] = pd.to_numeric(frame["Win Rate"], errors="coerce").map(
            lambda value: "n/a" if pd.isna(value) else f"{100.0 * float(value):.2f}%"
        )
        frame["Covered Span (days)"] = pd.to_numeric(
            frame["Covered Span (days)"], errors="coerce"
        ).map(format_number)
        return frame.loc[:, columns].reset_index(drop=True)

    columns = [
        "Model",
        "MSE",
        "RMSE",
        "MAPE",
        "CRPS",
        "RTG",
        "Stability",
        "Improvement",
        "Datasets",
        "Releases",
    ]
    required = {
        "model",
        "MSE",
        "RMSE",
        "MAPE",
        "CRPS",
        "RTG",
        "Stability",
        "Improvement",
        "n_datasets",
        "n_releases",
    }
    if not required.issubset(frame.columns):
        return pd.DataFrame(columns=columns)
    frame = frame.loc[:, list(required)].rename(
        columns={"model": "Model", "n_datasets": "Datasets", "n_releases": "Releases"}
    )
    for metric in ("MSE", "RMSE", "MAPE", "CRPS", "RTG", "Stability", "Improvement"):
        frame[metric] = pd.to_numeric(frame[metric], errors="coerce").map(format_number)
    return frame.loc[:, columns].reset_index(drop=True)


def load_fixed_window_rank(root_dir: str = "results", window: str = "1d") -> pd.DataFrame:
    """Backward-compatible wrapper for the full fixed-window ranking."""

    return load_fixed_window_table(root_dir, window, "rank")

# ---------------------------------------------------------------------------
# Forecast snapshot utilities
# ---------------------------------------------------------------------------

def load_forecast_snapshots(root_dir: str, model_slug: str) -> dict[str, dict]:
    """Load all forecast JSON files for a given model slug.

    Returns a dict keyed by the dataset display name (from snapshot["dataset"]).
    """
    forecasts_dir = Path(root_dir) / model_slug / "forecasts"
    snapshots: dict[str, dict] = {}
    if not forecasts_dir.exists():
        return snapshots
    for fp in sorted(forecasts_dir.glob("*.json")):
        try:
            snap = json.loads(fp.read_text())
            snapshots[snap.get("dataset", fp.stem)] = snap
        except Exception:
            pass
    return snapshots


def _get_plotly():
    """Lazy import plotly to avoid import errors in non-chart contexts."""
    try:
        import plotly.graph_objects as go
        return go
    except ImportError:
        return None


def make_forecast_plot(snapshot: dict | None) -> object | None:
    """Build a plotly Figure showing context + actuals vs p50 forecast with p10/p90 band.

    Returns None if snapshot is None or plotly is unavailable.
    """
    go = _get_plotly()
    if go is None or snapshot is None:
        return None

    ctx = snapshot.get("context", [])
    actuals = snapshot.get("actuals", [])
    p50 = snapshot.get("p50", [])
    p10 = snapshot.get("p10", [])
    p90 = snapshot.get("p90", [])
    n_ctx = len(ctx)
    n_fut = len(actuals)

    timestamps = snapshot.get("timestamps", [])
    if timestamps and len(timestamps) == n_ctx + n_fut:
        ctx_x = timestamps[:n_ctx]
        fut_x = timestamps[n_ctx:]
        x_title = "Time"
        vline_x = ctx_x[-1] if n_ctx > 0 else fut_x[0]
    else:
        ctx_x = list(range(n_ctx))
        fut_x = list(range(n_ctx, n_ctx + n_fut))
        x_title = "Time step"
        vline_x = n_ctx - 1.0 if n_ctx > 0 else 0


    if n_ctx > 0:
        fut_x = [ctx_x[-1]] + list(fut_x)
        actuals = [ctx[-1]] + list(actuals)
        p50 = [ctx[-1]] + list(p50) if p50 else []
        p10 = [ctx[-1]] + list(p10) if p10 else []
        p90 = [ctx[-1]] + list(p90) if p90 else []
        n_fut = len(actuals)

    fig = go.Figure()


    # Context (historical)
    fig.add_trace(go.Scatter(
        x=ctx_x, y=ctx,
        mode="lines",
        name="Context (history)",
        line=dict(color="#94a3b8", width=1.5),
    ))

    # Actuals (ground truth for forecast horizon)
    fig.add_trace(go.Scatter(
        x=fut_x, y=actuals,
        mode="lines",
        name="Actual",
        line=dict(color="#3b82f6", width=2),
    ))

    # p10-p90 confidence band
    if p10 and p90 and len(p10) == n_fut and len(p90) == n_fut:
        fig.add_trace(go.Scatter(
            x=fut_x + fut_x[::-1],
            y=p90 + p10[::-1],
            fill="toself",
            fillcolor="rgba(239,68,68,0.15)",
            line=dict(color="rgba(255,255,255,0)"),
            name="p10-p90 band",
            showlegend=True,
        ))

    # p50 forecast
    if p50 and len(p50) == n_fut:
        fig.add_trace(go.Scatter(
            x=fut_x, y=p50,
            mode="lines",
            name="Forecast (p50)",
            line=dict(color="#ef4444", width=2, dash="dash"),
        ))

    dataset_label = snapshot.get("dataset", "").split("/")[0]
    model_label = snapshot.get("model", "")
    evaluated_at = snapshot.get("evaluated_at", "")[:10]
    fig.update_layout(
        title=dict(
            text=f"{dataset_label} — {model_label} ({evaluated_at})",
            font=dict(size=14),
        ),
        xaxis_title=x_title,
        yaxis_title="Value",
        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
        margin=dict(l=40, r=20, t=60, b=40),
        height=350,
        paper_bgcolor="rgba(0,0,0,0)",
        plot_bgcolor="rgba(0,0,0,0)",
        font=dict(color="black"),
        xaxis=dict(gridcolor="rgba(0,0,0,0.08)"),
        yaxis=dict(gridcolor="rgba(0,0,0,0.08)"),
    )
    # Vertical line separating context from forecast
    fig.add_vline(
        x=vline_x,
        line_dash="dot",
        line_color="rgba(0,0,0,0.3)",
        annotation_text="forecast start",
        annotation_font_color="black",
    )
    return fig


def make_rank_trend_plot(history_df: pd.DataFrame, metrics: list[str] | None = None) -> object | None:
    """Build a plotly Figure showing selected rank metrics over time (date on X axis).

    Args:
        history_df: DataFrame from load_baseline_rank_history() with a 'date' column.
        metrics: List of rank column names to plot. Defaults to MSE_Rank and CRPS_Rank.

    Returns None if fewer than 1 data point or plotly unavailable.
    """
    go = _get_plotly()
    if go is None or history_df is None or history_df.empty:
        return None

    if metrics is None:
        metrics = ["MSE_Rank", "CRPS_Rank", "RMSE_Rank", "MAPE_Rank"]

    date_col = "date" if "date" in history_df.columns else "week"
    present_metrics = [m for m in metrics if m in history_df.columns]
    if not present_metrics:
        return None

    COLORS = ["#f97316", "#a78bfa", "#34d399", "#60a5fa", "#fb7185", "#fbbf24"]

    fig = go.Figure()
    for i, col in enumerate(present_metrics):
        fig.add_trace(go.Scatter(
            x=history_df[date_col],
            y=history_df[col],
            mode="lines+markers",
            name=col.replace("_Rank", ""),
            line=dict(color=COLORS[i % len(COLORS)], width=2),
            marker=dict(size=6),
        ))

    fig.update_layout(
        title=dict(text="Selected model rank trend over time (lower is better)", font=dict(size=14)),
        xaxis_title=date_col.capitalize(),
        yaxis_title="Average rank",
        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
        margin=dict(l=40, r=20, t=60, b=40),
        height=320,
        paper_bgcolor="rgba(0,0,0,0)",
        plot_bgcolor="rgba(0,0,0,0)",
        font=dict(color="black"),
        xaxis=dict(gridcolor="rgba(0,0,0,0.08)"),
        yaxis=dict(gridcolor="rgba(0,0,0,0.08)"),
    )
    return fig


DOMAIN_COLORS = {
    "Finance": "#f97316",
    "Climate": "#60a5fa",
    "Ocean": "#34d399",
    "Hydrology": "#818cf8",
    "Transportation": "#14b8a6",
    "Hazards": "#fb7185",
    "Web": "#a78bfa",
}


def _today_start_utc8() -> datetime:
    """Return midnight today in UTC+8 as an aware datetime."""
    utc8 = timezone(timedelta(hours=8))
    return datetime.now(utc8).replace(hour=0, minute=0, second=0, microsecond=0)


def get_dataset_last_eval_time(ds_base: str, root_dir: str = "results") -> datetime | None:
    latest_dt: datetime | None = None

    history_path = Path(root_dir) / "eval_history.jsonl"
    if history_path.exists():
        try:
            with history_path.open(encoding="utf-8") as handle:
                for line in handle:
                    if not line.strip():
                        continue
                    entry = json.loads(line)
                    if _dataset_key(entry.get("dataset", "")) != _dataset_key(ds_base):
                        continue
                    dt = _parse_timestamp(entry.get("evaluated_at"))
                    if dt is not None and (latest_dt is None or dt > latest_dt):
                        latest_dt = dt
        except Exception:
            pass

    for model_dir in Path(root_dir).iterdir():
        if not model_dir.is_dir():
            continue
        meta_path = model_dir / "online_meta.json"
        if not meta_path.exists():
            continue
        try:
            meta = json.loads(meta_path.read_text())
        except Exception:
            continue
        matched = False
        for item in meta.get("datasets", []):
            raw_name = str(item.get("dataset", ""))
            if _dataset_key(raw_name) == _dataset_key(ds_base) or ds_base.lower() in raw_name.lower():
                matched = True
                break
        if not matched:
            continue
        dt = _parse_timestamp(meta.get("evaluated_at"))
        if dt is not None and (latest_dt is None or dt > latest_dt):
            latest_dt = dt
    return latest_dt


def make_domain_pie_chart(root_dir: str = "results") -> object | None:
    """Build a plotly Pie chart showing evaluated timestamp count per domain today.

    Reads eval_history.jsonl to count total evaluated timestamps since UTC+8 00:00 today,
    grouped by domain.
    """
    go = _get_plotly()
    if go is None:
        return None

    # Domain mapping from dataset_properties.csv
    props_path = Path(root_dir) / "dataset_properties.csv"
    if not props_path.exists():
        return None
    try:
        props = pd.read_csv(props_path)
    except Exception:
        return None
    domain_map: dict[str, str] = {}
    for _, row in props.iterrows():
        domain_map[str(row["dataset"]).lower()] = display_domain(str(row["domain"]))

    def get_domain(ds_name: str) -> str:
        ds_lower = ds_name.lower()
        if ds_lower not in domain_map:
            domain_map[ds_lower] = display_domain(lookup_dataset_domain(ds_name, root_dir))
        return domain_map[ds_lower]

    today_start = _today_start_utc8()
    utc8 = today_start.tzinfo

    # Read eval_history.jsonl (Cumulative)
    history_path = Path(root_dir) / "eval_history.jsonl"
    timestamp_counts: dict[str, int] = {}  # domain → total timestamps
    domain_datasets: dict[str, set[str]] = {}  # domain → set of dataset short names

    if history_path.exists():
        try:
            with open(history_path) as f:
                for line in f:
                    if not line.strip():
                        continue
                    entry = json.loads(line)
                    ds_base = entry.get("dataset", "")
                    pred_len = entry.get("prediction_length", 0)
                    domain = get_domain(ds_base)
                    timestamp_counts[domain] = timestamp_counts.get(domain, 0) + pred_len
                    domain_datasets.setdefault(domain, set()).add(dataset_short_name(ds_base))
        except Exception:
            pass

    # Fallback to one count per current dataset if history is unavailable.
    if not timestamp_counts:
        for ds_base, domain in domain_map.items():
            timestamp_counts[domain] = timestamp_counts.get(domain, 0) + 1
            domain_datasets.setdefault(domain, set()).add(dataset_short_name(ds_base))

    if not timestamp_counts:
        return None

    labels = list(timestamp_counts.keys())
    values = list(timestamp_counts.values())
    hover = [", ".join(sorted(domain_datasets.get(d, []))) for d in labels]
    colors = [DOMAIN_COLORS.get(label, "#94a3b8") for label in labels]

    fig = go.Figure(data=[go.Pie(
        labels=labels,
        values=values,
        hovertext=hover,
        hoverinfo="label+percent+text",
        textinfo="label+value+percent",
        textposition="inside",
        textfont=dict(size=14, color="white"),
        insidetextorientation="radial",
        marker=dict(colors=colors),
        hole=0.25,
    )])
    fig.update_layout(
        title=dict(
            text="Evaluated timestamps by domain",
            font=dict(size=16, weight="bold"),
        ),
        margin=dict(l=20, r=20, t=60, b=20),
        height=500,
        paper_bgcolor="rgba(0,0,0,0)",
        plot_bgcolor="rgba(0,0,0,0)",
        font=dict(color="black"),
        showlegend=True,
        legend=dict(
            orientation="h",
            yanchor="bottom",
            y=-0.1,
            xanchor="center",
            x=0.5
        )
    )
    return fig


# ---------------------------------------------------------------------------
# Per-domain status helpers
# ---------------------------------------------------------------------------

def load_eval_intervals(root_dir: str = "results") -> dict[str, int]:
    """Load per-dataset eval intervals from dataset_eval_intervals.json."""
    path = Path(root_dir) / "dataset_eval_intervals.json"
    if not path.exists():
        return {}
    try:
        return {str(k): int(v) for k, v in json.loads(path.read_text()).items()}
    except Exception:
        return {}


def _format_interval(seconds: int) -> str:
    if seconds >= 86400:
        d = seconds // 86400
        return f"{d}d"
    if seconds >= 3600:
        h = seconds // 3600
        return f"{h}h"
    if seconds >= 60:
        m = seconds // 60
        return f"{m}min"
    return f"{seconds}s"


def build_domain_status_md(
    domain: str,
    datasets: list[str],
    root_dir: str = "results",
) -> str:
    """Build a compact markdown status line for a domain tab.

    Shows data source, last eval time, last evaluated time for this domain,
    and per-dataset refresh intervals.
    """
    utc8 = timezone(timedelta(hours=8))

    # 1. Load data source name from online_status.json
    status_path = Path(root_dir) / "online_status.json"
    data_source = "TS-Bench"
    if status_path.exists():
        try:
            status = json.loads(status_path.read_text())
            data_source = status.get("data_source", "TS-Bench")
        except Exception:
            pass

    # 2. Last evaluated time for this domain (max of all datasets in the domain)
    domain_latest_dt: datetime | None = None
    for ds in datasets:
        ds_base = ds.split("/")[0].strip()
        eval_dt = get_dataset_last_eval_time(ds_base, root_dir)
        if eval_dt:
            if domain_latest_dt is None or eval_dt > domain_latest_dt:
                domain_latest_dt = eval_dt

    last_evaluated = "n/a"
    if domain_latest_dt:
        last_evaluated = domain_latest_dt.astimezone(utc8).strftime("%Y-%m-%d %H:%M:%S") + " (UTC+8)"

    # 3. Auto-refresh per-dataset
    intervals = load_eval_intervals(root_dir)
    ds_parts: list[str] = []
    for ds in datasets:
        short = dataset_short_name(ds)
        ds_base = ds.split("/")[0].strip()
        interval = intervals.get(ds_base)
        if interval:
            ds_parts.append(f"{short} every {_format_interval(interval)}")
        else:
            ds_parts.append(short)

    parts = [
        f"**Data**: {data_source}",
        f"**Last eval**: {last_evaluated}",
    ]
    if ds_parts:
        parts.append("**Refresh**: " + " · ".join(ds_parts))

    return " | ".join(parts)