src/catalog.py

"""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,
    }