src/view.py

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