Repository Public
1"""Research discipline for perpetual strategies: the Experiment of spq.research with the perp backtester.23Nothing is relaxed: selection code only sees train + validation bars — and, because funding, premium, mark and4index are columns of the same frames, only the funding settled before the test split —, the test split is5reachable once through `final_test()`, every configuration is a counted trial.67Benchmarks (no parameter, so they may be computed on any split):8- `perp_long_1x`: the perpetuals bought once at 1x, equal weights, and held — price PnL AND funding, same costs;9- `buy_and_hold`: the same assets bought once on the SPOT market and held (spot candles, the spot engine and its10 costs); when no spot data are supplied this key holds the perpetual's own price without funding and the report11 says so;12- `perp_long_1x_<SYM>` per perpetual when there are several.1314Stress windows specific to perps: `funding_stress` picks, inside one split, the windows where funding was the most15positive and the most negative on average (the regimes that hurt a long and a short respectively).16"""17from __future__ import annotations1819import numpy as np20import pandas as pd2122from .engine import BacktestConfig, buy_and_hold23from .metrics import _num, max_drawdown24from .perp_engine import PerpConfig, PerpResult, perp_hold, run_perp_backtest25from .research import Experiment26from .strategy import is_spot_leg272829class PerpExperiment(Experiment):30 _runner = staticmethod(run_perp_backtest)3132 def __init__(self, data, config: PerpConfig, *, fractions=(0.6, 0.2, 0.2), ledger=None,33 spot_data: dict[str, pd.DataFrame] | None = None):34 if not isinstance(config, PerpConfig):35 raise TypeError("PerpExperiment needs a PerpConfig")36 super().__init__(data, config, fractions=fractions, ledger=ledger)37 self.spot_data = dict(spot_data or {})38 self.perps = [s for s in self.symbols if not is_spot_leg(s)]3940 def benchmarks(self, split: str) -> dict:41 start, end = self.splits.window(split)42 data = self.lookahead_data() if split == "test" else self.dev_data()43 perp = {s: data[s] for s in self.perps}44 out = {"perp_long_1x": perp_hold(perp, self.config, start=start, end=end)}45 if len(self.perps) > 1:46 for s in self.perps:47 out[f"perp_long_1x_{s}"] = perp_hold({s: perp[s]}, self.config, start=start, end=end)48 spot = {s: self.spot_data[s] for s in self.perps if s in self.spot_data}49 if len(spot) == len(self.perps):50 common = None51 for df in spot.values():52 common = df.index if common is None else common.intersection(df.index)53 spot = {s: df.loc[common] for s, df in spot.items()}54 cfg = BacktestConfig(self.config.interval, fee_bps=self.config.spot_fee_bps,55 slippage_bps=self.config.spot_slippage_bps, initial_cash=self.config.initial_cash)56 out["buy_and_hold"] = buy_and_hold(spot, cfg, start=start, end=end)57 self.buy_and_hold_is_spot = True58 else:59 no_funding = {s: df.drop(columns=[c for c in ("funding", "premium", "index", "mark") if c in df.columns])60 for s, df in perp.items()}61 out["buy_and_hold"] = perp_hold(no_funding, self.config, start=start, end=end)62 self.buy_and_hold_is_spot = False63 return out646566def funding_regimes(data: dict[str, pd.DataFrame], start, end, *, window_bars: int, top: int = 2) -> list[dict]:67 """Non-overlapping windows of `window_bars` bars inside [start, end) with the highest and the lowest mean68 funding rate (average over the perpetuals that have a `funding` column, per bar)."""69 cols = [df["funding"] for s, df in data.items() if not is_spot_leg(s) and "funding" in df.columns]70 if not cols:71 return []72 f = pd.concat(cols, axis=1).mean(axis=1, skipna=True)73 f = f[(f.index >= start) & (f.index < end)]74 if len(f) < 2 * window_bars or f.notna().sum() < window_bars:75 return []76 roll = f.rolling(window_bars, min_periods=max(2, window_bars // 2)).mean().to_numpy()77 out = []78 for label, sign in (("highest funding", 1.0), ("lowest funding", -1.0)):79 score = np.where(np.isfinite(roll), sign * roll, -np.inf)80 score[: window_bars - 1] = -np.inf81 for _ in range(top):82 e = int(np.argmax(score))83 if not np.isfinite(score[e]):84 break85 out.append({"regime": label, "start": f.index[e - window_bars + 1], "end": f.index[e],86 "mean_rate": float(roll[e])})87 score[max(0, e - window_bars + 1): e + window_bars] = -np.inf88 return out899091def funding_stress(result: PerpResult, benchmark: PerpResult, data, start, end, *, window_bars: int,92 bars_per_year: float, top: int = 2) -> list[dict]:93 """Strategy and benchmark during the most extreme funding regimes of the split (same bars)."""94 out = []95 for w in funding_regimes(data, start, end, window_bars=window_bars, top=top):96 s = result.equity.loc[w["start"]:w["end"]]97 bq = benchmark.equity.loc[w["start"]:w["end"]]98 if len(s) < 2 or not s.iloc[0] > 0:99 continue100 f = result.funding101 paid = f[(f["ts"] >= w["start"]) & (f["ts"] <= w["end"])]["amount"]102 out.append({103 "regime": w["regime"], "window": [w["start"].isoformat(), w["end"].isoformat()],104 "mean_funding_rate_per_bar": _num(w["mean_rate"]),105 "mean_funding_rate_annualised": _num(w["mean_rate"] * bars_per_year),106 "strategy_return": _num(s.iloc[-1] / s.iloc[0] - 1.0),107 "strategy_max_drawdown": _num(max_drawdown(s.to_numpy())),108 "strategy_funding_pnl": _num(float(sum(paid)) if len(paid) else 0.0),109 "benchmark_return": _num(bq.iloc[-1] / bq.iloc[0] - 1.0) if len(bq) > 1 and bq.iloc[0] > 0 else None,110 })111 return out112