Repository Public
1"""Pairs / cointegration spread, long-only: inside each pair, rotate into the leg that is cheap relative to the other."""23import numpy as np45META = {6 "name": "pairs_rotation",7 "version": "1",8 "universe": ["ETH", "BTC", "LINK", "SOL"],9 "interval": "1h",10 "lookback": 1500,11 "params": {"pairs": [["ETH", "BTC"], ["SOL", "LINK"]], "window": 720, "entry_z": 1.0, "df_t_max": -3.34},12}131415def _spread_state(log_a, log_b):16 """OLS of log A on log B over the window; z-score of the last residual and Dickey-Fuller t of the residuals."""17 x = log_b - log_b.mean()18 y = log_a - log_a.mean()19 sxx = float(x @ x)20 if not sxx > 0:21 return 0.0, 0.022 resid = y - (float(x @ y) / sxx) * x23 sd = resid.std(ddof=2)24 if not sd > 0:25 return 0.0, 0.026 lag, diff = resid[:-1], np.diff(resid)27 denom = float(lag @ lag)28 rho = float(lag @ diff) / denom29 err = diff - rho * lag30 se = np.sqrt(float(err @ err) / (len(diff) - 1) / denom)31 return resid[-1] / sd, (rho / se if se > 0 else 0.0)323334def target_weights(data, params):35 window, entry_z, df_t_max = int(params["window"]), float(params["entry_z"]), float(params["df_t_max"])36 pairs = [(a, b) for a, b in params["pairs"] if a in data and b in data]37 weights = {symbol: 0.0 for symbol in data}38 if not pairs or window < 30:39 return weights40 budget = 1.0 / len(pairs)41 for a, b in pairs:42 ca, cb = data[a]["close"].to_numpy(), data[b]["close"].to_numpy()43 if len(ca) < window or len(cb) < window:44 continue45 z, df_t = _spread_state(np.log(ca[-window:]), np.log(cb[-window:]))46 if df_t < df_t_max and z < -entry_z: # A cheap relative to B, and the spread looks stationary47 weights[a] += budget48 elif df_t < df_t_max and z > entry_z: # B cheap relative to A49 weights[b] += budget50 else: # no evidence either way: hold both legs51 weights[a] += budget / 2.052 weights[b] += budget / 2.053 return weights54