"""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)), }