tests/test_catalog.py

"""Tests for the catalog: the global comparison backend.

Covers the leaderboard sweep, precomputed equity curves, the signal scorecard,
the baseline adapters, and the consensus aggregator.
"""

from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from src import catalog, charts, config
from src.adapters import AdapterError, BaselineAdapter, build_windows, get_adapter
from src.metrics import directional_accuracy
from src.store import SignalStore, validate_signal_frame
from src.ui import compare_tab as CT


def prices_frame(n=500, seed=3, start="2024-01-01"):
    rng = np.random.default_rng(seed)
    close = pd.Series(100 * np.exp(np.cumsum(rng.normal(0.0004, 0.02, n))))
    ts = pd.date_range(start, periods=n, freq="D", tz="UTC")
    return pd.DataFrame({
        "ts": ts, "open": close * 0.999, "high": close * 1.02,
        "low": close * 0.98, "close": close, "volume": 1000.0, "source": "test",
    })


@pytest.fixture
def seeded_store(tmp_path):
    """A small offline store with prices and two baseline models' signals."""
    store = SignalStore(repo_id=None, local_root=tmp_path / "store", offline=True)
    for asset in ("BTC-USD", "ETH-USD"):
        px = prices_frame()
        store.write_prices(asset, "1d", px)
        close = px.set_index("ts")["close"]
        for slug, method in (("baseline-naive", "naive"), ("baseline-drift", "drift")):
            a = BaselineAdapter(f"baseline/{method}", context_len=100).load()
            stamps, wins = build_windows(close, 100)
            frame = a.predict(wins).as_frame(stamps, a.inference_version())
            store.write_signals(slug, f"baseline/{method}", a.resolved_revision,
                                asset, "1d", frame,
                                inference_version=a.inference_version())
    return store


# ---------------------------- baselines ----------------------------


@pytest.mark.parametrize("method", ["naive", "drift", "seasonal"])
def test_baselines_produce_valid_signals(method):
    a = get_adapter("baseline", f"baseline/{method}", context_len=64).load()
    s = pd.Series(np.linspace(100, 200, 300),
                  index=pd.date_range("2024-01-01", periods=300, freq="D", tz="UTC"))
    stamps, wins = build_windows(s, 64)
    out = validate_signal_frame(
        a.predict(wins[:40]).as_frame(stamps[:40], a.inference_version()))
    assert len(out) == 40
    assert ((out["q10"] <= out["q50"]) & (out["q50"] <= out["q90"])).all()


def test_baseline_rejects_unknown_method():
    with pytest.raises(AdapterError, match="unknown baseline method"):
        get_adapter("baseline", "baseline/crystal-ball")


def test_naive_baseline_predicts_the_last_value():
    a = BaselineAdapter("baseline/naive", context_len=10).load()
    assert a.predict(np.arange(1, 11, dtype="float64")[None, :]).q50[0] == pytest.approx(10.0)


def test_drift_baseline_extrapolates_the_window_slope():
    a = BaselineAdapter("baseline/drift", context_len=11).load()
    assert a.predict(np.arange(0, 11, dtype="float64")[None, :]).q50[0] == pytest.approx(11.0)


def test_baselines_are_deterministic():
    a = BaselineAdapter("baseline/drift", context_len=32).load()
    _, wins = build_windows(pd.Series(
        np.random.default_rng(1).normal(100, 5, 200),
        index=pd.date_range("2024-01-01", periods=200, freq="D", tz="UTC")), 32)
    assert np.allclose(a.predict(wins[:20]).q50, a.predict(wins[:20]).q50)


def test_baseline_revision_is_pinned():
    a = BaselineAdapter("baseline/naive").load()
    assert a.resolved_revision.startswith("baseline-")
    assert a.inference_version() != config.PLACEHOLDER_VERSION


def test_flat_forecast_has_undefined_direction_not_zero_accuracy():
    """A random walk never claims a direction, so accuracy is NaN, not 0%."""
    idx = pd.date_range("2024-01-01", periods=80, freq="D", tz="UTC")
    ref = pd.Series(np.linspace(100, 160, 80), index=idx)
    assert pd.isna(directional_accuracy(ref.shift(-1).ffill(), ref.copy(), ref))


def test_chronos_chunking_is_smaller_for_sampling_models():
    bolt = get_adapter("chronos", "amazon/chronos-bolt-small")
    t5 = get_adapter("chronos", "amazon/chronos-t5-small")
    assert bolt.chunk_size > t5.chunk_size
    assert bolt._is_bolt and not t5._is_bolt


