"""The Arena's visible layer: the design's own markup, rendered from state. Gradio is the backend here and nothing else -- one `gr.HTML` sink, a hidden textbox the click bridge writes into, and a hidden button it clicks. Every control on the page is the `') cards = "".join(_model_card(m, m["slug"] in selected, data) for m in data["models"]) add = ( f'') return ( f'

' '

' f'Models' f'{e(sel_note)}' f'

{"".join(tabs)}

' f'

{cards}{add}

' ) def _model_card(model: dict, on: bool, data: dict) -> str: """One model card. Rendered from declared capabilities only.""" color = model["color"] caps = model["caps"] # The two capability chips the design specifies: what the model emits, and # what hardware it costs. Both come from the adapter's declaration. paths = caps.get("output") == "ohlcv_paths" gpu = caps.get("hardware") == "gpu" chips = [ ("OHLCV paths" if paths else "quantile line", "var(--accent-amber-strong)" if paths else "var(--text-secondary)", "var(--accent-amber-dim)" if paths else "var(--border-default)"), ("your GPU quota" if gpu else "CPU · instant", "var(--mute-orange)" if gpu else "var(--accent-moss-strong)", "var(--mute-orange)" if gpu else "var(--accent-moss-dim)"), ] chip_html = "".join( f'' f'{e(label)}' for label, fg, bd in chips) chk_style = (f'background:{a(color)};color:var(--stone-950);border-color:{a(color)}' if on else "") return ( f'') # -------------------------------------------------------------------------- # Controls # -------------------------------------------------------------------------- def controls(state: dict, data: dict) -> str: assets = "".join( f'' for slug, spec in config.ASSETS.items()) tfs = "".join( f'' for i, slug in enumerate(config.TIMEFRAMES)) options = _horizon_options(state["tf"]) hzs = "".join( f'' for i, h in enumerate(options)) hours = state["horizon"] * (1 if state["tf"] == "1h" else 24) hz_note = f"= {hours}H AHEAD" gear = _gear(state) if state.get("gear") else "" quota = _quota_banner(data) if data.get("quota_blocked") else "" run_label = ("▶ Forecast" if state["mode"] == "playground" else f"▶ Forecast {len(state['selected'])} models") runtime = data.get("runtime_hint", DASH) blocked = bool(data.get("quota_blocked")) return ( f'

' '

' 'ASSET' f'{assets}

' '

' 'TIMEFRAME' f'

{tfs}

' '

' 'HORIZON' f'

{hzs}

' f'{e(hz_note)}

' '

' f'' f'{gear}{quota}' f'{e(runtime)}' f'' '

' ) def _horizon_options(timeframe: str) -> list[int]: """Horizon choices, capped at what the timeframe allows.""" ceiling = config.MAX_HORIZON.get(timeframe, 96) return [h for h in (12, 24, 48, 96) if h <= ceiling] or [ceiling] def _gear(state: dict) -> str: rows = [ ("Sampled paths", str(state.get("n_samples", config.DEFAULT_N_SAMPLES))), ("Quantiles", "/".join(f"{q:g}" for q in config.QUANTILE_LEVELS)), ("Seed", str(state.get("seed", 42))), ] body = "".join( '

' f'{e(label)}' f'{e(value)}

' for label, value in rows) return ( '

' f'Sampling' f'{body}' '' 'SEED IS ARCHIVED WITH THE FORECAST

') def _quota_banner(data: dict) -> str: return ( '' '⚠' f'' f'{e(data.get("quota_msg", "GPU-tier model — sign in to use your quota"))}' f'SIGN IN ↗') # -------------------------------------------------------------------------- # Chart panel # -------------------------------------------------------------------------- def chart_section(state: dict, data: dict) -> str: runs = data.get("runs", []) mode = state["mode"] primary = runs[0] if runs else None title = "Matchup" if mode == "matchup" else "Forecast" who = (f"{len(runs)} MODELS" if mode == "matchup" else (data["models_by_slug"].get(state["selected"][0], {}).get("name", "") .upper() if state["selected"] else "")) meta = f'{state["asset"]} · {state["tf"]} · H+{state["horizon"]} · {who}' # The capability gate, in one place. `ghost_capable` is true when the # *selected* model declares OHLCV paths -- not when a result happens to # carry them. emits_paths = bool(primary and primary.capabilities.get("output") == "ohlcv_paths" and mode == "playground") # A model can emit paths and still have none to draw: forecasts archived # before paths were stored keep their quantiles and nothing else. Offering # a toggle that renders nothing is worse than saying why it is empty. paths_missing = bool(emits_paths and primary.result.paths is None) ghost_capable = emits_paths and not paths_missing ghost_locked = bool(primary and primary.capabilities.get("output") != "ohlcv_paths" and mode == "playground") if ghost_capable: on = state.get("ghosts", True) toggle = ( f'') elif paths_missing: toggle = ( '' '✕' f'' 'Paths not archived — forecast again to see them') elif ghost_locked: toggle = ( '' '✕' f'' 'Quantile line only — no OHLCV paths') else: toggle = "" if primary: frozen = (f'FROZEN {primary.issued_ts.strftime("%Y-%m-%d %H:%M")} UTC · ' f'SEED {primary.result.seed}') else: frozen = "NO FORECAST ISSUED YET" # A forecast rebuilt from the archive has to say how old it is. Showing # only its issue timestamp reads as current at a glance. provenance = "" if primary is not None and getattr(primary, "from_cache", False): age = view.age_label(primary.issued_ts) badge = "LAST ARCHIVED" if getattr(primary, "backfilled", False): badge = "LAST ARCHIVED · BACKFILL" provenance = ( '' f'{e(badge)}' f'ISSUED {e(age.upper())}') # How it is actually doing. The point of showing an old forecast is # that the answer is already partly known. score = view.live_scorecard(primary) if score: err = (f' · {score["median_err"]:.2f}% ERR' if score["median_err"] is not None else "") provenance += ( '' f'{e(score["label"])}{e(err)}' f'' f'{e(score["note"].upper())}') drawing = data.get("chart") or {} body = _chart_body(drawing) if drawing.get("svg") else _chart_empty(data) return ( f'

' '

' f'{e(title)}' f'{e(meta)}' '

{provenance}{toggle}' f'{e(frozen)}

' f'{_legend(data)}{body}

' ) def _legend(data: dict) -> str: runs = data.get("runs", []) if not runs: return "" items = [] for i, run in enumerate(runs): color = chart.series_color(i) panel = data["panels"].get(run.model_slug, {}) name = data["models_by_slug"].get(run.model_slug, {}).get("name", run.model_slug) width = panel.get("width") band = (f'10 / 50 / 90 · {pct(width, 1)} MEAN WIDTH' if width is not None else "10 / 50 / 90") grade = panel.get("grade", "-") dash = chart.FAN_DASH[i % 3] if len(runs) > 1 else "none" items.append( '

' '' f'{e(name)}' f'{e(band)}' f'{e(grade)}' f'' f'{e(panel.get("verdict_short", ""))}

') return ('

{"".join(items)}

') def _chart_body(drawing: dict) -> str: axis = "".join( f'{e(g["label"])}' for g in drawing.get("grid", [])) times = "".join( f'{e(t["label"])}' for t in drawing.get("time_labels", [])) left = a(drawing.get("now_left", "50%")) return ( '

' f'{drawing["svg"]}' f'' f'{e(drawing.get("boundary_label", "NOW"))} ' f'{e(drawing.get("now_time", ""))}' f'FORECAST →' 'VOLUME' '

' f'{times}

' f'

{axis}

') def _chart_empty(data: dict) -> str: err = data.get("error") if err: return _error_state(err) return ( '

' '▱' f'' 'No forecast issued' '' 'Pick a model and press Forecast. The result is frozen at issue and ' 'archived before it is drawn.

') def _error_state(err: dict) -> str: """The designed failure states: quota, load failure, rejected enrollment.""" kind = err.get("kind", "load_failure") glyph = {"quota": "⚠", "load_failure": "✕", "enroll": "✕", "no_data": "∅", "no_gpu": "⚠"}.get(kind, "✕") color = {"quota": "var(--mute-orange)", "no_gpu": "var(--mute-orange)"}.get(kind, "var(--fin-down)") extra = "" if kind == "quota": extra = (f'' 'SIGN IN ↗') elif kind == "no_gpu": # Nothing the visitor can do here, so the only honest offer is a Space # they control with hardware they choose. extra = (f'RUN IT ON YOUR OWN HARDWARE ↗') elif kind == "load_failure": # A model that will not load on this hardware is not always a bug in # the Space: duplicating it gives the user a box they control. extra = (f'DUPLICATE THIS SPACE ↗') return ( '

' f'' f'{e(glyph)}' f'' f'{e(err.get("title", "Forecast failed"))}' '' f'{e(err.get("message", ""))}' f'

{extra}' f'

') # -------------------------------------------------------------------------- # Track record # -------------------------------------------------------------------------- def trackrecord_section(state: dict, data: dict) -> str: mode = state["mode"] slugs = state["selected"] scope = (f'{state["asset"]} · {state["tf"]} · H+{state["horizon"]} · ' f'LAST {view.THUMBS} RESOLVED').upper() if mode == "matchup" and len(slugs) > 1: question = "Same stats, side by side — best in row highlighted." body = _scoreboard(state, data) else: primary = slugs[0] if slugs else None panel = data["panels"].get(primary, {}) name = data["models_by_slug"].get(primary, {}).get("name", primary or "") question = f"How wrong has {name} been here recently?" body = _empty_record() if panel.get("empty", True) else _playground_record(panel, data) return ( f'

' '

' f'Track record' f'' f'{e(question)}' f'{e(scope)}

' f'{body}

') def _empty_record() -> str: return ( '

' '∅' f'' 'No history yet' 'Forecasts made here start ' 'the record. The first resolved forecast for this model, asset and timeframe ' 'appears in this panel.' f'

') def _playground_record(panel: dict, data: dict) -> str: stats = [ ("Empirical 80% coverage", pct(panel.get("coverage")), f'target 80% · {count(panel.get("resolved"))} resolved', panel.get("verdict_color", "var(--text-primary)")), ("Median abs error", pct(panel.get("median_err", 0) / 100 if panel.get("median_err") is not None else None, 2), "at horizon close", "var(--text-primary)"), ("Mean band width", pct(panel.get("width"), 1), "10→90 span, % of price", "var(--text-primary)"), ("Backfilled share", pct(panel.get("backfilled_share")), "labelled in the record", "var(--mute-blue)"), ] stat_html = "".join( '

' f'

{e(label)}

' f'

{e(value)}

' f'

{e(note)}

' '

' for label, value, note, color in stats) cells = "".join(_thumb(t, data) for t in panel.get("thumbs", [])) vcolor = panel.get("verdict_color", "var(--text-tertiary)") return ( '

{stat_html}

' f'

{cells}

' '

' f'' f'{e(panel.get("verdict", ""))}' '{e(_verdict_note(panel))}' f'ALL {e(count(panel.get("total")))} ARCHIVED ↗' '

') def _verdict_note(panel: dict) -> str: coverage = panel.get("coverage") if coverage is None: return ("Not enough resolved forecasts here to say anything about " "calibration yet.") gap = coverage - config.NOMINAL_COVERAGE if abs(gap) <= 0.05: return ("Bands hold up on this pair: the realized path stayed inside the " "80% envelope about as often as it should.") if gap < 0: return ("Bands are narrower than the realized error justifies here. " "Read the median, not the confidence.") return ("Bands are wider than the realized error justifies here — the model " "is hedging more than this pair has needed.") def _thumb(t: dict, data: dict) -> str: spark = t["spark"] backfill = ('BACKFILL' if t["backfilled"] else "") return ( f'

' '

' f'{e(t["date"])}' f'{e(t["hit"])}

' '' '

' f'{e(t["err"])}{backfill}

') def _scoreboard(state: dict, data: dict) -> str: """The matchup table: same stats, aligned, best in row highlighted.""" slugs = [s for s in state["selected"] if s in data["panels"]] if not slugs: return _empty_record() panels = [data["panels"][s] for s in slugs] cols = f'1.15fr repeat({len(slugs)},minmax(0,1fr))' heads = "".join( '

' f'' f'{e(data["models_by_slug"].get(s, {}).get("name", s))}' f'{e(p.get("grade", "-"))}

' for i, (s, p) in enumerate(zip(slugs, panels))) # (label, hint, value fn, "best" fn or None). Coverage is scored on # distance from nominal, not on being highest -- a narrower band is not a # better one, and the design says so on the page. rows = [ ("80% band coverage", "TARGET 80%", lambda p: pct(p.get("coverage")), lambda p: (-abs(p["coverage"] - config.NOMINAL_COVERAGE) if p.get("coverage") is not None else None)), ("Median abs error", "LOWER BETTER", lambda p: pct(p["median_err"] / 100 if p.get("median_err") is not None else None, 2), lambda p: (-p["median_err"] if p.get("median_err") is not None else None)), ("Mean band width", "CONTEXT ONLY", lambda p: pct(p.get("width"), 1), None), ("Resolved forecasts", "SAMPLE SIZE", lambda p: count(p.get("resolved")), lambda p: p.get("resolved")), ("Calibration verdict", "FROM RESOLVED ONLY", lambda p: p.get("verdict", DASH), None), ] body = [] for label, hint, value_of, best_of in rows: best_index = None if best_of is not None: scored = [(i, best_of(p)) for i, p in enumerate(panels)] scored = [(i, v) for i, v in scored if v is not None] if scored: best_index = max(scored, key=lambda kv: kv[1])[0] cells = [] for i, p in enumerate(panels): is_best = i == best_index fg = ("var(--accent-moss-strong)" if is_best else "var(--text-primary)") if label.startswith("Calibration"): fg = p.get("verdict_color", "var(--text-primary)") cells.append( '

' f'{e(value_of(p))}' '{"BEST" if is_best else ""}' '

') body.append( f'

' '

' f'{e(label)}' f'{e(hint)}' f'

{"".join(cells)}

') return ( f'

' '

' 'METRIC

' f'{heads}

{"".join(body)}' '

' 'i' 'Highlighted cell is best in row. Coverage closest to 80% ' 'wins — a narrower band is not a better one.

') # -------------------------------------------------------------------------- # Rail # -------------------------------------------------------------------------- def rail(state: dict, data: dict) -> str: return ('') def _standings(state: dict, data: dict) -> str: rows = data.get("standings_rows", []) enrolled = len(data.get("models", [])) tabs = "".join( f'' for c in ("All", "Crypto", "Equities")) if rows: body = "".join( f'' for r in rows) else: body = ('

' 'No forecasts have resolved yet. Standings appear ' 'once the resolver has outcomes to score.

') right = ('{enrolled} ENROLLED') return ( f'

' f'{_panel_title("Arena standings", right)}' f'

{tabs}

' f'{body}' '

' '' 'GRADES FROM RESOLVED FORECASTS ONLY' f'' 'FULL TABLE ↗

') def _volatility(data: dict) -> str: vol = data.get("volatility") or {} rows = vol.get("rows", []) if not rows: return "" body = "".join( '

' '

' f'{e(r["label"])}' f'{e(r["value"])}

' '

' f'

' for r in rows) right = ('{e(vol.get("verdict", ""))}') return ( f'

{_panel_title("Volatility read", right)}' '

' f'{body}' '{e(vol.get("note", ""))}

') def _log(state: dict, data: dict) -> str: entries = state.get("log", [])[:view.LOG_LIMIT] if entries: body = "".join( '

' f'{e(l["time"])}' '' f'{e(l["name"])}' f'{e(l["meta"])}' '' f'

' for l in entries) else: body = ('

' '' 'Forecasts you issue this session appear here.

') right = ('THIS SESSION') return ( f'

{_panel_title("Forecast log", right)}{body}' '

' '' 'ARCHIVED IMMUTABLY · SEED + CONFIG SAVED

') def _extend(state: dict, data: dict) -> str: cov = data.get("coverage") or {} pct_label = cov.get("pct_label", DASH) width = cov.get("pct_width", "0%") rows = "".join( '

' f'{e(label)}' f'{e(value)}

' for label, value in cov.get("rows", [])) label = "Contribute a backfill run (uses your quota)" note = state.get("backfill_note") note_html = (f'{e(note)}' if note else "") return ( f'

{_panel_title("Extend the record")}' '

' '

' f'' f'{e(cov.get("scope", ""))}' f'{e(pct_label)}

' '

' f'

{rows}

' f'' f'{note_html}' '' 'BACKFILLED ENTRIES ARE LABELLED IN THE RECORD

') def _enroll(state: dict, data: dict) -> str: note = state.get("enroll_note") or "A smoke test runs before a model is enrolled" ok = state.get("enroll_ok") color = ("var(--accent-moss-strong)" if ok else "var(--fin-down)" if ok is False else "var(--text-tertiary)") families = "".join( f'' for f in data.get("families", ())) return ( f'

{_panel_title("Enroll a model")}' '

' f'' f'

{families}

' '

' f'

' f'{e(note)}' 'Only models loadable through a vetted adapter family are ' 'accepted. No user-supplied code is ever executed.

') # -------------------------------------------------------------------------- # Footer # -------------------------------------------------------------------------- def methodology(data: dict) -> str: return ( '

' 'i' 'Forecasts are frozen at issue and never revised; quantiles ' 'from N sampled paths where applicable; seeds shown; backfilled entries ' 'labelled; grades computed from resolved forecasts only. Nothing here is ' 'financial advice.' f'METHODOLOGY ↗

') def footer(data: dict) -> str: nav = data.get("nav") or {} products = [i.get("name", "") for i in nav.get("items", [])][:12] prod_html = "".join( f'{e(p)}' for p in products) resources = "".join( f'{e(label)}' for label, url in data.get("footer_resources", ())) return ( f'') # -------------------------------------------------------------------------- # The page # -------------------------------------------------------------------------- def page(state: dict, data: dict) -> str: """The whole document body, from state. Pure -- no I/O anywhere below.""" return ( '

' f'{sidebar.render(data.get("nav") or {}, active="Forecast Arena", emit=emit)}' '

' f'{header(state, data)}' '

' f'{title_row(state, data)}' f'{stats_bar(data.get("stats", []))}' f'{models_section(state, data)}' f'{controls(state, data)}' '

' '

' f'{chart_section(state, data)}' f'{trackrecord_section(state, data)}' '

' f'{rail(state, data)}' '

' f'{methodology(data)}' f'{footer(data)}' '