# ===================================================================== # 量化策略流派 · 官方教学示例（统计套利 / 配对交易 Stat Arb Pairs） # 策略：协整两标的 -> OLS 估计对冲比 -> 价差 Z-Score 均值回归 # 说明：本代码为社区教学示例，由站点整理发布，仅供学习交流。 # 示例代码由站点提供，站点不对代码的安全性、合规性与收益 # 表现作任何明示或默示担保；据此交易风险自负。 # 依赖：numpy（pip install numpy） # 运行：python stat_arb_pairs.py # ===================================================================== import numpy as np def gen_pair(n=1500, seed=5): """生成协整对：y = beta*x + 平稳(均值回归)价差。""" rng = np.random.default_rng(seed) x = np.cumsum(rng.normal(0, 1, n)) + 100 # 随机游走（非平稳） beta = 1.5 spread = np.zeros(n) for t in range(1, n): # OU 过程：平稳、可回归 spread[t] = spread[t - 1] + 0.05 * (0 - spread[t - 1]) + 1.0 * rng.normal() y = beta * x + spread return x, y def ols_hedge(x, y): X = np.vstack([x, np.ones_like(x)]).T coef, _, _, _ = np.linalg.lstsq(X, y, rcond=None) return coef[0] def half_life(spread): s = spread[1:] - spread[:-1] lag = spread[:-1] b = np.polyfit(lag, s, 1)[0] return -np.log(2) / b if b < 0 else np.nan def backtest(x, y, w=20, entry=2.0, exit_z=0.5, cost=0.0005): beta = ols_hedge(x, y) spread = y - beta * x n = len(spread) ma = np.full(n, np.nan) sd = np.full(n, np.nan) for i in range(w - 1, n): seg = spread[i - w + 1:i + 1] ma[i] = seg.mean() sd[i] = seg.std() z = (spread - ma) / sd pos = np.zeros(n) for t in range(w, n): if np.isnan(z[t]): continue if pos[t - 1] == 0: if z[t] > entry: pos[t] = -1 # 价差偏高：卖 y 买 x elif z[t] < -entry: pos[t] = 1 else: pos[t] = pos[t - 1] if abs(z[t]) > exit_z else 0 ret = np.zeros(n) for t in range(1, n): if pos[t - 1] != 0: ry = y[t] / y[t - 1] - 1 rx = x[t] / x[t - 1] - 1 r = pos[t - 1] * (ry - beta * rx) # 多价差 = 多 y 空 x（按 beta） if pos[t] != pos[t - 1]: r -= cost ret[t] = r eq = np.cumprod(1 + ret) sharpe = ret[1:].mean() / (ret[1:].std() + 1e-9) * np.sqrt(252) mdd = (eq - np.maximum.accumulate(eq)).min() return dict(total=eq[-1] - 1, sharpe=sharpe, mdd=mdd, beta=beta, hl=half_life(spread)) if __name__ == "__main__": x, y = gen_pair() m = backtest(x, y) print("统计套利 / 配对交易（协整 + 价差 Z-Score）回测") print(f" 样本K线数 : {len(x)}") print(f" 对冲比β : {m['beta']:.3f}") print(f" 价差半衰期 : {m['hl']:.2f} 根K线") print(f" 累计收益 : {m['total'] * 100:8.1f}%") print(f" 年化夏普 : {m['sharpe']:8.2f}") print(f" 最大回撤 : {m['mdd'] * 100:8.1f}%")