Repository Public
1"""Every perp strategy of the library: contract, purity, valid weights, behaviour on hand-made situations,2and agreement between the backtester and the signal runner."""3from __future__ import annotations45import json6import subprocess7import sys89import numpy as np10import pandas as pd11import pytest1213from conftest import LIBRARY, PERP_LIBRARY, ROOT, make_data14from spq.perp_engine import PerpConfig, run_perp_backtest15from spq.strategy import is_spot_leg, load_strategy, sha256_file, validate_perp_weights1617EXPECTED = {"ts_momentum_perp", "ma_trend_perp", "zscore_reversion_perp", "pairs_coint_perp", "funding_carry_perp",18 "funding_carry_neutral_perp", "vol_breakout_perp", "har_rv_perp"}19IDS = [f"{d.parent.name}/{d.name}" for d in PERP_LIBRARY]20PERPS = ["BTC", "ETH", "SOL", "HYPE"]212223def synthetic(strategy, n, seed, funding=1.25e-5):24 """Random walks with different drifts; a spot leg tracks its perpetual closely; funding has high episodes."""25 drifts = [0.0005, -0.0004, 0.0002, 0.0]26 out = {}27 rng = np.random.default_rng(seed)28 for i, sym in enumerate(s for s in strategy.universe if not is_spot_leg(s)):29 df = make_data(n, (sym,), seed=seed + 31 * i, drift=drifts[i % 4], sigma=0.012, p0=50.0 * (i + 1))[sym]30 f = np.full(n, funding) + rng.normal(0, 2e-6, n)31 for start in range(200 + 150 * i, n, 900): # episodes of persistently high funding32 f[start:start + 250] += 6e-533 out[sym] = df.assign(funding=f, premium=f * 8)34 if strategy.name == "pairs_coint_perp": # a cointegrated pair: ETH follows BTC plus a stationary spread35 noise = np.zeros(n)36 shocks = rng.normal(0, 0.004, n)37 for t in range(1, n):38 noise[t] = 0.9 * noise[t - 1] + shocks[t]39 btc = out["BTC"]40 factor = 0.05 * np.exp(1.1 * np.log(btc["close"].to_numpy() / 50.0) + noise) * 50.0 / btc["close"].to_numpy()41 eth = btc[["open", "high", "low", "close"]].mul(factor, axis=0)42 eth["open"] = np.concatenate([[eth["close"].iloc[0]], eth["close"].to_numpy()[:-1]])43 eth["high"] = eth[["open", "close"]].max(axis=1) * 1.00144 eth["low"] = eth[["open", "close"]].min(axis=1) * 0.99945 out["ETH"] = eth.assign(volume=1.0, funding=out["ETH"]["funding"], premium=out["ETH"]["premium"])46 for sym in strategy.universe:47 if is_spot_leg(sym):48 base = out[sym[:-5]]49 out[sym] = base[["open", "high", "low", "close", "volume"]] * 1.000450 return {s: out[s] for s in strategy.universe}515253def test_the_perp_library_is_complete_and_the_spot_library_is_untouched():54 assert {d.parent.name for d in PERP_LIBRARY} == EXPECTED55 assert not any(d.parent.name.endswith("_perp") for d in LIBRARY)56 assert {d.parent.name for d in LIBRARY} == {"ts_momentum", "ma_trend", "mean_reversion", "donchian_breakout",57 "vol_target_trend", "xs_momentum", "pairs_rotation", "garch_regime"}58 assert all(load_strategy(d).kind == "spot" for d in LIBRARY)59 for d in PERP_LIBRARY:60 readme = (d / "README.md").read_text()61 assert "**Hypothesis.**" in readme and "**Mechanism.**" in readme and "**What would refute it.**" in readme, d62 assert "before any backtest" in readme, d63 assert "directional" in (ROOT / "strategies/funding_carry_perp/1/README.md").read_text()64 lab = " ".join((ROOT / "README.md").read_text().lower().split())65 assert "order-book" in lab and "cannot be backtested" in lab # microstructure strategies are not faked666768@pytest.mark.parametrize("directory", PERP_LIBRARY, ids=IDS)69def test_perp_strategy_contract(directory):70 s = load_strategy(directory) # strict: passes the static purity check71 assert s.kind == "perp" and s.static_problems == [] and s.sha256 == sha256_file(directory / "strategy.py")72 assert s.name == directory.parent.name and s.meta["version"] == directory.name and s.meta["interval"] == "1h"73 perps = [x for x in s.universe if not is_spot_leg(x)]74 assert set(perps) <= set(PERPS) and s.meta["max_leverage"] <= 2.0 and s.lookback <= 120075 lev, cap = s.meta["max_leverage"], s.meta["max_gross_exposure"]7677 full = synthetic(s, s.lookback + 600, seed=5)78 active = False79 for end in range(s.lookback, s.lookback + 600, 67):80 view = {sym: df.iloc[end - s.lookback:end] for sym, df in full.items()}81 before = {sym: df.copy() for sym, df in view.items()}82 raw = s.raw(view)83 w = validate_perp_weights(raw, s.universe, lev, cap)84 assert set(w) == set(s.universe) and sum(abs(x) for x in w.values()) <= cap + 1e-985 assert all(0 <= w[x] <= 1 for x in s.universe if is_spot_leg(x))86 assert s.raw(view) == raw # pure: same inputs, same output87 for sym in view:88 pd.testing.assert_frame_equal(view[sym], before[sym]) # inputs are not modified89 active |= any(x != 0 for x in w.values())90 assert active, "the strategy never took a position on the synthetic data"9192 # degenerate inputs never raise and never produce an invalid allocation93 tiny = {sym: df.iloc[:3] for sym, df in full.items()}94 assert all(x == 0 for x in validate_perp_weights(s.raw(tiny), s.universe, lev, cap).values())95 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()}96 validate_perp_weights(s.raw(flat), s.universe, lev, cap)97 no_funding = {sym: df.drop(columns=[c for c in ("funding", "premium") if c in df.columns]).iloc[:s.lookback]98 for sym, df in full.items()}99 validate_perp_weights(s.raw(no_funding), s.universe, lev, cap)100 nan_funding = {sym: (df.assign(funding=np.nan) if "funding" in df.columns else df).iloc[:s.lookback]101 for sym, df in full.items()}102 validate_perp_weights(s.raw(nan_funding), s.universe, lev, cap)103 subset = s.universe[:1] if s.name != "pairs_coint_perp" else s.universe104 validate_perp_weights(s.raw({sym: full[sym].iloc[:s.lookback] for sym in subset}), subset, lev, cap)105 # the default sensitivity grid (every numeric parameter halved and doubled) stays inside the caps106 from spq.research import one_at_a_time_grid107108 view = {sym: df.iloc[-s.lookback:] for sym, df in full.items()}109 for params in one_at_a_time_grid(s.params):110 validate_perp_weights(s.raw(view, params), s.universe, lev, cap)111112113@pytest.mark.parametrize("directory", PERP_LIBRARY, ids=IDS)114def test_perp_strategy_runs_in_the_backtester_and_the_signal_runner(directory, tmp_path):115 s = load_strategy(directory)116 every = 24 if s.name == "har_rv_perp" else (4 if s.name == "pairs_coint_perp" else 1)117 n = s.lookback + (24 * 12 if every == 24 else 300)118 data = synthetic(s, n, seed=9)119 res = run_perp_backtest(s, data, PerpConfig("1h", rebalance_threshold=0.05, decision_every=every))120 assert len(res.decisions) > 0 and np.isfinite(res.equity.to_numpy()).all() and (res.equity > 0).all()121 assert res.ledger["reconciles"] and not res.wiped_out and res.liquidations == []122 assert res.exposure.max() <= s.meta["max_gross_exposure"] * 1.25 # prices move between two rebalancings123124 perps = [x for x in s.universe if not is_spot_leg(x)]125 stamp = lambda ts: ts.strftime("%Y-%m-%dT%H:%M:%SZ") # noqa: E731126 payload = {127 "candles": {sym: [[stamp(ts), *map(float, row)] for ts, row in128 zip(df.index[-s.lookback:], df[["open", "high", "low", "close", "volume"]].to_numpy()[-s.lookback:])]129 for sym, df in data.items()},130 "funding": {sym: [[stamp(ts), float(f)] for ts, f in zip(data[sym].index[-s.lookback:],131 data[sym]["funding"].to_numpy()[-s.lookback:])]132 for sym in perps}}133 f = tmp_path / "in.json"134 f.write_text(json.dumps(payload))135 cmd = [sys.executable, "-m", "spq.signal", "--strategy", str(directory), "--sha256", s.sha256, "--input", str(f)]136 p = subprocess.run(cmd + ["--contract", "2"], capture_output=True, text=True, cwd=ROOT, timeout=300)137 assert p.returncode == 0, p.stderr138 out = json.loads(p.stdout)139 last_bar = data[perps[0]].index[-1]140 assert out["schema_version"] == 2 and out["strategy_sha256"] == s.sha256 and out["as_of"] == stamp(last_bar)141 if res.decisions.index[-1] == last_bar: # the engine's entry point = the backtest's last decision142 assert out["weights"] == pytest.approx(res.decisions.iloc[-1].to_dict(), abs=1e-12)143 # a caller that only knows the spot contract gets nothing144 p = subprocess.run(cmd, capture_output=True, text=True, cwd=ROOT, timeout=300)145 assert p.returncode == 2 and p.stdout == ""146 p = subprocess.run([sys.executable, "-m", "spq.describe", "--strategy", str(directory)], capture_output=True,147 text=True, cwd=ROOT, timeout=120)148 assert p.returncode == 4 and p.stdout == ""149150151def load(name):152 return load_strategy(ROOT / "strategies" / name / "1")153154155def series(closes, start="2024-01-01"):156 c = np.asarray(closes, float)157 idx = pd.date_range(start, periods=len(c), freq="1h", tz="UTC", name="ts")158 return pd.DataFrame({"open": c, "high": c, "low": c, "close": c, "volume": 1.0, "funding": 1.25e-5}, index=idx)159160161def test_momentum_goes_long_what_rose_and_short_what_fell():162 s = load("ts_momentum_perp")163 up, down = np.linspace(100, 150, 400), np.linspace(100, 60, 400)164 w = s({"BTC": series(up), "ETH": series(down), "SOL": series(up), "HYPE": series(down)})165 assert w == {"BTC": 0.25, "ETH": -0.25, "SOL": 0.25, "HYPE": -0.25}166167168def test_ma_trend_sizes_inversely_to_volatility():169 s = load("ma_trend_perp")170 rng = np.random.default_rng(0)171 calm = 100 * np.exp(np.cumsum(0.0008 + rng.normal(0, 0.002, 400)))172 wild = 100 * np.exp(np.cumsum(-0.0030 + rng.normal(0, 0.02, 400)))173 w = s({"BTC": series(calm), "ETH": series(wild), "SOL": series(calm), "HYPE": series(wild)})174 assert w["BTC"] > 0 > w["ETH"] and abs(w["BTC"]) > 3 * abs(w["ETH"])175 assert w["BTC"] == 0.5 # capped at max_weight: 0.15 / 0.19 > 0.5176 vol = np.diff(np.log(wild[-169:])).std(ddof=1) * np.sqrt(8760)177 assert w["ETH"] == pytest.approx(-0.6 / 4 / vol)178179180def test_zscore_reversion_fades_a_spike_exits_at_the_mean_and_stops_on_a_runaway():181 s = load("zscore_reversion_perp")182 rng = np.random.default_rng(1)183 base = 100 + rng.normal(0, 0.2, 300)184185 def w(closes):186 return s({"BTC": series(closes), "ETH": series(base), "SOL": series(base), "HYPE": series(base)})["BTC"]187188 spike = base.copy()189 spike[-1] += 0.2 * 3 # about 3 standard deviations above the mean190 assert w(spike) == -0.25 # fade it: short191 dip = base.copy()192 dip[-1] -= 0.2 * 3193 assert w(dip) == 0.25194 back = spike.copy()195 back = np.append(back[1:], 100.0) # next bar: back at the mean -> flat196 assert w(back) == 0.0197 runaway = base.copy()198 runaway[-6:] += np.array([0.6, 0.9, 1.3, 2.0, 3.5, 6.0]) # keeps going: |z| > 4 -> stopped out199 assert w(runaway) == 0.0200 assert w(base) == 0.0201202203def test_pairs_trades_a_stretched_stationary_spread_market_neutral():204 s = load("pairs_coint_perp")205 rng = np.random.default_rng(2)206 n = 800207 common = np.cumsum(rng.normal(0, 0.01, n))208 noise = np.zeros(n)209 for t in range(1, n):210 noise[t] = 0.7 * noise[t - 1] + rng.normal(0, 0.002) # stationary spread211 btc = 60000 * np.exp(common)212 eth = 3000 * np.exp(1.2 * common + noise)213 calm = s({"BTC": series(btc), "ETH": series(eth)})214 stretched = eth.copy()215 stretched[-1] *= np.exp(3.0 * noise[-336:].std()) # ETH suddenly rich against BTC216 w = s({"BTC": series(btc), "ETH": series(stretched)})217 assert w["ETH"] < 0 < w["BTC"] # short the rich leg, long the other218 beta = w["BTC"] / -w["ETH"]219 assert 1.0 < beta < 1.4 and abs(w["ETH"]) + abs(w["BTC"]) == pytest.approx(1.0)220 assert abs(calm["ETH"]) + abs(calm["BTC"]) in (0.0, pytest.approx(1.0))221 # two independent random walks: the spread is not stationary, nothing is traded222 other = 3000 * np.exp(np.cumsum(rng.normal(0, 0.01, n)))223 w = s({"BTC": series(btc), "ETH": series(other)})224 assert w == {"BTC": 0.0, "ETH": 0.0}225226227def test_funding_carry_shorts_only_persistent_large_funding_and_the_neutral_version_hedges_it():228 s, neutral = load("funding_carry_perp"), load("funding_carry_neutral_perp")229 n = 1100230 flat = np.full(n, 100.0)231 rng = np.random.default_rng(3)232233 def frame(rate):234 return series(flat).assign(funding=rate)235236 floor = np.full(n, 1.25e-5) + rng.normal(0, 1e-7, n) # the venue's floor rate: 10.95 % a year237 high = floor.copy()238 high[-72:] = 6e-5 # 52 % a year for three days239 negative = floor.copy()240 negative[-72:] = -6e-5241 erratic = floor.copy()242 erratic[-72:] = np.where(np.arange(72) % 2 == 0, 3e-4, -2e-4) # large on average, but not persistent243 data = {"BTC": frame(high), "ETH": frame(floor), "SOL": frame(negative), "HYPE": frame(erratic)}244 w = s(data)245 assert w == {"BTC": -0.25, "ETH": 0.0, "SOL": 0.0, "HYPE": 0.0} # short the payer, at full size (z >> 2)246 spot = {f"{k}.SPOT": series(flat).drop(columns=["funding"]) for k in data}247 w = neutral({**data, **spot})248 assert w["BTC"] == pytest.approx(-0.1125) and w["BTC.SPOT"] == pytest.approx(0.1125) # 0.9 / (2 x 4)249 assert sum(w.values()) == pytest.approx(0.0) and all(w[k] == 0 for k in ("ETH", "SOL", "HYPE", "ETH.SPOT"))250 # the decision only reads settlements up to the decision bar: appending a future one changes nothing251 later = {k: pd.concat([df, df.iloc[[-1]].set_axis(df.index[[-1]] + pd.Timedelta(hours=1)).assign(funding=-1.0)])252 for k, df in data.items()}253 assert s({k: df.iloc[:-1] for k, df in later.items()}) == s(data)254255256def test_vol_breakout_follows_a_large_move_for_a_while_then_stands_down():257 s = load("vol_breakout_perp")258 quiet = 100 * (1 + 0.002 * (-1.0) ** np.arange(500)) # +/- 0.2 % every bar, no 24-hour move at all259260 def w(closes):261 return s({"BTC": series(closes), "ETH": series(quiet), "SOL": series(quiet), "HYPE": series(quiet)})["BTC"]262263 assert w(quiet) == 0.0264 jump = quiet.copy()265 jump[-10:] *= 1.05 # +5 % in one bar, 10 bars ago266 assert w(jump) == 0.25267 crash = quiet.copy()268 crash[-10:] *= 0.95269 assert w(crash) == -0.25270 old = quiet.copy()271 old[-120:] *= 1.05 # the same jump 120 bars ago: beyond `hold`272 assert w(old) == 0.0273274275def test_har_forecast_sizes_down_in_volatile_regimes():276 s = load("har_rv_perp")277 rng = np.random.default_rng(5)278279 def walk(sig, drift=0.0005):280 return 100 * np.exp(np.cumsum(drift + rng.normal(0, 1, 1200) * sig))281282 calm = walk(np.full(1200, 0.003))283 sig = np.full(1200, 0.003)284 sig[-48:] = 0.03 # volatility explodes over the last two days285 turbulent = walk(sig)286 steady = walk(np.full(1200, 0.012))287 falling = walk(np.full(1200, 0.003), drift=-0.0005)288 w = s({"BTC": series(calm), "ETH": series(turbulent), "SOL": series(steady), "HYPE": series(falling)})289 assert w["BTC"] == 0.5 # low forecast volatility: capped at max_weight290 assert w["ETH"] == 0.0 # forecast far above the window's median: aside291 assert 0 < w["SOL"] < 0.25 # 0.15 / forecast volatility of about 110 %292 assert w["HYPE"] == -0.5 # a falling series is shorted293 # HAR needs enough days: with half the window nothing is forecast294 assert s({"BTC": series(calm[:300])}) == {"BTC": 0.0}295