Repository Public
1"""BTC-ETH cointegration spread, market-neutral: long one leg and short the other when the spread is stretched."""23import numpy as np45META = {6 "name": "pairs_coint_perp",7 "version": "1",8 "kind": "perp",9 "universe": ["BTC", "ETH"],10 "interval": "1h",11 "lookback": 800,12 "max_leverage": 2.0,13 "max_gross_exposure": 2.0,14 "params": {"window": 336, "state_bars": 96, "entry_z": 2.0, "exit_z": 0.5, "stop_z": 4.0, "df_t_max": -3.04,15 "gross": 1.0},16}171819def _spread(log_y, log_x):20 """OLS of log Y on log X over the window: (hedge ratio, z-score of the last residual, Dickey-Fuller t)."""21 x = log_x - log_x.mean()22 y = log_y - log_y.mean()23 sxx = float(x @ x)24 if not sxx > 0:25 return 0.0, 0.0, 0.026 beta = float(x @ y) / sxx27 resid = y - beta * x28 sd = resid.std(ddof=2)29 if not sd > 0:30 return beta, 0.0, 0.031 lag, diff = resid[:-1], np.diff(resid)32 denom = float(lag @ lag)33 rho = float(lag @ diff) / denom34 err = diff - rho * lag35 se = np.sqrt(float(err @ err) / (len(diff) - 1) / denom)36 return beta, resid[-1] / sd, (rho / se if se > 0 else 0.0)373839def target_weights(data, params):40 window, state_bars = int(params["window"]), int(params["state_bars"])41 entry_z, exit_z, stop_z = float(params["entry_z"]), float(params["exit_z"]), float(params["stop_z"])42 df_t_max, gross = float(params["df_t_max"]), float(params["gross"])43 weights = {symbol: 0.0 for symbol in data}44 if "BTC" not in data or "ETH" not in data or window < 30 or state_bars < 1 or not exit_z < entry_z < stop_z:45 return weights46 log_x = np.log(data["BTC"]["close"].to_numpy())47 log_y = np.log(data["ETH"]["close"].to_numpy())48 n = len(log_x)49 if n < window + 1:50 return weights51 # Replay the rule over the last `state_bars` bars. At each bar the hedge ratio is re-estimated on the window52 # ENDING at that bar, so no estimate ever uses a later price.53 state, blocked, beta = 0, False, 0.054 for end in range(max(window, n - state_bars + 1), n + 1):55 beta, z, df_t = _spread(log_y[end - window:end], log_x[end - window:end])56 if blocked and abs(z) < exit_z:57 blocked = False58 if state == 0:59 if not blocked and df_t < df_t_max and beta > 0 and entry_z < abs(z) < stop_z:60 state = -1 if z > 0 else 1 # spread rich: short it; spread cheap: long it61 elif abs(z) > stop_z or beta <= 0:62 state, blocked = 0, True63 elif (state == 1 and z > -exit_z) or (state == -1 and z < exit_z):64 state = 065 if state == 0 or not beta > 0:66 return weights67 # long the spread = long ETH, short beta units of BTC per unit of ETH (in notional, as the fit is log-log)68 weights["ETH"] = state * gross / (1.0 + beta)69 weights["BTC"] = -state * gross * beta / (1.0 + beta)70 return weights71