Spaces:
Running on Zero
Running on Zero
| """Bit Trading Company — Backtest Lab. | |
| The visible chrome and every control is the design's own markup, rendered by | |
| `src/ui/shell.py` and wired back to Python through `src/ui/bridge.py`. Gradio | |
| owns the plots and the transport; it no longer owns the layout. | |
| That split exists because the design is built from `<button>` elements with | |
| exact inline styles — 23 buttons against a single `<input>` in the whole file — | |
| while Gradio renders a different DOM for the same concepts. Restyling Gradio's | |
| components from outside only ever approximated it. See docs/DESIGN.md. | |
| Everything below the chrome is unchanged: the Compare catalog, the engine, the | |
| store and the charts are the same code they were. | |
| """ | |
| from __future__ import annotations | |
| import copy | |
| import logging | |
| import os | |
| import uuid | |
| import gradio as gr | |
| import pandas as pd | |
| from src import catalog, charts, config, extension, runtime, strategies | |
| from src.runtime import RunError, RunRecord, RunRequest | |
| from src.ui import components as C | |
| from src.ui import compare_tab as CT | |
| from src.ui import shell, theme | |
| from src.ui.bridge import (ACTION_ELEMENT_ID, BRIDGE_JS, BRIDGE_LOAD_JS, | |
| TRIGGER_ELEMENT_ID, parse_action, parse_pair) | |
| from src.ui.format import EM, count, esc, money, num, pct, tone | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(message)s") | |
| log = logging.getLogger("bit.app") | |
| GLOSSARY = [ | |
| ("SHARPE", "Annualized mean excess return over return volatility. " | |
| "Above 1 is good; above 3 usually means a bug."), | |
| ("SORTINO", "Sharpe with only downside deviation in the denominator."), | |
| ("MAX DRAWDOWN", "Worst peak-to-trough decline of the equity curve."), | |
| ("PROFIT FACTOR", "Gross profit over gross loss."), | |
| ("R-MULTIPLE", "Trade P&L in units of initial risk."), | |
| ("MAE / MFE", "Worst and best unrealized excursion while open."), | |
| ("WALK-FORWARD", "Train on a rolling window, test on the next unseen one."), | |
| ("OOS", "Out of sample: data the parameters never saw."), | |
| ("BASELINE", "A naive forecast. If a learned model cannot beat these, it " | |
| "has not earned its inference cost."), | |
| ] | |
| NUMERIC_PARAMS = { | |
| "commission_bps": float, "slippage_bps": float, "size_pct": float, | |
| "leverage": float, "sl_pct": float, "tp_pct": float, "trail_pct": float, | |
| "train_months": int, "test_months": int, "roll_months": int, | |
| "holdout_months": int, | |
| } | |
| # -------------------------------------------------------------------------- | |
| # State | |
| # -------------------------------------------------------------------------- | |
| def default_state() -> dict: | |
| return { | |
| "strategy": "SMA Crossover", | |
| "asset": "BTC-USD", | |
| "timeframe": "1d", | |
| "range": "3Y", | |
| "model": None, | |
| "params": dict(strategies.defaults_for("SMA Crossover")), | |
| "costs_on": True, | |
| "commission_bps": config.DEFAULT_COMMISSION_BPS, | |
| "slippage_bps": config.DEFAULT_SLIPPAGE_BPS, | |
| "slippage_model": "fixed", | |
| "sizing_mode": "fixed_pct", | |
| "size_pct": 1.0, | |
| "leverage": 1.0, | |
| "sl_pct": None, "tp_pct": None, "trail_pct": None, | |
| "validation_mode": "walk_forward", | |
| "train_months": 12, "test_months": 3, "roll_months": 3, | |
| "holdout_months": config.DEFAULT_HOLDOUT_MONTHS, | |
| "acc": {"strategy": True, "universe": True, | |
| "costs": False, "sizing": False, "validation": False}, | |
| "cfg_id": uuid.uuid4().hex[:4].upper(), | |
| "coverage": "", | |
| "needs_signals": False, | |
| "log_scale": False, "cvd": False, | |
| "metric": "OOS Sharpe", "topn": 15, | |
| "filters": {"assets": [], "timeframes": [], "strategies": [], | |
| "models": [], "min_trades": 0, | |
| "require_oos": True, "hide_baselines": False}, | |
| "sig_asset": "BTC-USD", "sig_tf": "1d", | |
| "tab": "compare", # "compare" | "backtest" | |
| "user": None, | |
| # The bridge fires through two event paths for robustness; the nonce | |
| # is what stops the same click being executed twice. | |
| "last_nonce": "", | |
| } | |
| def to_request(st: dict) -> RunRequest: | |
| return RunRequest( | |
| strategy=st["strategy"], asset=st["asset"], timeframe=st["timeframe"], | |
| date_range=st["range"], model_slug=st.get("model") or "", | |
| params=dict(st["params"]), | |
| costs_on=st["costs_on"], commission_bps=float(st["commission_bps"]), | |
| slippage_bps=float(st["slippage_bps"]), slippage_model=st["slippage_model"], | |
| sizing_mode=st["sizing_mode"], size_pct=float(st["size_pct"]), | |
| leverage=float(st["leverage"]), | |
| sl_pct=(float(st["sl_pct"]) / 100.0 if st["sl_pct"] else None), | |
| tp_pct=(float(st["tp_pct"]) / 100.0 if st["tp_pct"] else None), | |
| trail_pct=(float(st["trail_pct"]) / 100.0 if st["trail_pct"] else None), | |
| validation_mode=st["validation_mode"], | |
| train_months=int(st["train_months"]), test_months=int(st["test_months"]), | |
| roll_months=int(st["roll_months"]), holdout_months=int(st["holdout_months"]), | |
| ) | |
| def refresh_context(st: dict) -> dict: | |
| """Recompute what depends on the current asset/timeframe/strategy.""" | |
| preset = strategies.PRESETS.get(st["strategy"]) | |
| st["needs_signals"] = bool(preset and preset.needs_signals) | |
| models = runtime.available_models(st["asset"], st["timeframe"]) | |
| if st["needs_signals"] and st.get("model") not in models: | |
| st["model"] = models[0] if models else None | |
| cov = runtime.price_coverage_for(st["asset"], st["timeframe"]) | |
| st["coverage"] = (f"cached {cov[0]} to {cov[1]} · {len(models)} models" | |
| if cov else "no cached coverage") | |
| return st | |
| def apply_action(st: dict, raw: str) -> tuple[dict, bool]: | |
| """Fold one UI action into state. Returns (state, should_run).""" | |
| action = parse_action(raw) | |
| if action is None or action.is_noop: | |
| return st, False | |
| # Same click delivered twice (textbox change *and* trigger click) must not | |
| # run the backtest twice. | |
| nonce = raw.split("|", 1)[1] if "|" in raw else "" | |
| if nonce and nonce == st.get("last_nonce"): | |
| return st, False | |
| st["last_nonce"] = nonce | |
| k, v = action.key, action.value | |
| if k == "tab" and v in ("compare", "backtest"): | |
| st["tab"] = v | |
| elif k == "acc": | |
| st["acc"][v] = not st["acc"].get(v, False) | |
| elif k == "strategy" and v in strategies.PRESETS: | |
| st["strategy"] = v | |
| st["params"] = dict(strategies.defaults_for(v)) | |
| elif k == "asset" and v in config.ASSETS: | |
| st["asset"] = v | |
| elif k == "tf" and v in config.TIMEFRAMES: | |
| st["timeframe"] = v | |
| elif k == "range" and v in runtime.RANGE_YEARS: | |
| st["range"] = v | |
| elif k == "model": | |
| st["model"] = v or None | |
| elif k == "slippage" and v in ("fixed", "volume_scaled"): | |
| st["slippage_model"] = v | |
| elif k == "sizing" and v in ("fixed_pct", "vol_target"): | |
| st["sizing_mode"] = v | |
| elif k == "validation" and v in ("walk_forward", "split", "holdout", "none"): | |
| st["validation_mode"] = v | |
| elif k == "costs": | |
| _, val = parse_pair(v) if "=" in v else ("", v) | |
| st["costs_on"] = str(val) == "1" | |
| elif k == "param": | |
| name, val = parse_pair(v) | |
| st = _set_param(st, name, val) | |
| elif k == "logscale": | |
| st["log_scale"] = not st["log_scale"] | |
| elif k == "cvd": | |
| st["cvd"] = not st["cvd"] | |
| elif k == "metric" and v in CT.RANK_METRICS: | |
| st["metric"] = v | |
| elif k == "topn": | |
| try: | |
| st["topn"] = max(5, min(50, int(v))) | |
| except ValueError: | |
| pass | |
| elif k == "filter": | |
| st = _apply_filter(st, v) | |
| elif k == "reset": | |
| st["filters"] = default_state()["filters"] | |
| st["metric"], st["topn"] = "OOS Sharpe", 15 | |
| elif k == "sigasset" and v in config.ASSETS: | |
| st["sig_asset"] = v | |
| elif k == "sigtf" and v in config.TIMEFRAMES: | |
| st["sig_tf"] = v | |
| elif k == "example": | |
| st.update(strategy="Chronos Forecast Follower", asset="BTC-USD", | |
| timeframe="1d", range="3Y", | |
| params=dict(strategies.defaults_for("Chronos Forecast Follower"))) | |
| st = refresh_context(st) | |
| return st, True | |
| elif k == "run": | |
| return refresh_context(st), True | |
| return refresh_context(st), False | |
| # Which filter kinds are multi-select lists, and what each is validated against. | |
| _FILTER_LISTS = { | |
| "asset": ("assets", lambda: set(config.ASSETS)), | |
| "tf": ("timeframes", lambda: set(config.TIMEFRAMES)), | |
| "strategy": ("strategies", lambda: set(catalog.CATALOG_STRATEGIES)), | |
| "model": ("models", lambda: set(config.SEED_MODELS)), | |
| } | |
| def _apply_filter(st: dict, value: str) -> dict: | |
| """Fold `kind=value` into the filter set. Unknown values are ignored.""" | |
| kind, val = parse_pair(value) | |
| filters = dict(st["filters"]) | |
| if kind in _FILTER_LISTS: | |
| key, allowed = _FILTER_LISTS[kind] | |
| if val not in allowed(): | |
| return st | |
| chosen = list(filters.get(key, [])) | |
| # Clicking a selected chip clears it, so "none selected" means all. | |
| filters[key] = [x for x in chosen if x != val] if val in chosen \ | |
| else chosen + [val] | |
| elif kind == "min_trades": | |
| try: | |
| filters["min_trades"] = max(0, min(500, int(val))) | |
| except ValueError: | |
| return st | |
| elif kind in ("require_oos", "hide_baselines"): | |
| filters[kind] = not filters.get(kind, False) | |
| else: | |
| return st | |
| st["filters"] = filters | |
| return st | |
| def _set_param(st: dict, name: str, raw: str) -> dict: | |
| """Set a numeric field, ignoring anything that is not a number.""" | |
| if name in NUMERIC_PARAMS: | |
| cast = NUMERIC_PARAMS[name] | |
| if raw == "": | |
| st[name] = None if name in ("sl_pct", "tp_pct", "trail_pct") else st[name] | |
| return st | |
| try: | |
| st[name] = cast(float(raw)) | |
| except (TypeError, ValueError): | |
| pass | |
| return st | |
| preset = strategies.PRESETS.get(st["strategy"]) | |
| valid = {p[0] for p in (preset.params if preset else ())} | |
| if name in valid: | |
| try: | |
| st["params"][name] = float(raw) | |
| except (TypeError, ValueError): | |
| pass | |
| return st | |
| # -------------------------------------------------------------------------- | |
| # Renderers | |
| # -------------------------------------------------------------------------- | |
| def render_left(st: dict) -> str: | |
| preset = strategies.PRESETS.get(st["strategy"]) | |
| params = [(k, label, default) for k, label, default, _lo, _hi | |
| in (preset.params if preset else ())] | |
| return shell.left_panel( | |
| st, | |
| presets=[p.name for p in strategies.PRESETS.values() if p.available], | |
| assets=runtime.available_assets(), | |
| timeframes=list(config.TIMEFRAMES), | |
| models=runtime.available_models(st["asset"], st["timeframe"]), | |
| preset_params=params, | |
| ) | |
| ON_SPACE = bool(os.environ.get("SPACE_ID")) | |
| def render_top(st: dict, rec: RunRecord | None) -> str: | |
| """The header: nav, run status, glossary, sign-in. | |
| The context chip only appears on the Backtest tab -- on Compare it would be | |
| describing a run the user is not looking at. | |
| """ | |
| common = dict(tab=st.get("tab", "compare"), glossary=GLOSSARY, | |
| user=st.get("user"), on_space=ON_SPACE) | |
| if rec is None or st.get("tab") != "backtest": | |
| status = (f"RUN {rec.run_id} COMPLETE" if rec else "NO RUN LOADED") | |
| tone = "ok" if rec else "idle" | |
| return shell.top_bar(status=status, tone=tone, **common) | |
| mode = {"walk_forward": "WALK-FORWARD", "holdout": "HOLDOUT", | |
| "split": "SPLIT", "none": "NO SPLIT"}.get(st["validation_mode"], "") | |
| try: | |
| a, b = runtime.window_for(st["asset"], st["timeframe"], st["range"]) | |
| span = f"{a.date()} to {b.date()}" | |
| except Exception: | |
| span = st["range"] | |
| ctx = f'{st["asset"]} · {st["timeframe"].upper()} · {span} · {mode}' | |
| return shell.top_bar(context=ctx, status=f"RUN {rec.run_id} COMPLETE", | |
| tone="ok", elapsed=f"{rec.elapsed_s:.1f}s", **common) | |
| def render_stat_band(rec: RunRecord | None) -> str: | |
| if rec is None: | |
| return "" | |
| r = rec.result | |
| a, i, o = r.metrics_all, r.metrics_is, r.metrics_oos | |
| def seg(m, fmt, key, *args): | |
| return EM if m.bars == 0 else fmt(getattr(m, key), *args) | |
| def isoos(fmt, key, *args): | |
| return f"IS {seg(i, fmt, key, *args)} · OOS {seg(o, fmt, key, *args)}" | |
| bench = (float(r.benchmark_equity.iloc[-1] / r.benchmark_equity.iloc[0] - 1.0) | |
| if len(r.benchmark_equity) else float("nan")) | |
| gap = a.total_return - bench if pd.notna(bench) else float("nan") | |
| spec = [ | |
| ("Total return", pct(a.total_return), isoos(pct, "total_return"), | |
| tone(a.total_return)), | |
| ("CAGR", pct(a.cagr), isoos(pct, "cagr"), tone(a.cagr)), | |
| ("Sharpe", num(a.sharpe), isoos(num, "sharpe"), tone(a.sharpe)), | |
| ("Sortino", num(a.sortino), isoos(num, "sortino"), tone(a.sortino)), | |
| ("Max drawdown", pct(a.max_drawdown), isoos(pct, "max_drawdown"), "down"), | |
| ("Win rate", pct(a.win_rate, 0, signed=False), | |
| f"IS {seg(i, pct, 'win_rate', 0, False)} · OOS {seg(o, pct, 'win_rate', 0, False)}", ""), | |
| ("Profit factor", num(a.profit_factor), isoos(num, "profit_factor"), | |
| tone(a.profit_factor - 1.0)), | |
| ("Trades", count(a.trade_count), | |
| f"IS {seg(i, count, 'trade_count')} · OOS {seg(o, count, 'trade_count')}", ""), | |
| ("Exposure", pct(a.exposure, 0, signed=False), | |
| f"IS {seg(i, pct, 'exposure', 0, False)} · OOS {seg(o, pct, 'exposure', 0, False)}", ""), | |
| ("vs buy & hold", pct(gap), f"costs paid {money(r.costs_paid)}", tone(gap)), | |
| ] | |
| cells = [shell.stat_cell(l, v, s, tone_class=("up" if t == "bit-up" else | |
| "down" if t == "bit-down" else "")) | |
| for l, v, s, t in spec] | |
| notes = [shell.note(esc(n), danger=True) for n in getattr(r.plan, "notes", [])] | |
| if r.metrics_holdout is not None: | |
| h = r.metrics_holdout | |
| ok = h.total_return > 0 | |
| notes.append(shell.note( | |
| f"<b>LOCKED HOLDOUT</b> · return {pct(h.total_return)} · " | |
| f"Sharpe {num(h.sharpe)} · {h.bars} bars never used for any " | |
| f"parameter choice." + ("" if ok else " <b>It loses money here.</b>"), | |
| danger=not ok)) | |
| return shell.stat_band(cells, notes) | |
| def render_table(df, *, align_right=(), empty="no rows", max_height="430px") -> str: | |
| headers, rows = shell.frame_to_rows(df) | |
| return shell.table(headers, rows, align_right=set(align_right), | |
| empty=empty, max_height=max_height) | |
| def trades_frame(rec: RunRecord | None) -> pd.DataFrame: | |
| cols = ["#", "Entry", "Exit", "Side", "Entry px", "Exit px", "Size", | |
| "Gross", "Costs", "Net", "R", "Bars", "MAE", "Segment", "Trigger"] | |
| if rec is None or rec.result.trades.empty: | |
| return pd.DataFrame(columns=cols) | |
| t = rec.result.trades | |
| return pd.DataFrame({ | |
| "#": t["id"], | |
| "Entry": t["entry_ts"].dt.strftime("%Y-%m-%d %H:%M"), | |
| "Exit": t["exit_ts"].dt.strftime("%Y-%m-%d %H:%M"), | |
| "Side": t["side"].str.upper(), | |
| "Entry px": t["entry_px"].round(2), "Exit px": t["exit_px"].round(2), | |
| "Size": t["size"].round(4), "Gross": t["gross_pnl"].round(2), | |
| "Costs": t["costs"].round(2), "Net": t["net_pnl"].round(2), | |
| "R": t["r_multiple"].round(2), "Bars": t["duration_bars"], | |
| "MAE": (t["mae"] * 100).round(1), "Segment": t["segment"], | |
| "Trigger": t["trigger"], | |
| }) | |
| def report_markdown(rec: RunRecord | None) -> str: | |
| if rec is None: | |
| return "_Run a backtest to generate the report._" | |
| import json | |
| from dataclasses import asdict | |
| r, req = rec.result, rec.request | |
| a, o, h = r.metrics_all, r.metrics_oos, r.metrics_holdout | |
| bench = (float(r.benchmark_equity.iloc[-1] / r.benchmark_equity.iloc[0] - 1.0) | |
| if len(r.benchmark_equity) else float("nan")) | |
| ratio = (o.sharpe / r.metrics_is.sharpe) if r.metrics_is.sharpe else float("nan") | |
| grade, checks = runtime.overfit_verdict(rec) | |
| lines = [ | |
| f"### {rec.label}", | |
| f"`RUN {rec.run_id} · {rec.created_at} · {req.validation_mode.upper()} · " | |
| f"COSTS {'ON' if req.costs_on else 'OFF'}`", "", | |
| f"Over {a.bars} bars and {a.trade_count} trades the strategy returns " | |
| f"**{pct(a.total_return)}** (CAGR {pct(a.cagr)}, Sharpe {num(a.sharpe)}) " | |
| f"against **{pct(bench)}** for buy and hold, with a maximum drawdown of " | |
| f"{pct(a.max_drawdown)}. Modelled costs of {money(r.costs_paid)} are " | |
| f"already deducted.", "", | |
| (f"Out-of-sample Sharpe is {num(o.sharpe)}, {num(ratio)} of in-sample." | |
| if o.bars else | |
| "**No out-of-sample period was produced**, so every number above is " | |
| "in-sample."), "", | |
| (f"On the locked holdout ({h.bars} bars) it returns {pct(h.total_return)} " | |
| f"at Sharpe {num(h.sharpe)}." if h else ""), | |
| "", f"**Verdict: {grade}**", "", | |
| ] | |
| lines += [f"- {m} {t}" for m, t in checks] | |
| lines += ["", "#### Config snapshot", "```json", | |
| json.dumps(asdict(req), indent=2, sort_keys=True), "```"] | |
| return "\n".join(lines) | |
| def build_overview(rec: RunRecord, *, log_scale: bool, cvd: bool): | |
| r = rec.result | |
| bpy = config.bars_per_year(rec.request.asset, rec.request.timeframe) | |
| window = {"1d": 90, "1h": 24 * 30, "15m": 4 * 24 * 14}.get( | |
| rec.request.timeframe, 90) | |
| costs_html = ( | |
| shell.note(f"<b>COSTS PAID TOTAL: {money(r.costs_paid)}</b>. " | |
| "The costed number is the real one.") | |
| if rec.request.costs_on else | |
| shell.note("<b>COSTS ARE OFF.</b> These numbers are not achievable.", | |
| danger=True)) | |
| return ( | |
| charts.equity_curve(r.equity, r.benchmark_equity, plan=r.plan, | |
| log_scale=log_scale, cvd=cvd), | |
| charts.regime_strip(r.prices), | |
| charts.underwater_chart(r.equity), | |
| charts.rolling_sharpe_chart(r.equity, window, bpy), | |
| charts.price_with_trades(r.prices, r.trades, cvd=cvd), | |
| charts.pnl_histogram(r.trades, cvd=cvd), | |
| charts.holding_period_histogram(r.trades), | |
| charts.mae_mfe_scatter(r.trades, cvd=cvd), | |
| costs_html, | |
| ) | |
| # -------------------------------------------------------------------------- | |
| # App | |
| # -------------------------------------------------------------------------- | |
| def build_app() -> gr.Blocks: | |
| all_tfs = list(config.TIMEFRAMES) | |
| boot = default_state() | |
| with gr.Blocks(theme=theme.bit_theme(), css=theme.full_css(), | |
| head=BRIDGE_JS, title="Bit · Backtest Lab", | |
| analytics_enabled=False) as demo: | |
| state = gr.State(boot) | |
| history = gr.State([]) | |
| current = gr.State(None) | |
| # Bridge target. It must be RENDERED -- `visible=False` removes the | |
| # element from the DOM entirely, leaving the click listener with | |
| # nothing to write into -- so it is rendered and hidden in CSS | |
| # (`#bit-action` is clipped to a 1px box, see theme.py). | |
| action_box = gr.Textbox(elem_id=ACTION_ELEMENT_ID, visible=True, | |
| label="", show_label=False, | |
| container=False, interactive=True) | |
| # Clicked by the bridge after it writes the payload. A real Button | |
| # click is the event path Gradio always honours. | |
| action_trigger = gr.Button("", elem_id=TRIGGER_ELEMENT_ID, | |
| visible=True, variant="secondary") | |
| top_html = gr.HTML(render_top(boot, None)) | |
| # `gr.LoginButton`, plain and unwrapped. It is what makes Gradio mount | |
| # the OAuth routes at all -- without it /login/huggingface is a 404 -- | |
| # and it is deliberately NOT nested in a custom sticky flex row or | |
| # given layout overrides. Dressing it up is what coincided with the | |
| # redirect loop, and a sign-in that works is worth more than one that | |
| # sits perfectly in the header bar. | |
| if ON_SPACE: | |
| with gr.Row(elem_classes="bit-loginrow"): | |
| gr.LoginButton(value="Sign in with Hugging Face", size="sm") | |
| with gr.Row(elem_classes="bit-zones", equal_height=False): | |
| # Strategy Builder belongs to the Backtest tab only. | |
| with gr.Column(elem_classes="bit-zone-left", min_width=0, | |
| visible=False) as left_col: | |
| left_html = gr.HTML(render_left(boot)) | |
| with gr.Column(elem_classes="bit-zone-center", min_width=0): | |
| # ---------------- Compare ---------------- | |
| with gr.Column(visible=True) as compare_view: | |
| catalog_meta = gr.HTML("") | |
| with gr.Tabs(): | |
| with gr.Tab("Leaderboard"): | |
| lb_controls = gr.HTML("") | |
| podium = gr.HTML("") | |
| lb_meta = gr.HTML("") | |
| lb_table = gr.HTML("") | |
| gr.HTML(shell.micro( | |
| "returns over time · top ranked · cumulative, " | |
| "costs included")) | |
| lb_overlay = gr.Plot() | |
| gr.HTML(shell.micro( | |
| "risk vs return · marker area = trade count")) | |
| lb_scatter = gr.Plot() | |
| with gr.Tab("Models"): | |
| models_note = gr.HTML("") | |
| with gr.Row(): | |
| acc_plot = gr.Plot() | |
| cal_plot = gr.Plot() | |
| model_bars = gr.Plot() | |
| score_table = gr.HTML("") | |
| with gr.Tab("Signals"): | |
| sig_panel = gr.HTML("") | |
| with gr.Tab("Run history"): | |
| runs_table = gr.HTML("") | |
| with gr.Tab("Coverage"): | |
| coverage_kpis = gr.HTML("") | |
| coverage_html = gr.HTML("") | |
| extend_panel = gr.HTML("") | |
| with gr.Row(): | |
| ext_model = gr.Dropdown(list(config.SEED_MODELS), | |
| label="Model", scale=2) | |
| ext_asset = gr.Dropdown(list(config.ASSETS), | |
| label="Asset", scale=2) | |
| ext_tf = gr.Dropdown(all_tfs, value="1d", | |
| label="Timeframe", scale=1) | |
| with gr.Row(): | |
| ext_start = gr.Textbox(label="Start (YYYY-MM-DD)", | |
| scale=2) | |
| ext_end = gr.Textbox(label="End (YYYY-MM-DD)", | |
| scale=2) | |
| with gr.Row(): | |
| estimate_btn = gr.Button("Estimate", size="sm", | |
| elem_classes="bit-ghost-btn") | |
| extend_btn = gr.Button("Extend coverage", size="sm", | |
| elem_classes="bit-run-btn") | |
| extend_out = gr.HTML("") | |
| with gr.Row(): | |
| add_family = gr.Dropdown( | |
| list(config.ALLOWED_ADAPTER_FAMILIES), | |
| value="chronos", label="Adapter family", scale=1) | |
| add_model_id = gr.Textbox(label="HF model id", scale=2) | |
| add_btn = gr.Button("Smoke test & add", size="sm", | |
| elem_classes="bit-ghost-btn", | |
| scale=1) | |
| add_out = gr.HTML("") | |
| # ---------------- Backtest ---------------- | |
| with gr.Column(visible=False) as backtest_view: | |
| # Before a run there is nothing honest to show, so the tabs | |
| # stay hidden rather than rendering a grid of empty axes. | |
| empty_html = gr.HTML(C.empty_state(), visible=True) | |
| with gr.Column(visible=False) as results_view: | |
| stat_html = gr.HTML("") | |
| with gr.Tabs(): | |
| with gr.Tab("Overview"): | |
| chart_opts = gr.HTML("") | |
| equity_plot = gr.Plot() | |
| regime_plot = gr.Plot() | |
| with gr.Row(): | |
| underwater_plot = gr.Plot() | |
| rolling_plot = gr.Plot() | |
| price_plot = gr.Plot() | |
| with gr.Row(): | |
| pnl_plot = gr.Plot() | |
| hold_plot = gr.Plot() | |
| mae_plot = gr.Plot() | |
| costs_note = gr.HTML("") | |
| with gr.Tab("Trades"): | |
| trades_head = gr.HTML("") | |
| trades_html = gr.HTML("") | |
| with gr.Tab("Robustness"): | |
| verdict_html = gr.HTML("") | |
| with gr.Row(): | |
| wf_plot = gr.Plot() | |
| mc_plot = gr.Plot() | |
| with gr.Tab("Report"): | |
| report_md = gr.Markdown("") | |
| report_equity = gr.Plot() | |
| gr.HTML(shell.footer()) | |
| # ------------------------------------------------------------------ | |
| # Wiring | |
| # ------------------------------------------------------------------ | |
| overview_out = [equity_plot, regime_plot, underwater_plot, rolling_plot, | |
| price_plot, pnl_plot, hold_plot, mae_plot, costs_note] | |
| compare_out = [lb_controls, podium, lb_table, lb_overlay, | |
| lb_scatter, lb_meta] | |
| view_out = [left_col, compare_view, backtest_view, empty_html, results_view] | |
| def views(st, rec): | |
| """Which zones are visible, given the tab and whether a run exists.""" | |
| backtest = st.get("tab") == "backtest" | |
| has_run = rec is not None | |
| return ( | |
| gr.update(visible=backtest), # left strategy builder | |
| gr.update(visible=not backtest), # compare | |
| gr.update(visible=backtest), # backtest | |
| gr.update(visible=backtest and not has_run), # empty state | |
| gr.update(visible=backtest and has_run), # results | |
| ) | |
| def compare_views(st): | |
| f = st["filters"] | |
| pod, table_df, overlay, scatter, meta = CT.build_leaderboard_view( | |
| runtime.get_store(), | |
| assets=f["assets"] or None, timeframes=f["timeframes"] or None, | |
| strategies_=f["strategies"] or None, models=f["models"] or None, | |
| metric_label=st["metric"], min_trades=f["min_trades"], | |
| hide_baselines=f["hide_baselines"], | |
| require_oos=f["require_oos"], top_n=st["topn"]) | |
| controls = shell.compare_controls( | |
| st, metrics=list(CT.RANK_METRICS), | |
| assets=runtime.available_assets(), | |
| timeframes=list(config.TIMEFRAMES), | |
| strategies_=list(catalog.CATALOG_STRATEGIES), | |
| models=sorted(config.SEED_MODELS)) | |
| return (controls, pod, | |
| render_table(table_df, align_right=range(4, 15), | |
| empty="catalog not generated yet"), | |
| overlay, scatter, meta) | |
| def on_action(raw, st, hist, rec, progress=gr.Progress()): | |
| st = copy.deepcopy(st) | |
| st, should_run = apply_action(st, raw) | |
| if not should_run: | |
| return (st, hist, rec, render_left(st), render_top(st, rec), | |
| gr.update(), shell.chart_controls(st), | |
| *(gr.update(),) * 9, gr.update(), | |
| gr.update(), gr.update(), gr.update(), | |
| *compare_views(st), *views(st, rec)) | |
| # Running always means the user is looking at the Backtest tab. | |
| st["tab"] = "backtest" | |
| progress(0.2, desc="Reading cached slices") | |
| try: | |
| progress(0.5, desc="Simulating trades") | |
| rec = runtime.execute(to_request(st)) | |
| except (RunError, ValueError) as exc: | |
| return (st, hist, None, render_left(st), | |
| shell.top_bar(status="RUN FAILED", tone="warn", | |
| tab="backtest", glossary=GLOSSARY, | |
| user=st.get("user"), on_space=ON_SPACE), | |
| shell.note(esc(str(exc)), danger=True), | |
| shell.chart_controls(st), | |
| *(gr.update(),) * 9, gr.update(), gr.update(), | |
| gr.update(), gr.update(), | |
| *compare_views(st), *views(st, None)) | |
| progress(0.85, desc="Building charts") | |
| hist = ([rec] + list(hist))[:40] | |
| return ( | |
| st, hist, rec, render_left(st), render_top(st, rec), | |
| render_stat_band(rec), shell.chart_controls(st), | |
| *build_overview(rec, log_scale=st["log_scale"], cvd=st["cvd"]), | |
| shell.micro(f"{len(rec.result.trades)} total · costs paid " | |
| f"{money(rec.result.costs_paid)} · fills at next bar open"), | |
| render_table(trades_frame(rec), align_right=range(4, 13), | |
| empty="no trades", max_height="520px"), | |
| report_markdown(rec), | |
| charts.equity_curve(rec.result.equity, rec.result.benchmark_equity, | |
| plan=rec.result.plan), | |
| *compare_views(st), *views(st, rec), | |
| ) | |
| action_out = [state, history, current, left_html, top_html, stat_html, | |
| chart_opts, | |
| *overview_out, trades_head, trades_html, report_md, | |
| report_equity, *compare_out, *view_out] | |
| # Two bindings on purpose. Setting a textbox value from JS does not | |
| # always wake Svelte's binding, and a hidden Button click does not | |
| # always survive either; whichever lands first wins, and the nonce | |
| # check in apply_action makes the loser a no-op. | |
| action_trigger.click(on_action, [action_box, state, history, current], | |
| action_out, show_progress="minimal") | |
| action_box.change(on_action, [action_box, state, history, current], | |
| action_out, show_progress="minimal") | |
| current.change( | |
| lambda rec: ( | |
| ("", charts.empty_figure("run a backtest first"), | |
| charts.empty_figure("run a backtest first")) | |
| if rec is None else ( | |
| (lambda g, c: ( | |
| f'<div style="background:var(--bg-panel);border:1px solid ' | |
| f'var(--border-default);padding:12px;margin-bottom:8px">' | |
| f'<div style="font-family:var(--font-styrene);' | |
| f'text-transform:uppercase;font-size:var(--text-sm);' | |
| f'letter-spacing:var(--tracking-wide)">Overfit verdict: ' | |
| f'<span style="color:var(--accent-amber-strong)">{g}</span>' | |
| f'</div>' + "".join( | |
| f'<div style="font-size:var(--text-2xs);' | |
| f'color:var(--text-secondary);line-height:1.7">' | |
| f'{m} {esc(t)}</div>' for m, t in c) + "</div>" | |
| ))(*runtime.overfit_verdict(rec)), | |
| charts.walk_forward_bars(rec.result.windows), | |
| charts.monte_carlo_cone( | |
| charts.monte_carlo_paths(rec.result.trades)), | |
| ) | |
| ), | |
| [current], [verdict_html, wf_plot, mc_plot]) | |
| estimate_btn.click(extension.estimate_ui, | |
| [ext_model, ext_asset, ext_tf, ext_start, ext_end], | |
| [extend_out]) | |
| extend_btn.click( | |
| lambda m, a, t, s_, e: ( | |
| (lambda html, _df: (html, render_table(runtime.coverage_frame())))( | |
| *extension.extend_ui(m, a, t, s_, e))), | |
| [ext_model, ext_asset, ext_tf, ext_start, ext_end], | |
| [extend_out, coverage_html]) | |
| add_btn.click( | |
| lambda f, m: ( | |
| (lambda html, _df: (html, render_table(runtime.coverage_frame())))( | |
| *extension.add_model_ui(f, m))), | |
| [add_family, add_model_id], [add_out, coverage_html]) | |
| def on_load(st): | |
| store = runtime.get_store() | |
| st = refresh_context(copy.deepcopy(st)) | |
| meta = catalog.catalog_meta(store) | |
| meta_html = shell.micro( | |
| f"{meta.get('leaderboard_rows', 0)} combinations · " | |
| f"{meta.get('scorecard_rows', 0)} model slices · " | |
| f"{meta.get('canonical_config', '')}") if meta else \ | |
| shell.micro("catalog not built") | |
| note, acc, cal, bars, score_df = CT.build_models_view(store, "1d") | |
| sig = CT.build_signals_view(store, st["sig_asset"], st["sig_tf"]) | |
| saved = catalog.load_saved_runs(store) | |
| return (st, render_top(st, None), render_left(st), meta_html, | |
| shell.chart_controls(st), *compare_views(st), | |
| note, acc, cal, bars, | |
| render_table(score_df, align_right=range(3, 10), | |
| empty="no scorecard rows"), | |
| sig, | |
| render_table(CT.runs_table([], saved), empty="no saved runs"), | |
| C.coverage_summary(runtime.coverage_map()), | |
| render_table(runtime.coverage_frame(), empty="no coverage"), | |
| extension.status_html()) | |
| demo.load(on_load, [state], | |
| [state, top_html, left_html, catalog_meta, chart_opts, | |
| *compare_out, | |
| models_note, acc_plot, cal_plot, model_bars, score_table, | |
| sig_panel, runs_table, coverage_kpis, coverage_html, | |
| extend_panel]) | |
| # Installing the bridge gets its OWN load with no fn and no outputs. | |
| # Gradio treats a `js=` return value as the output values, so attaching | |
| # it to a load that also has outputs wipes them in the browser -- which | |
| # is exactly what broke the UI while leaving the API working. | |
| demo.load(fn=None, inputs=None, outputs=None, js=BRIDGE_LOAD_JS) | |
| # No OAuth-annotated `demo.load` here, deliberately. A handler that | |
| # takes `gr.OAuthProfile` asks Gradio for an authenticated session; on | |
| # a *load* event that fires every render, an unauthenticated visitor is | |
| # sent to sign in, comes back, fires load again, and is sent to sign in | |
| # again -- the browser reports it as "redirected you too many times". | |
| # | |
| # `gr.LoginButton` already shows signed-in state on its own, and the | |
| # profile is read where it is actually needed: the user-initiated | |
| # extension handlers in src/extension.py. | |
| return demo | |
| demo = build_app() | |
| if __name__ == "__main__": | |
| gr.set_static_paths(paths=theme.static_paths()) | |
| demo.queue(max_size=32).launch( | |
| server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)), | |
| show_api=False) | |