src/trackrecord.py

"""Archiving, resolution and standings.

The pipeline is three steps, and the boundary between them is the point.

1. `archive` writes a forecast at the moment it is issued. Those rows are
   final. Nothing downstream ever rewrites them.
2. `resolve` looks for archived forecasts whose horizon has now elapsed,
   compares them against realised prices, and *appends* the outcome to a
   separate tree.
3. `build_standings` derives the leaderboard from the resolved outcomes and
   nothing else, so it is a pure function of the track record and can be
   regenerated from scratch at any time.

Keeping resolution out of the archive is what makes the record evidence rather
than marketing. If a bad forecast could be quietly corrected once the outcome
was known, the calibration numbers would measure nothing at all.
"""

from __future__ import annotations

import logging

import numpy as np
import pandas as pd

from . import config
from .adapters import ForecastResult
from .store import (FORECAST_COLUMNS, QUANTILE_COLUMNS, STANDINGS_COLUMNS,
                    TRACKRECORD_COLUMNS, ArenaStore, forecast_id, now_utc)

log = logging.getLogger("arena.trackrecord")


def _as_utc(ts) -> pd.Timestamp:
    """Normalise a caller-supplied "now" to tz-aware UTC."""
    if ts is None:
        return now_utc()
    t = pd.Timestamp(ts)
    return t.tz_localize("UTC") if t.tzinfo is None else t.tz_convert("UTC")


# --------------------------------------------------------------------------
# Archive
# --------------------------------------------------------------------------


def forecast_rows(result: ForecastResult, model_slug: str, asset: str,
                  timeframe: str, issued_ts, target_ts,
                  backfilled: bool = False, path_ref: str | None = None) -> pd.DataFrame:
    """Turn an adapter result into archive rows -- one per forecast step."""
    if len(target_ts) != result.horizon:
        raise ValueError(
            f"{len(target_ts)} target timestamps for a horizon of {result.horizon}")

    fid = forecast_id(model_slug, asset, timeframe, issued_ts, result.seed)
    dispersion = result.dispersion()

    data = {
        "forecast_id": fid,
        "model_slug": model_slug,
        "asset": asset,
        "timeframe": timeframe,
        "issued_ts": pd.to_datetime([issued_ts] * result.horizon, utc=True),
        "target_ts": pd.to_datetime(list(target_ts), utc=True),
        "step": np.arange(1, result.horizon + 1, dtype="int32"),
        "horizon_bars": np.int32(result.horizon),
        "seed": np.int64(result.seed),
        "n_samples": np.int32(result.n_samples),
        "context_len": np.int32(result.context_len),
        "inference_version": result.inference_version,
        "dispersion": dispersion.astype("float64"),
        # A reference, not the paths themselves: whole sampled paths are far
        # too heavy to inline into every archive row.
        "path_ref": path_ref or "",
        "backfilled": bool(backfilled),
    }
    for i, level in enumerate(result.levels):
        data[f"q{int(round(level * 100)):02d}"] = result.quantiles[:, i]

    return pd.DataFrame(data)[FORECAST_COLUMNS]


def archive(store: ArenaStore, result: ForecastResult, model_slug: str,
            asset: str, timeframe: str, issued_ts, target_ts,
            backfilled: bool = False) -> tuple[str, int]:
    """Archive a forecast. Returns (forecast_id, rows written; 0 if a duplicate).

    Also promotes it to the latest-forecast cache when it is the newest for
    this series, so a visitor who has pressed nothing still sees each model's
    most recent work. The promotion runs even for a duplicate: the archive row
    already existed, but the cache may not have been written yet.
    """
    rows = forecast_rows(result, model_slug, asset, timeframe,
                         issued_ts, target_ts, backfilled=backfilled)
    written = store.append_forecast(rows)

    if store.update_latest(rows) and result.paths is not None:
        # Only a handful of paths, evenly spread through the samples: enough
        # to redraw the ghosts, far short of the full sample set.
        store.put_latest_paths(model_slug, asset, timeframe,
                               _thin_paths(result.paths))

    return str(rows["forecast_id"].iloc[0]), written


def _thin_paths(paths: np.ndarray, keep: int = config.GHOST_PATHS) -> np.ndarray:
    """Evenly spaced sample paths, so the survivors represent the spread."""
    take = min(keep, paths.shape[0])
    picks = np.linspace(0, paths.shape[0] - 1, take).round().astype(int)
    return paths[picks]


# --------------------------------------------------------------------------
# Resolve
# --------------------------------------------------------------------------


