Bit-Trading-Company's picture
CI deploy af874966
e5eae78 verified
Raw
History Blame Contribute Delete
36.5 kB
"""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)