# ---------------------------- catalog build ----------------------------


def test_build_produces_all_three_artifacts(seeded_store):
    rep = catalog.build(seeded_store)
    assert rep.rows > 0 and rep.curves > 0 and rep.scorecard_rows > 0
    assert not rep.failed
    assert not catalog.load_leaderboard(seeded_store).empty
    assert not catalog.load_equity_curves(seeded_store).empty
    assert not catalog.load_scorecard(seeded_store).empty


def test_leaderboard_covers_signal_and_non_signal_strategies(seeded_store):
    catalog.build(seeded_store)
    lb = catalog.load_leaderboard(seeded_store)
    assert (lb["model_slug"] == "").any()
    assert (lb["model_slug"] != "").any()


def test_signal_strategies_fan_out_over_every_model(seeded_store):
    catalog.build(seeded_store)
    lb = catalog.load_leaderboard(seeded_store)
    ff = lb[lb["strategy"] == "Chronos Forecast Follower"]
    assert set(ff["model_slug"]) == {"baseline-naive", "baseline-drift"}


def test_every_row_carries_a_unique_key(seeded_store):
    catalog.build(seeded_store)
    assert catalog.load_leaderboard(seeded_store)["key"].is_unique


def test_thin_evidence_is_flagged(seeded_store):
    catalog.build(seeded_store)
    lb = catalog.load_leaderboard(seeded_store)
    assert (lb["significant"] == (lb["trades"] >= catalog.MIN_MEANINGFUL_TRADES)).all()


def test_ranking_excludes_thin_rows_when_asked(seeded_store):
    catalog.build(seeded_store)
    lb = catalog.load_leaderboard(seeded_store)
    strict = catalog.rank(lb, "oos_sharpe", significant_only=True)
    assert (strict["trades"] >= catalog.MIN_MEANINGFUL_TRADES).all()


def test_least_bad_drawdown_ranks_first():
    """Drawdowns are negative, so descending order puts the shallowest on top."""
    df = pd.DataFrame({"max_drawdown": [-0.5, -0.1, -0.3],
                       "significant": True, "trades": 50})
    assert catalog.rank(df, "max_drawdown").iloc[0]["max_drawdown"] == -0.1


def test_drawdown_ranking_in_the_compare_view_puts_shallowest_first(seeded_store):
    catalog.build(seeded_store)
    _p, table, *_ = CT.build_leaderboard_view(
        seeded_store, assets=None, timeframes=None, strategies_=None, models=None,
        metric_label="Max drawdown (least bad)", min_trades=0,
        hide_baselines=False, require_oos=False, top_n=5)
    vals = [float(v.rstrip("%").replace("+", "")) for v in table["Max DD"]]
    assert vals == sorted(vals, reverse=True)


def test_curves_for_returns_named_series(seeded_store):
    catalog.build(seeded_store)
    lb = catalog.load_leaderboard(seeded_store)
    curves = catalog.curves_for(catalog.load_equity_curves(seeded_store),
                                list(lb["key"])[:3])
    assert len(curves) == 3
    for s in curves.values():
        assert isinstance(s, pd.Series) and len(s) > 0


def test_curves_for_unknown_key_is_empty(seeded_store):
    catalog.build(seeded_store)
    assert catalog.curves_for(catalog.load_equity_curves(seeded_store), ["nope"]) == {}


def test_filters_narrow_the_board(seeded_store):
    catalog.build(seeded_store)
    lb = catalog.load_leaderboard(seeded_store)
    assert set(catalog.filter_leaderboard(lb, assets=["BTC-USD"])["asset"]) == {"BTC-USD"}
    assert not catalog.filter_leaderboard(lb, hide_baselines=True)["is_baseline_model"].any()


def test_catalog_meta_records_the_canonical_config(seeded_store):
    catalog.build(seeded_store)
    meta = catalog.catalog_meta(seeded_store)
    assert meta["leaderboard_rows"] > 0
    assert "walk-forward" in meta["canonical_config"]


# ---------------------------- scorecard & consensus ----------------------------


def test_scorecard_labels_baselines(seeded_store):
    catalog.build(seeded_store)
    sc = catalog.load_scorecard(seeded_store)
    assert sc["is_baseline"].all()
    assert {"coverage_q10_q90", "directional_accuracy", "beats_momentum"} <= set(sc.columns)


