市场是永恒的对话，配对交易是其中最安静的回声
2021 年 1 月，游戏驿站（GameStop，GME）在 WallStreetBets 论坛上被散户抱团做多，股价在两周内从 20 美元飙升至 483 美元。无数量化交易者在程序化追高的过程中爆仓。但也有一批机构在同期的期权市场里做多 GME 隐含波动率、做空 AMC 的相关性结构——利用的不是价格方向，而是两只高相关股票之间的价差回归特性。他们在那场混乱中赚了钱，不是因为预测对了方向，而是因为他们知道：当市场发疯时，相关性会短暂失效，但协整关系不会。
配对交易（Pairs Trading）的底层逻辑正是如此。它不赌方向，只赌价差的均值回归。找到两只长期来看"一起走路"的股票，当它们的价差突然拉开时，卖出涨多了的、买入跌少了的，等价差收回去获利。这个策略在数学上简洁优雅，但在工程上横跨了三个复杂环节：协整对筛选、动态对冲比率计算、实时价差监控与信号触发。
本文拆解这三环，提供生产级的 Python 实现。
一、为什么协整比相关更重要
在进入代码之前，必须先厘清一组容易被混淆的概念：相关性（Correlation）和协整性（Cointegration）。
相关性衡量的是两个序列"一起变动"的趋势——当 A 涨时 B 是否也倾向于涨。但它不关心两者的绝对水平。这意味着：即便 A 从 10 涨到 100、B 从 1 涨到 10，它们的相关性可能高达 0.95，但价差已经从 9 扩大到了 90。相关性能告诉你"同向还是反向"，但无法告诉你"价差会不会回来"。
协整性解决的就是这个问题。如果两个序列 A 和 B 是协整的，那么存在一个系数 β，使得 A - β·B 的差值围绕一个常数上下波动，不会漂移太远。用回归的话说，残差序列是平稳的（Stationary）。
这正是配对交易的核心假设：残差的均值回归特性。 只要协整关系成立，无论两只股票各自涨跌多少，价差终将收敛。均值回归的周期可能是几天，也可能是几周——但只要协整结构未被破坏，它就一定会发生。
数学定义：若序列 X_t 和 Y_t 满足：存在 α、β 使得 Z_t = X_t - β·Y_t - α 是平稳序列（ADF 检验 p-value < 0.05），则称 X_t 与 Y_t 协整。
这个定义里有几个关键点：系数 β 不是固定的（随着时间推移，两只股票的关系可能漂移），平稳性需要检验（不是凭直觉），以及残差 Z_t 是我们监控的对象。
二、从数千只股票中筛选协整对：两步走策略
2.1 第一步：粗筛——相关性 + 行业相关性
在全部股票中做两两协整检验的成本是 O(n²)，以 3000 只股票为例，需要检验约 450 万对，计算量极大。因此我们采用两级筛选：
粗筛条件：
- 过去 252 个交易日日收益率相关性 ≥ 0.60（可选阈值，保守可设为 0.70）
- 同属一个行业或上下游产业链（可选但强烈推荐，减少伪相关）
理由：协整的对必然在长期有较高的相关性，但高相关性不一定协整。粗筛将候选对从百万级压缩到万级以内。
2.2 第二步：精筛——Engle-Granger 两步法协整检验
粗筛之后，对每一对候选股票进行正式的协整检验。这里使用 statsmodels 中的 coint 函数，采用 Engle-Granger 方法：
import numpy as np
from statsmodels.tsa.stattools import coint
import pandas as pd
import os
import requests
def fetch_historical_prices(symbols: list[str], interval: str = "1d", limit: int = 252) -> pd.DataFrame:
"""
通过 TickDB REST API 获取历史日线收盘价。
返回 DataFrame，index 为日期，columns 为股票代码。
"""
api_key = os.environ.get("TICKDB_API_KEY")
if not api_key:
raise ValueError("请设置环境变量 TICKDB_API_KEY")
headers = {"X-API-Key": api_key}
result = pd.DataFrame()
for symbol in symbols:
try:
response = requests.get(
"https://api.tickdb.ai/v1/market/kline",
headers=headers,
params={"symbol": symbol, "interval": interval, "limit": limit},
timeout=(3.05, 10)
)
data = response.json()
if data.get("code") == 0:
klines = data["data"]
prices = pd.DataFrame(klines)["close"].values
result[symbol] = prices
else:
print(f"[WARN] {symbol} 获取失败: {data.get('message')}")
except Exception as e:
print(f"[ERROR] {symbol} 请求异常: {e}")
return result
def coint_test(series1: np.ndarray, series2: np.ndarray) -> dict:
"""
Engle-Granger 协整检验。
返回：p_value、t_stat、临界值、beta（Hedge Ratio）
"""
t_stat, p_value, crit_values = coint(series1, series2)
# 计算动态 hedge ratio（OLS 斜率）
beta = np.polyfit(series2, series1, 1)[0]
return {
"p_value": p_value,
"t_stat": t_stat,
"crit_1pct": crit_values[0], # 1% 临界值
"crit_5pct": crit_values[1], # 5% 临界值
"beta": beta,
"is_cointegrated": p_value < 0.05
}
def screen_cointegration_pairs(
symbol_pool: list[str],
lookback: int = 252,
corr_threshold: float = 0.65,
pvalue_threshold: float = 0.05
) -> pd.DataFrame:
"""
两步筛选协整对：
1. 按相关性粗筛
2. 对粗筛候选对做 Engle-Granger 协整检验
返回符合协整条件的配对列表。
"""
print(f"正在获取 {len(symbol_pool)} 只股票的历史数据...")
prices = fetch_historical_prices(symbol_pool, limit=lookback)
prices = prices.dropna(axis=1, how="any")
if prices.shape[1] < 2:
raise ValueError("有效股票数量不足，无法进行配对筛选")
returns = prices.pct_change().dropna()
# 计算相关矩阵
corr_matrix = returns.corr()
pairs = []
checked = set()
print("开始协整检验（这可能需要几分钟）...")
symbols = prices.columns.tolist()
for i, sym_a in enumerate(symbols):
for sym_b in symbols[i + 1:]:
pair_key = tuple(sorted([sym_a, sym_b]))
if pair_key in checked:
continue
checked.add(pair_key)
# 粗筛：相关性阈值
corr = corr_matrix.loc[sym_a, sym_b]
if corr < corr_threshold:
continue
# 精筛：协整检验
s1 = prices[sym_a].values
s2 = prices[sym_b].values
# ⚠️ 剔除 NaN 和常量序列
if np.any(np.isnan(s1)) or np.any(np.isnan(s2)):
continue
if np.std(s1) < 1e-8 or np.std(s2) < 1e-8:
continue
result = coint_test(s1, s2)
if result["is_cointegrated"]:
pairs.append({
"symbol_a": sym_a,
"symbol_b": sym_b,
"correlation": round(corr, 4),
"p_value": round(result["p_value"], 4),
"beta": round(result["beta"], 4),
"t_stat": round(result["t_stat"], 4),
})
df_pairs = pd.DataFrame(pairs).sort_values("p_value")
print(f"协整检验完成，共筛选出 {len(df_pairs)} 对有效协整对。")
return df_pairs
⚠️ 工程提示：上述双重循环在 symbol_pool 较大时计算量仍然不低。建议生产环境使用
numbaJIT 编译或将协整检验并行化到多进程。粗筛的corr_threshold也可视机器性能调整——阈值越低，候选对越多，最终结果越完整，但耗时越长。
三、用卡尔曼滤波动态计算 Hedge Ratio
3.1 为什么 OLS 不够用
在协整检验中，我们用 OLS 计算了静态的 β（Hedge Ratio）——即假设两只股票的价差关系是恒定不变的。但在真实市场中，这个关系会漂移。
举一个经典的例子：道琼斯工业指数中的两只成分股 XOM 和 CVX（两家能源巨头）。它们的长期相关性很高，但每当原油价格剧烈波动时，XOM 的业务对能源价格的敏感度会比 CVX 更高——此时两者的 β 会系统性地上移或下移。如果用历史 OLS 算出的固定 β 来构建价差序列，这个序列就不再是平稳的，Z-Score 的均值和标准差都会漂移，触发信号的阈值会越来越不准。
卡尔曼滤波（Kalman Filter）正是解决这个问题的工具。它本质上是一个递归贝叶斯估计器，在每个新的价格数据点到来时，用预测—更新两步循环来动态估算当前的 β 和 α。简单说：
- 预测步：根据上一时刻的状态估计当前状态，并给出估计的不确定度
- 更新步：当新数据到来时，结合观测值修正估计，降低不确定度
优势是：不需要存储全量历史数据，只需要维护当前状态的均值和协方差矩阵，适合实时流数据场景。
3.2 卡尔曼滤波配对交易实现
import numpy as np
from dataclasses import dataclass, field
from typing import Tuple
@dataclass
class KalmanPairFilter:
"""
基于卡尔曼滤波的动态协整对冲比率估计器。
状态向量: [alpha, beta]，即 Y_t = alpha + beta * X_t + epsilon_t
状态转移模型: 随机游走（假设 alpha/beta 随时间缓慢漂移）
观测模型: 线性关系 + 高斯噪声
"""
delta: float = 1e-4 # 状态转移噪声参数（控制 beta 漂移速度）
Ve: float = 1e-3 # 观测噪声方差
# 内部状态（由 filter_step 维护）
alpha: float = 0.0
beta: float = 0.0
P: np.ndarray = field(
default_factory=lambda: np.eye(2) * 1e-4 # 状态协方差矩阵
)
R: np.ndarray = field(
default_factory=lambda: np.eye(2) * 1 # 预测误差协方差
)
def filter_step(self, y: float, x: float) -> Tuple[float, float, float, float]:
"""
单步卡尔曼滤波更新。
参数:
y: 被对冲资产价格（symbol_a）
x: 对冲资产价格（symbol_b）
返回:
(alpha, beta, spread, spread_std)
其中 spread = y - beta * x - alpha（当前残差）
spread_std 为残差的标准差估计
"""
# --- 预测步 ---
# 状态随机游走：alpha_{t} = alpha_{t-1}, beta_{t} = beta_{t-1}
# F = [[1, 0], [0, 1]]（简化为恒等转移）
# Q = delta * P（状态噪声与当前协方差成正比）
Q = self.delta * np.abs(self.P)
self.R = self.P + Q
# --- 更新步 ---
# 观测: y_t = H_t @ theta_t + v_t
# H = [1, x]
H = np.array([[1.0, x]])
# 观测预测: y_pred = H @ theta
y_pred = self.alpha + self.beta * x
# 预测误差（残差/创新）
e = y - y_pred
# 观测预测误差方差: S = H @ R @ H.T + Ve
S = float(H @ self.R @ H.T + self.Ve)
# 卡尔曼增益: K = R @ H.T @ S^(-1)
K = self.R @ H.T / S
# 更新状态估计
state = np.array([self.alpha, self.beta])
state = state + K.flatten() * e
self.alpha, self.beta = state
# 更新状态协方差: P = (I - K @ H) @ R
I_KH = np.eye(2) - np.outer(K.flatten(), H)
self.P = I_KH @ self.R
# 当前残差（spread）和其标准差
spread = y - self.beta * x - self.alpha
# spread 的条件方差（卡尔曼滤波给出的理论方差）
spread_var = S
spread_std = np.sqrt(spread_var)
return self.alpha, self.beta, spread, spread_std
class PairsMonitor:
"""
配对监控器：维护一组协整对，实时更新卡尔曼滤波状态，
计算 Z-Score，当超过阈值时触发告警。
"""
def __init__(
self,
pairs: list[Tuple[str, str]],
z_threshold: float = 2.0,
rolling_window: int = 60
):
self.pairs = pairs
self.z_threshold = z_threshold
self.rolling_window = rolling_window
# 每对维护一个 KalmanPairFilter
self.filters: dict[str, KalmanPairFilter] = {}
# 历史残差（用于计算滚动 Z-Score）
self.spread_history: dict[str, list[float]] = {}
# 残差的滚动均值和标准差
self.spread_means: dict[str, float] = {}
self.spread_stds: dict[str, float] = {}
for sym_a, sym_b in pairs:
key = f"{sym_a}|{sym_b}"
self.filters[key] = KalmanPairFilter()
self.spread_history[key] = []
self.spread_means[key] = 0.0
self.spread_stds[key] = 1.0 # 初始避免除零
def update(self, sym_a: str, price_a: float, sym_b: str, price_b: float) -> dict | None:
"""
输入新的价格数据对，返回触发告警的信息（如果 Z-Score 超过阈值）。
返回 None 表示未触发。
"""
key = f"{sym_a}|{sym_b}"
if key not in self.filters:
return None
kf = self.filters[key]
# 卡尔曼滤波更新
alpha, beta, spread, spread_std = kf.filter_step(price_a, price_b)
# 维护滚动历史
history = self.spread_history[key]
history.append(spread)
if len(history) > self.rolling_window:
history.pop(0)
# 计算滚动 Z-Score（使用卡尔曼方差作为当前 spread_std）
# 当历史数据足够时，用样本均值和标准差替代卡尔曼方差
if len(history) >= 20:
self.spread_means[key] = np.mean(history)
self.spread_stds[key] = np.std(history, ddof=1) + 1e-9
z_score = (spread - self.spread_means[key]) / self.spread_stds[key]
# 触发告警条件：Z-Score 绝对值超过阈值
if abs(z_score) >= self.z_threshold:
return {
"pair": key,
"spread": round(spread, 4),
"z_score": round(z_score, 4),
"beta": round(beta, 4),
"alpha": round(alpha, 4),
"direction": "short_spread" if z_score > 0 else "long_spread",
"alert_time": pd.Timestamp.now().isoformat(),
}
return None
⚠️ 生产环境注意事项：
rolling_window的选择取决于配对特性，高频场景可设为 20-60，低频日线策略建议 60-120。delta参数控制卡尔曼滤波对 β 漂移的敏感度：值越大，β 更新越快（但也更噪声）。建议用历史数据做网格搜索优化。- Z-Score 触发后，应检查残差的 ADF 统计量，确认平稳性未被破坏，再决定是否建仓。
四、TickDB WebSocket 实时数据接入
上一节的 PairsMonitor 类需要一个实时价格流来驱动。在生产环境中，我们需要同时订阅多只股票的价格推送，并高效地路由到对应的配对监控器。
以下是 TickDB WebSocket 的生产级连接封装，支持心跳保活、指数退避重连和限频处理：
import json
import time
import random
import threading
import os
import requests
import websocket # pip install websocket-client
class TickDBWebSocketClient:
"""
TickDB WebSocket 生产级客户端。
包含心跳保活、指数退避重连、限频自适应。
⚠️ 高频场景（<100ms 更新频率）建议使用 aiohttp + asyncio 重写。
"""
def __init__(self, on_price_update, api_key: str | None = None):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
self.on_price_update = on_price_update
self.ws = None
self.running = False
self._thread: threading.Thread | None = None
self._base_delay = 2.0 # 初始重连等待（秒）
self._max_delay = 60.0 # 最大重连等待
self._jitter_factor = 0.1 # 抖动比例
# 限频状态
self._rate_limit_until: float = 0.0
def connect(self):
if self.running:
return
self.running = True
self._thread = threading.Thread(target=self._run_loop, daemon=True)
self._thread.start()
def _run_loop(self):
retry = 0
while self.running:
try:
ws_url = f"wss://ws.tickdb.ai/v1/market?api_key={self.api_key}"
self.ws = websocket.create_connection(
ws_url,
timeout=30,
enable_multithread=True
)
print("[TickDB WS] 连接建立成功")
self._base_delay = 2.0 # 重置退避
self._listen()
except Exception as e:
print(f"[TickDB WS] 连接异常: {e}，准备重连...")
self.ws = None
delay = min(self._base_delay * (2 ** retry), self._max_delay)
jitter = random.uniform(0, delay * self._jitter_factor)
time.sleep(delay + jitter)
retry = min(retry + 1, 10)
def _listen(self):
while self.running and self.ws and self.ws.connected:
try:
msg = self.ws.recv()
data = json.loads(msg)
# 处理限频响应（code: 3001）
if data.get("code") == 3001:
retry_after = int(data.get("headers", {}).get(
"Retry-After",
self.ws.sock.options.get("timeout", 5)
))
print(f"[TickDB WS] 触发限频，等待 {retry_after}s")
time.sleep(retry_after)
continue
# 处理心跳响应
if data.get("type") == "pong":
continue
# 分发数据
self.on_price_update(data)
except websocket.WebSocketTimeoutException:
# 发送心跳 ping
try:
self.ws.send(json.dumps({"cmd": "ping"}))
except Exception:
break
except Exception as e:
print(f"[TickDB WS] 接收异常: {e}")
break
def subscribe(self, symbols: list[str], channels: list[str] = ["kline_1m"]):
"""
订阅行情数据。
channels 可选: kline_1m, kline_5m, trades, depth
"""
if not self.ws or not self.running:
raise RuntimeError("WebSocket 未连接，请先调用 connect()")
subscribe_msg = {
"cmd": "subscribe",
"params": {
"symbols": symbols,
"channels": channels
}
}
self.ws.send(json.dumps(subscribe_msg))
print(f"[TickDB WS] 已订阅: {symbols} @ {channels}")
def unsubscribe(self, symbols: list[str]):
if not self.ws or not self.running:
return
self.ws.send(json.dumps({
"cmd": "unsubscribe",
"params": {"symbols": symbols}
}))
def disconnect(self):
self.running = False
if self.ws:
self.ws.close()
if self._thread:
self._thread.join(timeout=5)
print("[TickDB WS] 连接已关闭")
4.1 价格路由与配对监控联动
import pandas as pd
from collections import defaultdict
def build_pairs_monitor(
symbol_pool: list[str],
corr_threshold: float = 0.65,
z_threshold: float = 2.0
) -> Tuple[PairsMonitor, list[str]]:
"""
从股票池构建配对监控器，同时返回需要订阅的全部股票代码。
"""
# 筛选协整对（见第二节）
pairs_df = screen_cointegration_pairs(
symbol_pool,
corr_threshold=corr_threshold
)
# 取前 10 对最强的协整对（按 p_value 排序）
top_pairs = pairs_df.head(10)
pairs = [
(row["symbol_a"], row["symbol_b"])
for _, row in top_pairs.iterrows()
]
monitor = PairsMonitor(pairs, z_threshold=z_threshold)
# 收集需要订阅的全部股票
all_symbols = list(set(
s for pair in pairs for s in pair
))
return monitor, all_symbols
def on_price_update_factory(monitor: PairsMonitor):
"""
构建价格回调函数，维护每个 symbol 的最新价格，
当同一配对的两只股票价格都到位后，触发监控器更新。
"""
latest_prices: dict[str, float] = {}
pending_pairs: dict[str, set] = defaultdict(set)
def callback(data: dict):
symbol = data.get("symbol")
channel = data.get("channel")
if channel == "kline_1m":
# 取当前 K 线收盘价（实时更新）
price = float(data.get("close", 0))
elif channel == "trades":
# 取最新成交价
trades = data.get("data", [])
if trades:
price = float(trades[-1].get("price", 0))
else:
return
if not price or price <= 0:
return
latest_prices[symbol] = price
# 检查哪些配对的两个标的都有最新价格
for pair_key, filter_obj in monitor.filters.items():
sym_a, sym_b = pair_key.split("|")
if symbol == sym_a:
price_b = latest_prices.get(sym_b)
if price_b:
alert = monitor.update(sym_a, price, sym_b, price_b)
if alert:
trigger_alert(alert)
elif symbol == sym_b:
price_a = latest_prices.get(sym_a)
if price_a:
alert = monitor.update(sym_a, price_a, sym_b, price)
if alert:
trigger_alert(alert)
return callback
def trigger_alert(alert: dict):
"""
触发告警通知。可扩展接入飞书/钉钉/Slack/PagerDuty。
这里以标准输出演示。
"""
direction_label = "卖出价差（做空 spread）" if alert["direction"] == "short_spread" else "买入价差（做多 spread）"
print("\n" + "=" * 60)
print(f"🚨 配对交易信号触发")
print(f" 配对: {alert['pair']}")
print(f" Z-Score: {alert['z_score']}")
print(f" 当前价差: {alert['spread']}")
print(f" 动态对冲比率 β: {alert['beta']}")
print(f" 建议操作: {direction_label}")
print(f" 触发时间: {alert['alert_time']}")
print("=" * 60 + "\n")
if __name__ == "__main__":
# 示例股票池（需替换为实际可用标的）
# 实际使用时替换为真实的美股/港股代码
SYMBOL_POOL = [
"JPM.US", "BAC.US", "WFC.US", # 银行板块
"XOM.US", "CVX.US", # 能源
"MCD.US", "SBUX.US", # 餐饮
"HD.US", "LOW.US", # 家居零售
"UNH.US", "JNJ.US", # 医疗
]
# ⚠️ 生产环境建议：从 TickDB /symbols/available 接口获取全部可用标的
# 并按行业分组进行批量筛选
monitor, symbols_to_subscribe = build_pairs_monitor(
SYMBOL_POOL,
corr_threshold=0.65,
z_threshold=2.0
)
callback = on_price_update_factory(monitor)
ws_client = TickDBWebSocketClient(
on_price_update=callback,
api_key=os.environ.get("TICKDB_API_KEY")
)
ws_client.connect()
ws_client.subscribe(symbols_to_subscribe, channels=["kline_1m"])
# 保持主线程运行
try:
while True:
time.sleep(10)
except KeyboardInterrupt:
ws_client.disconnect()
⚠️ 工程预警：
- K 线频道
kline_1m在 TickDB 中推送的是当前未完成 K 线的实时更新，而非已封闭的历史 K 线。如果你需要收盘价驱动的配对监控（以 K 线封闭为触发点），应使用kline_1m的close字段，并增加"价格变化才触发"的去重逻辑。- 上述代码在单线程中顺序处理所有配对更新。在配对数量超过 50 对、或订阅标的超过 200 只时，建议使用
asyncio重构为异步架构，避免 GIL 成为瓶颈。
五、完整策略闭环：筛选 → 监控 → 告警
将前四节的内容串联起来，配对交易的实时监控流程如下：
┌─────────────────────────────────────────────────────────┐
│ 第一阶段：离线筛选 │
│ │
│ 3000+ 股票池 → 相关性粗筛 → Engle-Granger 协整检验 │
│ → 选取最优 N 对 → 输出静态 beta 备选值 │
└───────────────────────┬─────────────────────────────────┘
│ 协整对列表 + 初始参数
▼
┌─────────────────────────────────────────────────────────┐
│ 第二阶段：实时监控 │
│ │
│ TickDB WebSocket → 多标的行情流 │
│ │ │
│ ├── 价格路由（symbol → pair lookup） │
│ │ │
│ └── PairsMonitor.update(price_a, price_b) │
│ │ │
│ ├── KalmanPairFilter.filter_step() │
│ │ 动态更新 α 和 β │
│ ├── 维护滚动残差历史窗口 │
│ │ │
│ └── 计算 Z-Score → 阈值比较 │
│ │ │
│ └── Z-Score ≥ 2.0 → trigger_alert│
└─────────────────────────────────────────────────────────┘
信号含义解读：
| Z-Score | 含义 | 策略含义 |
|---|---|---|
| Z > +2.0 | 价差高于历史均值 2 个标准差 | short_spread：卖空 symbol_a、买多 symbol_b（等价差回归） |
| Z < -2.0 | 价差低于历史均值 2 个标准差 | long_spread：买多 symbol_a、卖空 symbol_b（等价差回归） |
| β | >2.0 持续 |
六、实战参数建议与常见陷阱
6.1 参数配置参考
| 参数 | 日线策略建议 | 分钟线策略建议 | 说明 |
|---|---|---|---|
| 相关性阈值 | 0.70–0.80 | 0.60–0.70 | 日线要求更稳定的相关性 |
| 协整 p-value | < 0.05 | < 0.10 | 分钟线噪声大，阈值适当放宽 |
| Kalman delta | 1e-4 | 1e-3 | 分钟线 beta 漂移更快 |
| Z-Score 阈值 | 2.0–2.5 | 1.5–2.0 | 高频场景信号更密集 |
| 滚动窗口 | 60–120 天 | 20–60 根 K 线 | 高频场景用更短的窗口 |
| 持仓周期 | 5–20 天 | 30 分钟–2 小时 | 视均值回归速度而定 |
6.2 常见陷阱
陷阱一：过拟合协整检验
用 252 天数据检验协整，选取 p-value 最小的配对，很可能只是噪声导致的虚假协整。正确的做法是滚动窗口检验——用前 180 天筛选配对，后 60 天做样本外验证，只有在样本外依然协整的配对才用于实盘。
陷阱二：忽视交易成本
配对交易需要同时操作两只股票，双边佣金 + 价差滑点会显著侵蚀利润。当 Z-Score = 2.0 时，预期回归幅度可能只有 1%–2%，如果单边交易成本超过 0.3%，利润就会被侵蚀过半。实盘前必须用包含真实成本的模拟盘验证。
陷阱三：卡尔曼滤波初始状态不稳
滤波前 10–20 个数据点的估计波动很大，Z-Score 可能在这段窗口内频繁触发误报。解决方式：在初始化时用 OLS 预热 beta 值，或设置"预热期"——前 N 个数据点只用来估计，不触发告警。
陷阱四：忽视极端行情下的协整破裂
2020 年 3 月新冠恐慌期间，大量传统上协整的股票对（如 SPY 和 QQQ 中的成分股）出现了短暂的协整破裂——这是由于流动性危机引发的无条件抛售，破坏了均值回归的结构。风控机制必须包含最大持仓时间和强制止损。
结语
配对交易的魅力在于它不试图预测市场，而是利用市场的惯性——当两只股票的关系暂时偏离历史均值时，押注这种偏离会被修复。这个逻辑在数学上优雅，在工程上却相当复杂：需要协整统计的学术功底、卡尔曼滤波的状态估计能力，以及可靠实时数据流的基础设施。
本文覆盖了从协整对筛选到实时告警的完整链路。核心难点从来不是某个单点算法，而是三个模块的工程化衔接：用历史数据选出真正协整的对、用卡尔曼滤波在线估计动态对冲比率、用可靠的 WebSocket 流实时计算 Z-Score 并触发信号。任何一环的松懈都会让整个系统失效。
下一步行动
如果你想亲手运行本文代码：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台生成 API Key
- 设置环境变量
TICKDB_API_KEY，将SYMBOL_POOL替换为真实标的（可从/v1/symbols/available接口获取） - 运行本文完整代码，观察 Z-Score 告警输出
如果你需要更长的历史 K 线数据做配对筛选：
协整检验的可靠性高度依赖历史数据长度。建议使用 TickDB 提供的 10 年级别美股历史 K 线数据进行离线配对筛选，再将筛选结果导入实时监控流程。联系 [email protected] 了解专业版数据接口。
如果你习惯用 AI 辅助开发：
在 AI 助手中搜索并安装 tickdb-market-data SKILL，可快速接入 TickDB 数据能力，加速配对筛选和实时监控的原型开发。
本文不构成任何投资建议。配对交易涉及复杂的统计模型和交易成本，实际收益受市场微观结构影响显著。在实盘部署前，请务必进行充分的样本外测试和风险评估。