"""State, and the derivations the UI renders from. The whole UI is a pure function of the dict `default_state()` returns. Actions fold that dict forward; the renderer turns it into markup. Nothing in the view layer reaches for a model, a store or a clock, which is what lets the entire interface be tested without a browser or a network. Numbers that do not exist render as an em dash, never as `0.00`. A model with no resolved forecasts has no coverage -- saying `0%` would claim it was wrong every time, which is the opposite of the truth. """ from __future__ import annotations import numpy as np import pandas as pd from . import config from .store import now_utc from .trackrecord import grade_for DASH = "—" MAX_MATCHUP = 3 LOG_LIMIT = 8 THUMBS = 10 def default_state() -> dict: return { "mode": "playground", "selected": [], # filled at boot from the registry "asset": "BTC-USD", "tf": "1h", "horizon": config.DEFAULT_HORIZON["1h"], "ghosts": True, "gear": False, "cls": "All", "seed": 42, "n_samples": config.DEFAULT_N_SAMPLES, "hfid": "", "family": "kronos", "enroll_note": "", "enroll_ok": None, "log": [], "error": None, "session_forecasts": 0, "signed_in": False, "user": "", "backfill_requested": False, "backfill_note": "", "nonce": "", } # -------------------------------------------------------------------------- # Formatting # -------------------------------------------------------------------------- def pct(value, digits: int = 0) -> str: if value is None or (isinstance(value, float) and not np.isfinite(value)): return DASH return f"{value * 100:.{digits}f}%" def num(value, digits: int = 2) -> str: if value is None or (isinstance(value, float) and not np.isfinite(value)): return DASH return f"{value:,.{digits}f}" def count(value) -> str: if value is None: return DASH return f"{int(value):,}" def age_label(when, now=None) -> str: """How long ago something happened, in the coarsest unit that is honest. A forecast pulled from the archive has to say how old it is. "12:00 UTC" alone reads as current; "issued 3h ago" is the same fact and cannot be misread as fresh. """ if when is None: return DASH when = pd.Timestamp(when) when = when.tz_localize("UTC") if when.tzinfo is None else when.tz_convert("UTC") now = pd.Timestamp(now) if now is not None else pd.Timestamp.utcnow() now = now.tz_localize("UTC") if now.tzinfo is None else now.tz_convert("UTC") seconds = (now - when).total_seconds() if seconds < 0: return "just now" if seconds < 90: return "just now" minutes = seconds / 60 if minutes < 90: return f"{int(round(minutes))}m ago" hours = minutes / 60 if hours < 36: return f"{int(round(hours))}h ago" return f"{int(round(hours / 24))}d ago" def live_scorecard(run) -> dict | None: """How an archived forecast is doing against what actually printed. This is the Arena's whole claim in one line, applied to the forecast on screen: of the bars that have printed so far, how many landed inside the 80% band, and how far the median is off. Returns None when nothing has printed yet, because "0 of 0" is not a result. Deliberately scored only on bars that exist. A horizon still running is partial evidence, and the label says how partial. """ realized = getattr(run, "realized", None) if realized is None or not len(realized): return None steps = {pd.Timestamp(t): i for i, t in enumerate(run.target_ts)} lo, hi = run.result.band() med = run.result.median() inside = 0 errors = [] matched = 0 for ts, close in zip(pd.to_datetime(realized["ts"], utc=True), realized["close"].to_numpy(dtype="float64")): step = steps.get(pd.Timestamp(ts)) if step is None or step >= len(med): continue matched += 1 if lo[step] <= close <= hi[step]: inside += 1 errors.append(abs(close - med[step]) / abs(close) * 100 if close else np.nan) if not matched: return None share = inside / matched complete = matched >= run.horizon if share >= 0.8: colour = "var(--accent-moss-strong)" elif share >= 0.6: colour = "var(--accent-amber-strong)" else: colour = "var(--mute-orange)" return { "inside": inside, "matched": matched, "share": share, "median_err": float(np.nanmedian(errors)) if errors else None, "complete": complete, "colour": colour, "label": f"{inside}/{matched} IN BAND", "note": ("horizon complete" if complete else f"{matched} of {run.horizon} bars in so far"), } def grade_color(grade: str) -> str: head = (grade or "")[:1] if head == "A": return "var(--accent-moss-strong)" if head == "B": return "var(--accent-amber-strong)" if head in ("C", "D"): return "var(--mute-orange)" return "var(--text-tertiary)" def verdict_for(coverage, resolved: int) -> tuple[str, str, str]: """(chip, short, colour) describing a model's calibration in words. The wording is deliberately unflattering where the data is unflattering: a model whose bands are too narrow is called overconfident on the page, not in a footnote. """ if resolved < config.MIN_RESOLVED_FOR_GRADE or coverage is None: return ("NOT YET GRADED", "too few resolved", "var(--text-tertiary)") gap = coverage - config.NOMINAL_COVERAGE if abs(gap) <= 0.05: return ("WELL CALIBRATED", "bands match reality", "var(--accent-moss-strong)") if gap < -0.15: return ("OVERCONFIDENT · BANDS TOO NARROW", f"runs {abs(gap) * 100:.0f}pts under", "var(--mute-orange)") if gap < 0: return ("SLIGHTLY OVERCONFIDENT", f"runs {abs(gap) * 100:.0f}pts under", "var(--accent-amber-strong)") if gap > 0.15: return ("UNDERCONFIDENT · BANDS TOO WIDE", f"runs {gap * 100:.0f}pts over", "var(--mute-orange)") return ("SLIGHTLY UNDERCONFIDENT", f"runs {gap * 100:.0f}pts over", "var(--accent-amber-strong)") # -------------------------------------------------------------------------- # Summaries # -------------------------------------------------------------------------- class Summary: """Per-(model, asset, timeframe) track record, from the precomputed panels. The Space never reads the raw track record. It is megabytes across dozens of files and grows with every resolved forecast; the resolver collapses it to `arena/panels.json` and that is what boots here. A missing panel is a model with no resolved history, which renders the designed empty state -- not an error. """ def __init__(self, standings: pd.DataFrame, panels: dict | None = None): self.standings = standings if standings is not None else pd.DataFrame() panels = panels or {} self._panels = panels.get("panels", {}) self.totals = panels.get("totals", {}) def class_of(self, asset: str) -> str: spec = config.ASSETS.get(asset) return spec.asset_class if spec else "other" def standing(self, model_slug: str, asset: str) -> dict: """The standings row for this model in this asset's class.""" if not len(self.standings): return {} rows = self.standings[ (self.standings["model_slug"] == model_slug) & (self.standings["asset_class"] == self.class_of(asset))] return rows.iloc[0].to_dict() if len(rows) else {} def raw_panel(self, model_slug: str, asset: str, timeframe: str) -> dict: return self._panels.get(f"{model_slug}|{asset}|{timeframe}", {}) def panel(self, model_slug: str, asset: str, timeframe: str, horizon: int = 0) -> dict: """Everything the Track Record panel needs for one model.""" raw = self.raw_panel(model_slug, asset, timeframe) resolved = int(raw.get("resolved") or 0) coverage = raw.get("coverage_80") median_err = raw.get("median_pct_error") width = raw.get("mean_width") backfilled = raw.get("backfilled_share") chip, short, color = verdict_for(coverage, resolved) grade = raw.get("grade") or "-" return { "empty": resolved == 0, "resolved": resolved, "coverage": coverage, "median_err": median_err, "width": width, "backfilled_share": backfilled, "grade": grade, "grade_color": grade_color(grade), "verdict": chip, "verdict_short": short, "verdict_color": color, "thumbs": [_thumb_view(t) for t in raw.get("thumbs", [])], "total": int(raw.get("forecasts") or 0), } def _thumb_view(t: dict) -> dict: """A stored thumbnail, in the shape the renderer wants.""" hit = bool(t.get("hit")) err = t.get("err") return { "id": "", "date": t.get("date", DASH), "hit": "\u2713" if hit else "\u2715", "hit_color": "var(--fin-up)" if hit else "var(--fin-down)", "miss": not hit, "err": (f"{err:.2f}% ERR" if isinstance(err, (int, float)) else DASH), "backfilled": bool(t.get("backfilled")), "spark": t.get("spark") or {"band": "", "mid": "", "real": ""}, } # -------------------------------------------------------------------------- # Volatility # -------------------------------------------------------------------------- def volatility(run, history: pd.DataFrame, timeframe: str) -> dict: """Forecast dispersion against trailing realised volatility. For a path model the forecast figure is the standard deviation across sampled paths; for a quantile-only model it is the band half-width. Those are not the same statistic, and the label says which one is on screen rather than quietly presenting them as comparable. """ bars_per_day = 24 if timeframe == "1h" else 1 horizon_days = max(1.0, (run.horizon / bars_per_day)) if run else 1.0 # Before any forecast has been issued there is no history to read. An # empty frame has no columns at all, so this guards on the column rather # than on the length. if history is None or "close" not in getattr(history, "columns", []): return {"rows": [], "verdict": DASH, "note": "", "kind": ""} close = history["close"].to_numpy(dtype="float64") if len(close) < 10: return {"rows": [], "verdict": DASH, "note": "", "kind": ""} log_returns = np.diff(np.log(close)) # `a or b` on an array is a truthiness test, not a fallback. Slice widths # are clamped explicitly instead. recent_n = max(2, min(len(log_returns), int(bars_per_day))) week_n = max(2, min(len(log_returns), int(bars_per_day * 7))) realized_recent = float(np.std(log_returns[-recent_n:])) realized_week = float(np.std(log_returns[-week_n:])) scale = np.sqrt(bars_per_day) realized_recent_d = realized_recent * scale realized_week_d = realized_week * scale if run is None: forecast = None kind = "" else: # Terminal dispersion, expressed per-day so it is comparable with the # realised figures next to it. forecast = float(run.result.dispersion()[-1]) / np.sqrt(horizon_days) kind = ("SAMPLED-PATH DISPERSION" if run.capabilities.get("output") == "ohlcv_paths" else "BAND HALF-WIDTH") peak = max(v for v in (forecast or 0, realized_recent_d, realized_week_d, 1e-9)) rows = [ {"label": f"Forecast-implied vol", "value": pct(forecast, 2) + " / 24H", "pct": f"{min(100, (forecast or 0) / peak * 100):.0f}%", "fg": "var(--accent-amber-strong)"}, {"label": "Realized vol · recent", "value": pct(realized_recent_d, 2) + " / 24H", "pct": f"{min(100, realized_recent_d / peak * 100):.0f}%", "fg": "var(--text-secondary)"}, {"label": "Realized vol · 7d avg", "value": pct(realized_week_d, 2) + " / 24H", "pct": f"{min(100, realized_week_d / peak * 100):.0f}%", "fg": "var(--text-tertiary)"}, ] if forecast is None: verdict, note = DASH, "Run a forecast to compare its dispersion with realised volatility." else: ratio = forecast / (realized_week_d or 1e-9) if ratio > 1.15: verdict = "FORECAST > REALIZED" note = (f"{kind.capitalize()} is running {(ratio - 1) * 100:.0f}% above " f"trailing realised vol — this model is pricing a wider range " f"than the last week delivered.") elif ratio < 0.85: verdict = "FORECAST < REALIZED" note = (f"{kind.capitalize()} is {(1 - ratio) * 100:.0f}% below trailing " f"realised vol — the bands are tighter than recent moves justify.") else: verdict = "FORECAST ≈ REALIZED" note = (f"{kind.capitalize()} is close to trailing realised vol.") return {"rows": rows, "verdict": verdict, "note": note, "kind": kind} # -------------------------------------------------------------------------- # Headline stats # -------------------------------------------------------------------------- def headline_stats(registry: dict, standings: pd.DataFrame, archived: int | None, resolved: int | None) -> list[dict]: models = registry.get("models", {}) families = {m.get("family") for m in models.values() if m.get("family")} best_label, best_note, best_color = DASH, "no graded models yet", "var(--text-primary)" if standings is not None and len(standings): graded = standings[standings["grade"] != "-"] if len(graded): best = graded.assign(_gap=graded["calibration_gap"].abs()) \ .sort_values("_gap").iloc[0] best_label = str(best["grade"]) best_note = (f"{best['model_slug']} · " f"{pct(float(best['coverage_80']))} coverage") best_color = grade_color(best_label) awaiting = None if archived is not None and resolved is not None: awaiting = max(0, archived - resolved) return [ {"label": "Models enrolled", "value": count(len(models)), "note": f"{len(families)} adapter families", "mark": "var(--accent-amber)", "color": "var(--text-primary)"}, {"label": "Forecasts archived", "value": count(archived), "note": "frozen at issue, never revised", "mark": "var(--border-strong)", "color": "var(--text-primary)"}, {"label": "Resolved", "value": count(resolved), "note": (f"{count(awaiting)} awaiting horizon" if awaiting is not None else "awaiting the first resolver run"), "mark": "var(--border-strong)", "color": "var(--text-primary)"}, {"label": "Best calibrated", "value": best_label, "note": best_note, "mark": "var(--accent-moss)", "color": best_color}, ] def standings_rows(standings: pd.DataFrame, registry: dict, cls: str, selected: list[str]) -> list[dict]: """The rail's leaderboard, filtered by asset class.""" if standings is None or not len(standings): return [] df = standings if cls and cls != "All": wanted = {"Crypto": "crypto", "Equities": "equity"}.get(cls, cls.lower()) df = df[df["asset_class"] == wanted] if not len(df): return [] # Rank by calibration first, then by how much evidence there is. A model # with a perfect gap over 20 forecasts should not outrank one with a nearly # perfect gap over two thousand. df = df.assign(_gap=df["calibration_gap"].abs(), _graded=(df["grade"] != "-").astype(int)) df = df.sort_values(["_graded", "_gap", "resolved_count"], ascending=[False, True, False]) models = registry.get("models", {}) out = [] for i, (_, row) in enumerate(df.iterrows()): slug = str(row["model_slug"]) entry = models.get(slug, {}) out.append({ "rank": f"{i + 1:02d}", "slug": slug, "name": entry.get("display", slug), "grade": str(row["grade"]), "grade_color": grade_color(str(row["grade"])), "meta": (f"{count(row['resolved_count'])} RESOLVED · " f"{pct(float(row['coverage_80']))} COV · " f"{str(row['asset_class']).upper()}"), "selected": slug in selected, }) return out