def resolve(store: ArenaStore, model_slug: str, asset: str, timeframe: str,
            now=None) -> pd.DataFrame:
    """Resolve every archived forecast whose target bar has printed.

    Idempotent: rows already in the track record are skipped, so a re-run adds
    nothing. It never touches the archive.
    """
    now = _as_utc(now)

    forecasts = store.get_forecasts(model_slug, asset, timeframe)
    if not len(forecasts):
        return pd.DataFrame(columns=TRACKRECORD_COLUMNS)

    already = store.get_trackrecord(model_slug, asset, timeframe)
    seen = set(zip(already["forecast_id"].astype(str), already["step"].astype(int))) \
        if len(already) else set()

    due = forecasts[forecasts["target_ts"] <= now]
    if len(due):
        mask = [(str(f), int(s)) not in seen
                for f, s in zip(due["forecast_id"], due["step"])]
        due = due[np.asarray(mask)]
    if not len(due):
        return pd.DataFrame(columns=TRACKRECORD_COLUMNS)

    prices = store.get_prices(asset, timeframe,
                              start=due["target_ts"].min(),
                              end=due["target_ts"].max())
    if not len(prices):
        log.info("no prices to resolve %s/%s/%s against", model_slug, asset, timeframe)
        return pd.DataFrame(columns=TRACKRECORD_COLUMNS)

    realized = prices.set_index("ts")["close"]
    joined = due.copy()
    joined["realized_close"] = joined["target_ts"].map(realized)
    # A target bar that has not printed -- a market holiday, a gap in the cache
    # -- is left unresolved rather than filled with the nearest neighbour.
    # Resolving against a price from the wrong bar would be a silent lie.
    joined = joined[joined["realized_close"].notna()]
    if not len(joined):
        return pd.DataFrame(columns=TRACKRECORD_COLUMNS)

    out = pd.DataFrame({
        "forecast_id": joined["forecast_id"].astype(str),
        "model_slug": model_slug,
        "asset": asset,
        "timeframe": timeframe,
        "issued_ts": joined["issued_ts"],
        "target_ts": joined["target_ts"],
        "step": joined["step"].astype("int32"),
        "horizon_bars": joined["horizon_bars"].astype("int32"),
        "realized_close": joined["realized_close"].astype("float64"),
        "q10": joined["q10"].astype("float64"),
        "q20": joined["q20"].astype("float64"),
        "q50": joined["q50"].astype("float64"),
        "q80": joined["q80"].astype("float64"),
        "q90": joined["q90"].astype("float64"),
        "resolved_ts": now,
        "backfilled": joined["backfilled"].astype(bool),
    })
    out["inside_80"] = ((out["realized_close"] >= out["q10"]) &
                        (out["realized_close"] <= out["q90"]))
    out["inside_60"] = ((out["realized_close"] >= out["q20"]) &
                        (out["realized_close"] <= out["q80"]))
    out["abs_error"] = (out["realized_close"] - out["q50"]).abs()
    out["pct_error"] = out["abs_error"] / out["realized_close"].abs().replace(0, np.nan) * 100.0

    out = out[TRACKRECORD_COLUMNS]
    store.append_trackrecord(model_slug, asset, timeframe, out)
    return out


# --------------------------------------------------------------------------
# Standings
# --------------------------------------------------------------------------


def grade_for(gap: float, resolved_count: int) -> str:
    """Map a calibration gap to a letter, or to '-' when there is too little.

    Refusing to grade below `MIN_RESOLVED_FOR_GRADE` is deliberate: with a
    handful of observations the coverage estimate is mostly noise, and a
    confident 'A' computed from four forecasts would be the single most
    misleading thing on the page.
    """
    if resolved_count < config.MIN_RESOLVED_FOR_GRADE:
        return "-"
    if gap is None or not np.isfinite(gap):
        return "-"
    for letter, threshold in config.GRADE_THRESHOLDS:
        if abs(gap) <= threshold:
            return letter
    return config.GRADE_THRESHOLDS[-1][0]


def build_standings(trackrecord: pd.DataFrame, now=None) -> pd.DataFrame:
    """Derive the standings table. A pure function of the track record.

    Grouped by (model, asset_class) rather than by asset: a model's calibration
    on crypto and on equities are different claims, but BTC and ETH are not
    independent enough for separate grades to mean much.
    """
    now = _as_utc(now)

    if trackrecord is None or not len(trackrecord):
        return pd.DataFrame(columns=STANDINGS_COLUMNS)

    df = trackrecord.copy()
    df["asset_class"] = df["asset"].map(
        lambda a: config.ASSETS[a].asset_class if a in config.ASSETS else "other")

    rows = []
    for (model_slug, asset_class), group in df.groupby(["model_slug", "asset_class"],
                                                       sort=True):
        resolved = len(group)
        coverage_80 = float(group["inside_80"].mean())
        coverage_60 = float(group["inside_60"].mean())
        median_err = float(group["pct_error"].median())
        gap = coverage_80 - config.NOMINAL_COVERAGE
        rows.append({
            "model_slug": model_slug,
            "asset_class": asset_class,
            "resolved_count": int(resolved),
            "forecast_count": int(group["forecast_id"].nunique()),
            "coverage_80": coverage_80,
            "coverage_60": coverage_60,
            "median_abs_pct_error": median_err,
            "calibration_gap": float(gap),
            "grade": grade_for(gap, resolved),
            "updated_ts": now,
        })

    out = pd.DataFrame(rows, columns=STANDINGS_COLUMNS)
    # Deterministic order, so that "same inputs produce the same standings" is
    # true of the bytes and not merely of the contents.
    return out.sort_values(["model_slug", "asset_class"]).reset_index(drop=True)


