src/ui/compare_tab.py

"""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 "
                       "<code>scripts/seed_store.py</code> 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)