Repository Public
1"""Every strategy of the library returns valid weights on synthetic data."""2from __future__ import annotations34import json5import subprocess6import sys78import numpy as np9import pandas as pd10import pytest1112from conftest import LIBRARY, ROOT, make_data13from spq.engine import BacktestConfig, run_backtest14from spq.strategy import load_strategy, sha256_file, validate_weights1516EXPECTED = {"ts_momentum", "ma_trend", "mean_reversion", "donchian_breakout", "vol_target_trend", "xs_momentum",17 "pairs_rotation", "garch_regime"}18IDS = [f"{d.parent.name}/{d.name}" for d in LIBRARY]19# universe-generic versions (META['universe'] == ['*']): one per rule that makes sense for any set of symbols20GENERIC = {"ts_momentum", "ma_trend", "donchian_breakout", "vol_target_trend", "mean_reversion", "xs_momentum"}21GENERIC_SYMBOLS = ["ETH", "UNI", "LINK", "POL"] # what an agent could pick; nothing in a generic strategy names them222324def symbols_of(strategy):25 return GENERIC_SYMBOLS if strategy.generic else strategy.universe262728def synthetic(strategy, n, seed):29 """Random walks with different drifts so that trends, breakouts and rank changes all occur."""30 drifts = [0.0004, -0.0003, 0.0001, 0.0]31 out = {}32 for i, sym in enumerate(symbols_of(strategy)):33 out.update(make_data(n, (sym,), seed=seed + 31 * i, drift=drifts[i % 4], sigma=0.012, p0=50.0 * (i + 1)))34 return out353637def test_the_library_is_complete():38 assert {d.parent.name for d in LIBRARY} == EXPECTED39 assert {d.parent.name for d in LIBRARY if d.name == "3"} == GENERIC40 for d in LIBRARY:41 readme = (d / "README.md").read_text()42 assert "**Hypothesis.**" in readme and "**What would refute it.**" in readme, d434445@pytest.mark.parametrize("directory", LIBRARY, ids=IDS)46def test_library_strategy_contract(directory):47 s = load_strategy(directory) # strict: passes the static purity check48 assert s.static_problems == [] and s.sha256 == sha256_file(directory / "strategy.py")49 assert s.name == directory.parent.name and s.meta["version"] == directory.name50 assert s.meta["interval"] == "1h" and s.universe in (["ETH", "BTC", "LINK", "SOL"], ["ETH", "LINK", "AAVE"], ["*"])51 assert s.generic == (directory.name == "3")52 universe = symbols_of(s)5354 full = synthetic(s, s.lookback + 400, seed=5)55 stride = 97 if s.name != "garch_regime" else 19956 seen_invested = False57 for end in range(s.lookback, s.lookback + 400, stride):58 view = {sym: df.iloc[end - s.lookback:end] for sym, df in full.items()}59 before = {sym: df.copy() for sym, df in view.items()}60 raw = s.raw(view)61 w = validate_weights(raw, universe) # dict, finite, each in [0, 1], known symbols, sum <= 162 assert set(w) == set(universe) and all(0.0 <= x <= 1.0 for x in w.values())63 assert sum(w.values()) <= 1.0 + 1e-964 assert s.raw(view) == raw # pure: same inputs, same output65 for sym in view:66 pd.testing.assert_frame_equal(view[sym], before[sym]) # inputs are not modified67 seen_invested |= sum(w.values()) > 068 assert seen_invested, "the strategy never invested on trending synthetic data"6970 # degenerate inputs: too little history, a constant price, a subset of the universe71 tiny = {sym: df.iloc[:3] for sym, df in full.items()}72 assert sum(validate_weights(s.raw(tiny), universe).values()) <= 1.073 flat = {sym: df.assign(open=10.0, high=10.0, low=10.0, close=10.0).iloc[:s.lookback] for sym, df in full.items()}74 assert sum(validate_weights(s.raw(flat), universe).values()) <= 1.075 two = {sym: full[sym].iloc[:s.lookback] for sym in universe[:2]}76 assert sum(validate_weights(s.raw(two), universe[:2]).values()) <= 1.0 + 1e-9777879@pytest.mark.parametrize("directory", LIBRARY, ids=IDS)80def test_library_strategy_runs_in_the_backtester_and_the_signal_runner(directory, tmp_path):81 s = load_strategy(directory)82 every = 24 if s.name == "garch_regime" else 183 n = s.lookback + (24 * 12 if every == 24 else 300)84 data = synthetic(s, n, seed=9)85 res = run_backtest(s, data, BacktestConfig("1h", rebalance_threshold=0.05, decision_every=every))86 assert len(res.decisions) > 0 and np.isfinite(res.equity.to_numpy()).all() and (res.equity > 0).all()87 assert (res.cash >= -1e-9).all() and res.exposure.between(0, 1 + 1e-9).all()88 assert (res.decisions.sum(axis=1) <= 1 + 1e-9).all() and (res.decisions.to_numpy() >= 0).all()8990 # the engine's entry point gives the same answer as the last decision of the backtest91 payload = {"candles": {sym: [[ts.strftime("%Y-%m-%dT%H:%M:%SZ"), *map(float, row)] for ts, row in92 zip(df.index[-s.lookback:], df.to_numpy()[-s.lookback:])] for sym, df in data.items()}}93 f = tmp_path / "in.json"94 f.write_text(json.dumps(payload))95 p = subprocess.run([sys.executable, "-m", "spq.signal", "--strategy", str(directory), "--sha256", s.sha256,96 "--input", str(f)], capture_output=True, text=True, cwd=ROOT, timeout=300)97 assert p.returncode == 0, p.stderr98 out = json.loads(p.stdout)99 assert out["strategy_sha256"] == s.sha256 and out["as_of"] == data["ETH"].index[-1].strftime("%Y-%m-%dT%H:%M:%SZ")100 last = res.decisions.iloc[-1]101 if res.decisions.index[-1] == data["ETH"].index[-1]:102 assert out["weights"] == pytest.approx(last.to_dict(), abs=1e-12)103