"""The global Compare interface. Answers the question the per-session Comparison tab could not: *of everything this Space has ever computed, what actually worked?* It reads the precomputed catalog rather than re-running backtests, so a view over 168 strategy/model/asset/timeframe combinations opens instantly. Three data sources are merged: * the catalog sweep in the signal store (every combination, one canonical config) * `runs/*.json` -- runs people explicitly saved, across sessions and users * the current browser session's runs Two editorial decisions are deliberate. Low-trade rows are flagged and excluded from the default ranking, because a Sharpe of 4 on 13 trades will otherwise sit on top of the board forever. And naive baselines are shown next to learned models everywhere, because "did this beat doing nothing clever?" is the only question that matters first. """ from __future__ import annotations import pandas as pd from .. import catalog, charts from .format import EM, count, money, num, pct from . import components as C RANK_METRICS = { "OOS Sharpe": "oos_sharpe", "Total return": "total_return", "CAGR": "cagr", "Sharpe (all)": "sharpe", "Sortino": "sortino", "Profit factor": "profit_factor", "Excess vs buy & hold": "excess_vs_hold", "Holdout Sharpe": "holdout_sharpe", "Max drawdown (least bad)": "max_drawdown", } TABLE_COLUMNS = [ ("strategy", "Strategy"), ("model_display", "Model"), ("asset", "Asset"), ("timeframe", "TF"), ("oos_sharpe", "OOS Sharpe"), ("sharpe", "Sharpe"), ("total_return", "Return"), ("cagr", "CAGR"), ("max_drawdown", "Max DD"), ("win_rate", "Win%"), ("profit_factor", "PF"), ("trades", "Trades"), ("excess_vs_hold", "vs Hold"), ("holdout_sharpe", "Holdout"), ("costs_paid", "Costs"), ] def leaderboard_table(df: pd.DataFrame) -> pd.DataFrame: """Format catalog rows for display, flagging thin evidence.""" if df is None or df.empty: return pd.DataFrame(columns=[label for _, label in TABLE_COLUMNS] + ["Note"]) out = pd.DataFrame() for col, label in TABLE_COLUMNS: if col not in df.columns: continue s = df[col] if col in ("total_return", "cagr", "max_drawdown", "excess_vs_hold"): out[label] = s.map(lambda v: pct(v)) elif col == "win_rate": out[label] = s.map(lambda v: pct(v, 0, signed=False)) elif col in ("oos_sharpe", "sharpe", "profit_factor", "holdout_sharpe"): out[label] = s.map(lambda v: num(v)) elif col == "costs_paid": out[label] = s.map(money) elif col == "trades": out[label] = s.map(count) else: out[label] = s.fillna(EM) if "significant" in df.columns: out["Note"] = df["significant"].map( lambda ok: "" if ok else f"< {catalog.MIN_MEANINGFUL_TRADES} trades") return out.reset_index(drop=True) def scorecard_table(sc: pd.DataFrame) -> pd.DataFrame: cols = ["Model", "Asset", "TF", "n", "Coverage", "Cal. error", "Dir. accuracy", "vs momentum", "vs random", "Band width"] if sc is None or sc.empty: return pd.DataFrame(columns=cols) out = pd.DataFrame({ "Model": sc["model_display"], "Asset": sc["asset"], "TF": sc["timeframe"], "n": sc["n"].map(count), "Coverage": sc["coverage_q10_q90"].map(lambda v: pct(v, 1, signed=False)), "Cal. error": sc["calibration_error"].map(lambda v: pct(v, 1)), "Dir. accuracy": sc["directional_accuracy"].map( lambda v: pct(v, 1, signed=False)), "vs momentum": sc["beats_momentum"].map(lambda v: pct(v, 1)), "vs random": sc["beats_random"].map(lambda v: pct(v, 1)), "Band width": sc["band_width_pct"].map(lambda v: pct(v, 2, signed=False)), }) return out.reset_index(drop=True) def runs_table(session_runs, saved: pd.DataFrame) -> pd.DataFrame: """Session runs and store-persisted runs in one list.""" cols = ["When", "Source", "Label", "Strategy", "Asset", "TF", "Return", "Sharpe", "OOS Sharpe", "Max DD", "Trades"] rows = [] for r in (session_runs or []): m = r.result.metrics_all rows.append({ "When": r.created_at.replace("T", " ")[:16], "Source": "this session", "Label": r.label, "Strategy": r.request.strategy, "Asset": r.request.asset, "TF": r.request.timeframe, "Return": pct(m.total_return), "Sharpe": num(m.sharpe), "OOS Sharpe": num(r.result.metrics_oos.sharpe) if r.result.metrics_oos.bars else EM, "Max DD": pct(m.max_drawdown), "Trades": count(m.trade_count), }) if saved is not None and not saved.empty: for _, r in saved.iterrows(): rows.append({ "When": str(r.get("created_at", "")).replace("T", " ")[:16], "Source": "signal store", "Label": r.get("label", ""), "Strategy": r.get("strategy", ""), "Asset": r.get("asset", ""), "TF": r.get("timeframe", ""), "Return": pct(r.get("total_return")), "Sharpe": num(r.get("sharpe")), "OOS Sharpe": num(r.get("oos_sharpe")), "Max DD": pct(r.get("max_drawdown")), "Trades": count(r.get("trades")), }) if not rows: return pd.DataFrame(columns=cols) return pd.DataFrame(rows, columns=cols).sort_values("When", ascending=False) # -------------------------------------------------------------------------- # View builders # -------------------------------------------------------------------------- def build_leaderboard_view(store, *, assets, timeframes, strategies_, models, metric_label, min_trades, hide_baselines, require_oos, top_n): """Everything the Leaderboard sub-tab renders, in one pass.""" lb = catalog.load_leaderboard(store) if lb.empty: empty = charts.empty_figure("catalog not generated yet") return (C.note("The catalog has not been generated yet. Run " "scripts/seed_store.py or extend coverage.", danger=True), pd.DataFrame(), empty, empty, C.micro("no rows")) metric = RANK_METRICS.get(metric_label, "oos_sharpe") filtered = catalog.filter_leaderboard( lb, assets=assets or None, timeframes=timeframes or None, strategies_=strategies_ or None, models=models or None, min_trades=int(min_trades or 0), require_oos=bool(require_oos), hide_baselines=bool(hide_baselines)) # Every metric here ranks high-to-low, drawdown included: drawdowns are # stored as negative numbers, so -0.10 sorts above -0.50 already and # "least bad" falls out of the default ordering. ascending = False top = int(top_n or 25) # Ranking prefers evidence. Rows with too few trades are excluded, and only # if that leaves nothing at all do we fall back to showing thin results -- # an empty board would be less honest than a flagged one. ranked = catalog.rank(filtered, metric, top=top, ascending=ascending, significant_only=True) if ranked.empty: ranked = catalog.rank(filtered, metric, top=top, ascending=ascending) curves = catalog.curves_for(catalog.load_equity_curves(store), list(ranked["key"])[:12]) labels = {} for _, r in ranked.iterrows(): k = r["key"] if k in curves: name = f"{r['strategy']} · {r['asset']} {r['timeframe']}" if r.get("model_slug"): name += f" · {r['model_display']}" labels[name] = curves[k] thin = int((~filtered["significant"]).sum()) if "significant" in filtered else 0 meta = C.micro( f"{len(filtered)} of {len(lb)} rows match · ranked by {metric_label} · " f"showing top {len(ranked)}" + (f" · {thin} rows below {catalog.MIN_MEANINGFUL_TRADES} trades excluded " f"from ranking" if thin else "")) return (C.podium(ranked, metric), leaderboard_table(ranked), charts.multi_return_overlay(labels), charts.risk_return_scatter(filtered), meta) def build_models_view(store, timeframe: str | None): sc = catalog.load_scorecard(store) lb = catalog.load_leaderboard(store) tf = None if timeframe in (None, "", "all") else timeframe sub = sc if tf is None else sc[sc["timeframe"] == tf] return (C.scorecard_note(sub), charts.model_accuracy_bars(sc, timeframe=tf), charts.calibration_scatter(sub), charts.model_leaderboard_bars(lb), scorecard_table(sub.sort_values("directional_accuracy", ascending=False) if not sub.empty else sub)) def build_signals_view(store, asset: str, timeframe: str): sc = catalog.load_scorecard(store) cons = catalog.model_consensus(sc, store, asset, timeframe) verdict = catalog.consensus_verdict(cons) return C.consensus_panel(cons, verdict, asset, timeframe)