def test_consensus_lists_every_model_with_a_direction(seeded_store):
    catalog.build(seeded_store)
    cons = catalog.model_consensus(catalog.load_scorecard(seeded_store),
                                   seeded_store, "BTC-USD", "1d")
    assert len(cons) == 2
    assert set(cons["direction"]) <= {"LONG", "SHORT", "FLAT"}
    assert (cons["weight"] >= 0).all()


def test_consensus_verdict_aggregates(seeded_store):
    catalog.build(seeded_store)
    cons = catalog.model_consensus(catalog.load_scorecard(seeded_store),
                                   seeded_store, "BTC-USD", "1d")
    v = catalog.consensus_verdict(cons)
    assert v["direction"] in ("LONG", "SHORT", "FLAT")
    assert 0.0 <= v["confidence"] <= 1.0
    assert v["n_models"] == len(cons)


def test_consensus_on_an_uncovered_slice_is_empty(seeded_store):
    sc = catalog.load_scorecard(seeded_store)
    assert catalog.model_consensus(sc, seeded_store, "SOL-USD", "1d").empty
    assert catalog.consensus_verdict(pd.DataFrame())["direction"] == "NO DATA"


# ---------------------------- compare tab ----------------------------


def test_leaderboard_view_renders_end_to_end(seeded_store):
    catalog.build(seeded_store)
    podium, table, overlay, scatter, meta = CT.build_leaderboard_view(
        seeded_store, assets=None, timeframes=None, strategies_=None, models=None,
        metric_label="OOS Sharpe", min_trades=0, hide_baselines=False,
        require_oos=False, top_n=10)
    assert "bit-podium" in podium
    assert not table.empty
    assert len(overlay.data) > 0 and len(scatter.data) > 0
    assert "rows match" in meta


def test_leaderboard_view_handles_an_empty_catalog(tmp_path):
    empty = SignalStore(repo_id=None, local_root=tmp_path / "s", offline=True)
    podium, table, *_ = CT.build_leaderboard_view(
        empty, assets=None, timeframes=None, strategies_=None, models=None,
        metric_label="OOS Sharpe", min_trades=0, hide_baselines=False,
        require_oos=False, top_n=10)
    assert "not been generated" in podium and table.empty


def test_models_view_renders(seeded_store):
    catalog.build(seeded_store)
    _note, acc, cal, _bars, table = CT.build_models_view(seeded_store, "1d")
    assert acc is not None and cal is not None
    assert not table.empty


def test_signals_view_renders(seeded_store):
    catalog.build(seeded_store)
    html = CT.build_signals_view(seeded_store, "BTC-USD", "1d")
    assert "bit-sig-row" in html or "No model signals" in html


def test_runs_table_merges_sources():
    saved = pd.DataFrame([{
        "created_at": "2026-01-01T00:00:00", "label": "saved one",
        "strategy": "SMA Crossover", "asset": "BTC-USD", "timeframe": "1d",
        "total_return": 0.2, "sharpe": 1.1, "oos_sharpe": 0.9,
        "max_drawdown": -0.1, "trades": 30,
    }])
    out = CT.runs_table([], saved)
    assert len(out) == 1 and out["Source"].iloc[0] == "signal store"


def test_leaderboard_table_flags_thin_rows():
    df = pd.DataFrame([{
        "strategy": "X", "model_display": "—", "asset": "BTC-USD", "timeframe": "1d",
        "oos_sharpe": 4.0, "sharpe": 4.0, "total_return": 1.0, "cagr": 0.5,
        "max_drawdown": -0.1, "win_rate": 0.6, "profit_factor": 2.0, "trades": 3,
        "excess_vs_hold": 0.2, "holdout_sharpe": 1.0, "costs_paid": 10.0,
        "significant": False,
    }])
    assert "trades" in CT.leaderboard_table(df)["Note"].iloc[0]


# ---------------------------- charts ----------------------------


def test_catalog_charts_survive_empty_input():
    import plotly.graph_objects as go

    for fig in (charts.multi_return_overlay({}),
                charts.risk_return_scatter(pd.DataFrame()),
                charts.model_accuracy_bars(pd.DataFrame()),
                charts.calibration_scatter(pd.DataFrame()),
                charts.model_leaderboard_bars(pd.DataFrame())):
        assert isinstance(fig, go.Figure)


def test_overlay_caps_the_number_of_series():
    idx = pd.date_range("2024-01-01", periods=50, freq="D", tz="UTC")
    curves = {f"s{i}": pd.Series(np.linspace(0, 1, 50), index=idx) for i in range(40)}
    assert len(charts.multi_return_overlay(curves, max_series=8).data) == 8