"""The catalog: everything this Space has ever computed, in one place. The Comparison tab used to see only the runs made in the current browser session. That cannot answer the question people actually have -- *which strategy, on which model, on which asset, has ever worked best?* -- because the answer lives across sessions, users and machines. So the catalog is **precomputed into the store** rather than recomputed live: * `comparisons/leaderboard.parquet` -- one row per (strategy x model x asset x timeframe) with the full IS / OOS / holdout metric set, all produced under one fixed canonical config so rows are comparable. * `comparisons/equity_curves.parquet` -- long-format cumulative returns, resampled to daily, so "returns over time" for dozens of algorithms renders instantly instead of re-running dozens of backtests per page load. * `comparisons/signal_scorecard.parquet` -- per-model forecast quality: calibration, directional accuracy against naive baselines, pinball loss. All three are regenerated after any coverage extension, so the catalog never describes a store that no longer exists. Session runs and saved `runs/*.json` are merged on top at read time. """ from __future__ import annotations import json import logging import time from dataclasses import dataclass, field import numpy as np import pandas as pd from . import config, strategies from .engine import BacktestConfig, Costs, Validation, run_backtest from .metrics import ( calibration_coverage, calibration_error, directional_accuracy, pinball_loss, ) from .store import SignalStore log = logging.getLogger("bit.catalog") LEADERBOARD = "comparisons/leaderboard.parquet" EQUITY_CURVES = "comparisons/equity_curves.parquet" SCORECARD = "comparisons/signal_scorecard.parquet" CATALOG_INDEX = "comparisons/catalog.json" # Strategies that take part in the catalog sweep. `Buy & Hold` is included # deliberately: a strategy that cannot beat holding the asset has not earned # its complexity, and the leaderboard should show that side by side. CATALOG_STRATEGIES = ( "Buy & Hold (benchmark)", "SMA Crossover", "RSI Mean Reversion", "Bollinger Breakout", "MACD Momentum", "Sentiment-Gated Momentum", "Chronos Forecast Follower", ) # Every catalog row is produced under exactly this config. Changing it # invalidates every published comparison, which is why it lives in one place. CANONICAL = dict( costs=Costs(enabled=True), validation=Validation(mode="walk_forward", train_months=12, test_months=3, roll_months=3, holdout_months=6), ) CURVE_POINTS_MAX = 800 # Below this many closed trades, a Sharpe ratio is noise dressed as a result. # The leaderboard still shows these rows -- hiding them would be its own kind of # lie -- but flags them and filters them out of the default ranked view. MIN_MEANINGFUL_TRADES = 20 def canonical_config(asset: str, timeframe: str, strategy: str) -> BacktestConfig: return BacktestConfig( asset=asset, timeframe=timeframe, strategy=strategy, params=strategies.defaults_for(strategy), **CANONICAL, ) def entry_key(strategy: str, model_slug: str, asset: str, timeframe: str) -> str: return f"{strategy}|{model_slug or '-'}|{asset}|{timeframe}" # -------------------------------------------------------------------------- # Building # -------------------------------------------------------------------------- @dataclass class BuildReport: rows: int = 0 curves: int = 0 scorecard_rows: int = 0 skipped: list[str] = field(default_factory=list) failed: list[str] = field(default_factory=list) elapsed_s: float = 0.0 def summary(self) -> str: return (f"catalog: {self.rows} leaderboard rows, {self.curves} curves, " f"{self.scorecard_rows} scorecard rows, {len(self.skipped)} skipped, " f"{len(self.failed)} failed in {self.elapsed_s:.1f}s") def _resample_curve(equity: pd.Series) -> pd.DataFrame: """Normalised cumulative return, thinned to a comparable daily series.""" if equity is None or len(equity) < 2: return pd.DataFrame(columns=["ts", "cum_return"]) cum = equity / float(equity.iloc[0]) - 1.0 # Daily is the common denominator across 15m / 1h / 1d runs, so every # series in the overlay shares an x-axis granularity. daily = cum.resample("1D").last().dropna() if len(daily) > CURVE_POINTS_MAX: step = int(np.ceil(len(daily) / CURVE_POINTS_MAX)) daily = daily.iloc[::step] return pd.DataFrame({"ts": daily.index, "cum_return": daily.to_numpy()}) def _scorecard_row(prices: pd.DataFrame, signals: pd.DataFrame, model_slug: str, asset: str, timeframe: str) -> dict | None: """Forecast quality for one model slice, independent of any trading rule.""" if signals.empty or prices.empty: return None close = prices["close"] aligned = signals.reindex(close.index).dropna(subset=["q50"]) if aligned.empty: return None actual_next = close.shift(-1).reindex(aligned.index) valid = actual_next.notna() if valid.sum() < 30: return None idx = valid.index[valid] actual_next = actual_next[valid] q10, q50, q90 = aligned.loc[idx, "q10"], aligned.loc[idx, "q50"], aligned.loc[idx, "q90"] ref = close.reindex(idx) prev_move = close.diff().reindex(idx).fillna(0.0) rng = np.random.default_rng(0) coin = pd.Series(rng.choice([-1.0, 1.0], size=len(idx)), index=idx) cov = calibration_coverage(actual_next, q10, q90) acc = directional_accuracy(actual_next, q50, ref) mom = directional_accuracy(actual_next, ref + prev_move, ref) rnd = directional_accuracy(actual_next, ref + coin, ref) spec = config.SEED_MODELS.get(model_slug) return { "model_slug": model_slug, "model_display": spec.display if spec else model_slug, "family": spec.family if spec else "unknown", "is_baseline": bool(spec and spec.family == "baseline"), "asset": asset, "timeframe": timeframe, "n": int(valid.sum()), "coverage_q10_q90": cov, "calibration_error": calibration_error(cov, 0.80), "directional_accuracy": acc, "beats_momentum": (acc - mom) if pd.notna(acc) and pd.notna(mom) else float("nan"), "beats_random": (acc - rnd) if pd.notna(acc) and pd.notna(rnd) else float("nan"), "baseline_momentum": mom, "baseline_random": rnd, "pinball_q50": pinball_loss(actual_next, q50, 0.50), "band_width_pct": float(((q90 - q10) / ref.abs().clip(lower=1e-9)).mean()), } def build(store: SignalStore, *, write: bool = True, strategies_subset: tuple[str, ...] = CATALOG_STRATEGIES, progress=None) -> BuildReport: """Sweep every covered combination and write the catalog artifacts.""" t0 = time.perf_counter() rep = BuildReport() manifest = store.load_manifest() price_keys = sorted(manifest.prices) rows: list[dict] = [] curves: list[pd.DataFrame] = [] scorecard: list[dict] = [] total = len(price_keys) for n, pk in enumerate(price_keys): asset, timeframe = pk.split("|") if progress: progress((n + 1) / max(total, 1), desc=f"catalog {asset} {timeframe}") prices = store.get_prices(asset, timeframe) if len(prices) < 120: rep.skipped.append(f"{pk}: only {len(prices)} bars") continue bpy = config.bars_per_year(asset, timeframe) models = sorted({e.model_slug for e in manifest.find_signals(asset=asset, timeframe=timeframe)}) signal_cache: dict[str, pd.DataFrame] = {} for m in models: sig = store.get_signals(m, asset, timeframe) signal_cache[m] = sig sc = _scorecard_row(prices, sig, m, asset, timeframe) if sc: scorecard.append(sc) for strategy in strategies_subset: preset = strategies.PRESETS.get(strategy) if preset is None or not preset.available: continue # Signal strategies fan out over models; the rest run once. targets = models if preset.needs_signals else [""] if preset.needs_signals and not models: rep.skipped.append(f"{strategy}|{pk}: no model coverage") continue for model_slug in targets: key = entry_key(strategy, model_slug, asset, timeframe) try: cfg = canonical_config(asset, timeframe, strategy) out = strategies.build(strategy, prices, cfg.params, signal_cache.get(model_slug, pd.DataFrame())) res = run_backtest(prices, out, cfg, bars_per_year=bpy) except Exception as e: log.warning("catalog cell failed %s: %s", key, e) rep.failed.append(f"{key}: {type(e).__name__}") continue a, i, o = res.metrics_all, res.metrics_is, res.metrics_oos h = res.metrics_holdout spec = config.SEED_MODELS.get(model_slug) bench = (float(res.benchmark_equity.iloc[-1] / res.benchmark_equity.iloc[0] - 1.0) if len(res.benchmark_equity) else float("nan")) rows.append({ "key": key, "strategy": strategy, "model_slug": model_slug, "model_display": spec.display if spec else ("—" if not model_slug else model_slug), "is_baseline_model": bool(spec and spec.family == "baseline"), "asset": asset, "timeframe": timeframe, "bars": a.bars, "trades": a.trade_count, "total_return": a.total_return, "cagr": a.cagr, "sharpe": a.sharpe, "sortino": a.sortino, "max_drawdown": a.max_drawdown, "volatility": a.volatility, "win_rate": a.win_rate, "profit_factor": a.profit_factor, "exposure": a.exposure, "costs_paid": res.costs_paid, "is_sharpe": i.sharpe if i.bars else float("nan"), "oos_sharpe": o.sharpe if o.bars else float("nan"), "oos_return": o.total_return if o.bars else float("nan"), "oos_max_drawdown": o.max_drawdown if o.bars else float("nan"), "oos_trades": o.trade_count if o.bars else 0, "has_oos": bool(o.bars), "holdout_sharpe": h.sharpe if h else float("nan"), "holdout_return": h.total_return if h else float("nan"), "has_holdout": h is not None, "oos_is_ratio": (o.sharpe / i.sharpe) if (i.bars and i.sharpe) else float("nan"), "benchmark_return": bench, "excess_vs_hold": a.total_return - bench if pd.notna(bench) else float("nan"), "significant": bool(a.trade_count >= MIN_MEANINGFUL_TRADES), "wf_windows": len(res.windows), "wf_positive": sum(1 for w in res.windows if w.metrics.total_return > 0), "generated_at": pd.Timestamp.now(tz="UTC").isoformat(), }) curve = _resample_curve(res.equity) if not curve.empty: curve["key"] = key curve["strategy"] = strategy curve["asset"] = asset curve["timeframe"] = timeframe curve["model_slug"] = model_slug curves.append(curve) lb = pd.DataFrame(rows) ec = pd.concat(curves, ignore_index=True) if curves else pd.DataFrame( columns=["ts", "cum_return", "key", "strategy", "asset", "timeframe", "model_slug"]) sc = pd.DataFrame(scorecard) rep.rows, rep.curves, rep.scorecard_rows = len(lb), len(curves), len(sc) rep.elapsed_s = time.perf_counter() - t0 if write: if not lb.empty: store.write_table(LEADERBOARD, lb) if not ec.empty: store.write_table(EQUITY_CURVES, ec) if not sc.empty: store.write_table(SCORECARD, sc) store.write_json(CATALOG_INDEX, { "generated_at": pd.Timestamp.now(tz="UTC").isoformat(), "leaderboard_rows": len(lb), "equity_curve_series": len(curves), "scorecard_rows": len(sc), "strategies": list(strategies_subset), "canonical_config": "costs on; walk-forward 12/3/3; 6mo locked holdout", "skipped": rep.skipped[:50], "failed": rep.failed[:50], }) return rep # -------------------------------------------------------------------------- # Reading # -------------------------------------------------------------------------- def _read(store: SignalStore, path: str) -> pd.DataFrame: df = store.read_parquet(path) return df if df is not None else pd.DataFrame() def load_leaderboard(store: SignalStore) -> pd.DataFrame: return _read(store, LEADERBOARD) def load_equity_curves(store: SignalStore) -> pd.DataFrame: df = _read(store, EQUITY_CURVES) if not df.empty: df["ts"] = pd.to_datetime(df["ts"], utc=True) return df def load_scorecard(store: SignalStore) -> pd.DataFrame: return _read(store, SCORECARD) def catalog_meta(store: SignalStore) -> dict: p = store._fetch(CATALOG_INDEX) if p is None: return {} try: return json.loads(p.read_text()) except Exception: return {} def load_saved_runs(store: SignalStore) -> pd.DataFrame: """Runs persisted to `runs/` -- across every session and every user.""" rows = [] try: from huggingface_hub import HfApi if store.offline: import pathlib files = [f"runs/{p.name}" for p in (store.local_root / "runs").glob("*.json")] \ if (store.local_root / "runs").exists() else [] else: api = HfApi(token=store.token) files = [f for f in api.list_repo_files( store.repo_id, repo_type=config.STORE_REPO_TYPE) if f.startswith("runs/") and f.endswith(".json")] except Exception as e: log.warning("could not list saved runs: %s", e) return pd.DataFrame() for f in files[:500]: p = store._fetch(f) if p is None: continue try: d = json.loads(p.read_text()) except Exception: continue cfg = d.get("config", {}) m = (d.get("metrics") or {}) allm, oos = m.get("all") or {}, m.get("oos") or {} rows.append({ "run_id": d.get("run_id", ""), "label": d.get("label", ""), "created_at": d.get("created_at", ""), "strategy": cfg.get("strategy", ""), "asset": cfg.get("asset", ""), "timeframe": cfg.get("timeframe", ""), "model_slug": cfg.get("model_slug", ""), "total_return": allm.get("total_return", float("nan")), "sharpe": allm.get("sharpe", float("nan")), "oos_sharpe": oos.get("sharpe", float("nan")), "max_drawdown": allm.get("max_drawdown", float("nan")), "trades": allm.get("trade_count", 0), "costs_paid": allm.get("costs_paid", float("nan")), "share_token": d.get("share_token", ""), "source": "saved run", }) return pd.DataFrame(rows) # -------------------------------------------------------------------------- # Views the UI asks for # -------------------------------------------------------------------------- def rank(df: pd.DataFrame, by: str = "oos_sharpe", top: int | None = None, ascending: bool = False, significant_only: bool = False) -> pd.DataFrame: """Sort the leaderboard. `significant_only` drops rows with too few trades to mean anything. A Sharpe of 4 on 13 trades outranks everything real if you let it, so the default ranked view uses this and says so. """ if df.empty or by not in df.columns: return df out = df if significant_only and "significant" in out.columns: out = out[out["significant"]] out = out.sort_values(by, ascending=ascending, na_position="last") return out.head(top) if top else out def filter_leaderboard(df: pd.DataFrame, *, assets=None, timeframes=None, strategies_=None, models=None, min_trades: int = 0, require_oos: bool = False, hide_baselines: bool = False) -> pd.DataFrame: if df.empty: return df out = df if assets: out = out[out["asset"].isin(assets)] if timeframes: out = out[out["timeframe"].isin(timeframes)] if strategies_: out = out[out["strategy"].isin(strategies_)] if models: out = out[out["model_slug"].isin(models) | (out["model_slug"] == "")] if min_trades: out = out[out["trades"] >= min_trades] if require_oos and "has_oos" in out.columns: out = out[out["has_oos"]] if hide_baselines and "is_baseline_model" in out.columns: out = out[~out["is_baseline_model"]] return out def curves_for(curves: pd.DataFrame, keys: list[str]) -> dict[str, pd.Series]: """Pull named cumulative-return series out of the stored long frame.""" if curves.empty or not keys: return {} sub = curves[curves["key"].isin(keys)] out: dict[str, pd.Series] = {} for key, g in sub.groupby("key"): g = g.sort_values("ts") out[str(key)] = pd.Series(g["cum_return"].to_numpy(), index=pd.DatetimeIndex(g["ts"])) return out def model_consensus(scorecard: pd.DataFrame, store: SignalStore, asset: str, timeframe: str) -> pd.DataFrame: """Every model's latest forecast for one asset, plus how much to trust it. This is the Signal Aggregator view: each model's most recent direction and implied move, weighted by how well that model has actually been calibrated on this slice, so a confident-but-miscalibrated model does not dominate. """ manifest = store.load_manifest() models = sorted({e.model_slug for e in manifest.find_signals(asset=asset, timeframe=timeframe)}) prices = store.get_prices(asset, timeframe) if prices.empty or not models: return pd.DataFrame() last_close = float(prices["close"].iloc[-1]) rows = [] for m in models: sig = store.get_signals(m, asset, timeframe) if sig.empty: continue last = sig.iloc[-1] ref = float(prices["close"].reindex([sig.index[-1]]).iloc[0]) \ if sig.index[-1] in prices.index else last_close edge = float(last["q50"]) / ref - 1.0 sc = scorecard[(scorecard.get("model_slug") == m) & (scorecard.get("asset") == asset) & (scorecard.get("timeframe") == timeframe)] \ if not scorecard.empty else pd.DataFrame() acc = float(sc["directional_accuracy"].iloc[0]) if len(sc) else float("nan") cal_err = float(sc["calibration_error"].iloc[0]) if len(sc) else float("nan") # Weight = how much better than a coin flip this model has been here. weight = max(0.0, (acc - 0.5) * 2) if pd.notna(acc) else 0.0 spec = config.SEED_MODELS.get(m) rows.append({ "model_slug": m, "model": spec.display if spec else m, "is_baseline": bool(spec and spec.family == "baseline"), "as_of": sig.index[-1], "reference_px": ref, "q10": float(last["q10"]), "q50": float(last["q50"]), "q90": float(last["q90"]), "edge": edge, "direction": "LONG" if edge > 0.0005 else ("SHORT" if edge < -0.0005 else "FLAT"), "band_width": (float(last["q90"]) - float(last["q10"])) / max(abs(ref), 1e-9), "directional_accuracy": acc, "calibration_error": cal_err, "weight": weight, }) df = pd.DataFrame(rows) if df.empty: return df return df.sort_values("weight", ascending=False).reset_index(drop=True) def consensus_verdict(consensus: pd.DataFrame) -> dict: """Weighted aggregate of the per-model directions.""" if consensus.empty: return {"direction": "NO DATA", "confidence": 0.0, "edge": 0.0, "n_models": 0, "agree": 0} w = consensus["weight"].fillna(0.0) if w.sum() <= 0: w = pd.Series(1.0, index=consensus.index) edge = float((consensus["edge"] * w).sum() / w.sum()) direction = "LONG" if edge > 0.0005 else ("SHORT" if edge < -0.0005 else "FLAT") agree = int((consensus["direction"] == direction).sum()) return { "direction": direction, "edge": edge, "confidence": float(agree / len(consensus)), "n_models": int(len(consensus)), "agree": agree, }