def regenerate_standings(store: ArenaStore, now=None,
                         force: bool = False) -> pd.DataFrame:
    """Rebuild the standings *and* the panel summaries from the track record.

    `force` overrides the shrink guard on both artefacts. Only pass it when a
    smaller table is genuinely correct -- a model retired, say -- and never to
    make a failing rebuild go away.
    """
    tr = store.all_trackrecords()
    standings = build_standings(tr, now=now)
    store.put_standings(standings, force=force)
    store.put_panels(build_panels(tr, standings, now=now), force=force)
    return standings


# --------------------------------------------------------------------------
# Panel summaries
# --------------------------------------------------------------------------

# How many resolved forecasts each panel keeps a thumbnail for.
PANEL_THUMBS = 10


def build_panels(trackrecord: pd.DataFrame, standings: pd.DataFrame,
                 now=None) -> dict:
    """Precompute what the Track Record panel needs, per (model, asset, tf).

    The Space cannot read the raw track record: it is 7 MB across 80 files and
    growing, and pulling it at boot would put that on every cold start. This
    collapses it to the numbers and the ten thumbnails the panel actually
    draws -- a few hundred kilobytes, written once by the resolver.

    The sparkline geometry is computed here rather than in the renderer for the
    same reason the rest of it is: so the Space does no arithmetic over the
    archive to draw a page.
    """
    now = _as_utc(now)
    doc = {"version": 1, "updated_ts": now.isoformat(), "panels": {}, "totals": {}}

    if trackrecord is None or not len(trackrecord):
        doc["totals"] = {"archived_forecasts": 0, "resolved_forecasts": 0,
                         "resolved_rows": 0, "resolver_last_ran": None}
        return doc

    tr = trackrecord
    doc["totals"] = {
        "resolved_forecasts": int(tr["forecast_id"].nunique()),
        "resolved_rows": int(len(tr)),
        "archived_forecasts": int(tr["forecast_id"].nunique()),
        "resolver_last_ran": (pd.to_datetime(tr["resolved_ts"], utc=True).max()
                              .strftime("%Y-%m-%d %H:%M UTC")
                              if "resolved_ts" in tr else None),
    }

    for (model_slug, asset, timeframe), rows in tr.groupby(
            ["model_slug", "asset", "timeframe"], sort=True):
        resolved = int(len(rows))
        coverage = float(rows["inside_80"].mean())
        gap = coverage - config.NOMINAL_COVERAGE
        realized = rows["realized_close"].abs().replace(0, np.nan)
        doc["panels"][f"{model_slug}|{asset}|{timeframe}"] = {
            "resolved": resolved,
            "forecasts": int(rows["forecast_id"].nunique()),
            "coverage_80": coverage,
            "coverage_60": float(rows["inside_60"].mean()),
            "median_pct_error": float(rows["pct_error"].median()),
            "mean_width": float(((rows["q90"] - rows["q10"]).abs()
                                 / realized).median()),
            "backfilled_share": float(rows["backfilled"].mean()),
            "grade": grade_for(gap, resolved),
            "calibration_gap": float(gap),
            "thumbs": _thumbs(rows),
        }
    return doc


def _thumbs(rows: pd.DataFrame) -> list[dict]:
    """The last `PANEL_THUMBS` resolved forecasts, with their sparklines."""
    out = []
    groups = sorted(rows.groupby("forecast_id"),
                    key=lambda kv: kv[1]["issued_ts"].max(), reverse=True)
    for _fid, group in groups[:PANEL_THUMBS]:
        group = group.sort_values("step")
        hit = bool(group["inside_80"].mean() >= 0.5)
        out.append({
            "date": pd.Timestamp(group["issued_ts"].iloc[0]).strftime("%m-%d %H:%M"),
            "hit": hit,
            "err": float(group["pct_error"].median()),
            "backfilled": bool(group["backfilled"].any()),
            "spark": _spark_paths(group),
        })
    return out


def _spark_paths(group: pd.DataFrame) -> dict:
    """Band, median and realised path as SVG `d` strings in a 94x38 box."""
    n = len(group)
    if n < 2:
        return {"band": "", "mid": "", "real": ""}
    lo = group["q10"].to_numpy(dtype="float64")
    hi = group["q90"].to_numpy(dtype="float64")
    mid = group["q50"].to_numpy(dtype="float64")
    real = group["realized_close"].to_numpy(dtype="float64")

    floor = float(min(lo.min(), real.min()))
    ceil = float(max(hi.max(), real.max()))
    span = (ceil - floor) or 1.0

    def xy(i, v):
        return f"{6 + i * (82 / max(1, n - 1)):.1f} {35 - (v - floor) / span * 32:.1f}"

    upper = [xy(i, hi[i]) for i in range(n)]
    lower = [xy(i, lo[i]) for i in range(n)]
    return {
        "band": "M" + " L".join(upper) + " L" + " L".join(reversed(lower)) + " Z",
        "mid": "M" + " L".join(xy(i, mid[i]) for i in range(n)),
        "real": "M" + " L".join(xy(i, real[i]) for i in range(n)),
    }