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