卡尔曼滤波配对交易策略 (Kalman Filter Pairs Trading)

策略原理

配对交易（Pairs Trading）是一种基于均值回归的市场中性策略。传统的配对交易通常使用普通最小二乘法（OLS）在固定窗口期内计算两只股票的对冲比例（Hedge Ratio/Beta）。然而，金融市场是动态变化的，资产之间的相关性会随时间漂移。

卡尔曼滤波（Kalman Filter） 是一种线性二次估计算法，能够通过一系列含有噪声的观测数据，递归地估计动态系统的内部状态。在配对交易中，我们利用卡尔曼滤波来动态估计对冲比例（Beta）和截距（Alpha），相比传统的移动窗口OLS，它具有以下优势：

1. 自适应性强：能够迅速捕捉两只股票之间关系的结构性变化。
2. 无需回溯窗口：它是一个递归过程，只需要上一时刻的状态和当前的观测值，不需要存储大量的历史数据。
3. 噪声平滑：能够有效分离市场噪声和真实的参数变化。

数学模型

我们将两只股票 $Y$（因变量）和 $X$（自变量）的价格关系建模为：

$$Y_t = \alpha_t + \beta_t X_t + \epsilon_t$$

其中：

• 状态变量 (State)：$[\beta_t, \alpha_t]$，即我们需要估计的对冲比例和截距。
• 观测变量 (Observation)：$Y_t$，即股票Y的当前价格。
• 观测矩阵：$[X_t, 1]$。

策略逻辑：

1. 利用卡尔曼滤波实时更新 $\beta$ 和 $\alpha$。
2. 计算价差（Spread） 或 预测误差：$e_t = Y_t - (\alpha_t + \beta_t X_t)$。
3. 计算价差的 Z-Score（标准分）。
4. 当 Z-Score > 阈值（如2.0），说明 $Y$ 相对 $X$ 被高估，做空 $Y$，做多 $X$。
5. 当 Z-Score < -阈值（如-2.0），说明 $Y$ 相对 $X$ 被低估，做多 $Y$，做空 $X$。
6. 当 Z-Score 回归到 0 附近时平仓。

PTrade 策略代码实现

以下代码在 PTrade 平台上实现了基于 pykalman 库的配对交易策略。

import numpy as np
import pandas as pd
from pykalman import KalmanFilter

def initialize(context):
    """
    初始化函数
    """
    # 1. 设定要进行配对交易的两只股票
    # 这里选取两只银行股作为示例：浦发银行(Y) 和 招商银行(X)
    g.stock_y = '600000.SS' 
    g.stock_x = '600036.SS'
    
    # 设定股票池
    set_universe([g.stock_y, g.stock_x])
    
    # 2. 策略参数设置
    g.entry_threshold = 2.0  # 开仓阈值 (Z-Score)
    g.exit_threshold = 0.5   # 平仓阈值 (Z-Score)
    g.window_size = 20       # 用于计算Spread标准差的滚动窗口
    
    # 3. 卡尔曼滤波初始化
    # 状态变量维度为2: [beta, alpha]
    # 观测变量维度为1: [price_y]
    g.kf = KalmanFilter(
        n_dim_obs=1, 
        n_dim_state=2,
        initial_state_mean=np.zeros(2),
        initial_state_covariance=np.ones((2, 2)),
        transition_matrices=np.eye(2),      # 状态转移矩阵，假设参数遵循随机游走
        observation_matrices=None,          # 观测矩阵在每一步动态构建
        observation_covariance=1.0,         # 观测噪声协方差
        transition_covariance=np.eye(2)*1e-4 # 状态转移噪声协方差 (delta)
    )
    
    # 初始化状态均值和协方差
    g.state_mean = np.zeros(2)
    g.state_cov = np.ones((2, 2))
    
    # 用于存储历史价差以计算标准差
    g.spread_history = []
    
    # 预热期：策略启动前几天不交易，先让滤波器收敛
    g.warmup_days = 10
    g.days_counter = 0

def handle_data(context, data):
    """
    每日运行逻辑
    """
    # 检查两只股票是否都在数据中（避免停牌导致报错）
    if g.stock_y not in data or g.stock_x not in data:
        return
    
    # 1. 获取最新收盘价
    price_y = data[g.stock_y]['close']
    price_x = data[g.stock_x]['close']
    
    # 2. 构建当前的观测矩阵 H = [[price_x, 1.0]]
    # 对应方程: Y = beta * X + alpha * 1
    current_observation_matrix = np.array([[price_x, 1.0]])
    
    # 3. 在线更新卡尔曼滤波状态
    # filter_update 会根据上一时刻的均值和协方差，结合当前观测值，计算新的均值和协方差
    g.state_mean, g.state_cov = g.kf.filter_update(
        filtered_state_mean=g.state_mean,
        filtered_state_covariance=g.state_cov,
        observation=price_y,
        observation_matrix=current_observation_matrix
    )
    
    # 提取当前的估计值
    beta = g.state_mean[0]
    alpha = g.state_mean[1]
    
    # 4. 计算当前的价差 (Spread) / 预测误差
    # Spread = 实际Y - 预测Y
    predicted_y = beta * price_x + alpha
    current_spread = price_y - predicted_y
    
    # 存入历史记录
    g.spread_history.append(current_spread)
    if len(g.spread_history) > g.window_size:
        g.spread_history.pop(0)
    
    # 增加计数器，预热期内不交易
    g.days_counter += 1
    if g.days_counter <= g.warmup_days:
        return
        
    # 5. 计算 Z-Score
    if len(g.spread_history) < g.window_size:
        return
        
    spread_std = np.std(g.spread_history)
    if spread_std == 0:
        return
        
    z_score = current_spread / spread_std
    
    # 打印日志方便调试
    # log.info("Date: %s, Beta: %.4f, Alpha: %.4f, Z-Score: %.4f" % (context.blotter.current_dt, beta, alpha, z_score))
    
    # 6. 交易逻辑
    curr_position_y = context.portfolio.positions[g.stock_y].amount
    curr_position_x = context.portfolio.positions[g.stock_x].amount
    
    # 获取当前总资金
    total_value = context.portfolio.portfolio_value
    # 设定每只股票的仓位比例（例如各半仓操作）
    target_value = total_value * 0.5
    
    # --- 信号判断 ---
    
    # 情况A: Z-Score > 2.0，Y相对于X太贵 -> 做空Y，做多X
    if z_score > g.entry_threshold:
        # PTrade回测通常不支持直接融券做空，这里用卖出持仓模拟做空平仓，或者假设是多空环境
        # 如果是纯股票多头策略，这里只能卖出Y，买入X
        
        # 简单的配对逻辑：持有X，不持有Y
        order_target_value(g.stock_y, 0)
        order_target_value(g.stock_x, target_value)
        
    # 情况B: Z-Score < -2.0，Y相对于X太便宜 -> 做多Y，做空X
    elif z_score < -g.entry_threshold:
        # 简单的配对逻辑：持有Y，不持有X
        order_target_value(g.stock_x, 0)
        order_target_value(g.stock_y, target_value)
        
    # 情况C: Z-Score 回归到 0 附近 (-0.5 < z < 0.5) -> 平仓
    elif abs(z_score) < g.exit_threshold:
        order_target_value(g.stock_y, 0)
        order_target_value(g.stock_x, 0)

代码关键点解析

1. 库的导入：

   • PTrade 环境支持 pykalman 库，这是实现该策略的核心。
   • numpy 用于矩阵运算。

2. 初始化 (initialize)：

   • transition_covariance (状态转移协方差)：通常设为一个很小的值（如 1e-4），这决定了我们允许 Beta 和 Alpha 随时间变化的剧烈程度。值越大，Beta 变化越快（适应性强但噪声大）；值越小，Beta 越稳定。
   • observation_matrices：在初始化时设为 None，因为 $X$ 的价格每天都在变，我们需要在 handle_data 中动态构建。

3. 在线更新 (filter_update)：

   • 这是卡尔曼滤波在实时交易中的核心用法。我们不重新拟合整个历史数据，而是基于上一时刻的状态 (g.state_mean, g.state_cov) 和当前的观测 (price_y, price_x) 来推断当前的状态。
   • 这使得计算非常高效。

4. Z-Score 计算：

   • Spread 本身的大小取决于股价的绝对值，不具备标准化的交易意义。
   • 我们维护一个 g.spread_history（例如过去20天），计算其标准差，从而得到 Z-Score。这代表当前的价差偏离了均值多少个标准差。

5. 交易执行：

   • 代码中使用了 order_target_value。
   • 注意：A股市场融券（做空）较为困难。上述代码演示的是逻辑上的“做多/做空”。在实际A股纯多头策略中，通常表现为“轮动”：当Y高估时，卖出Y买入X；当Y低估时，买入Y卖出X。如果需要真实的做空（融券），需要开通两融权限并使用 marginsec_open 等两融接口。

常见问题 (Q&A)

Q: 为什么使用卡尔曼滤波而不是简单的移动平均线？
A: 移动平均线（如布林带）假设均值和方差在窗口期内是相对静态的。卡尔曼滤波将 Beta 视为一个随时间游走的变量，能够更敏锐地捕捉两只股票相关性的结构性变化（例如某家公司基本面发生改变导致两者长期比价关系改变）。

Q: 如何选择配对的股票？
A: 应该选择同行业、基本面相似、历史价格走势高度相关的股票。例如：四大行之间、两桶油（中石油/中石化）、航空股之间等。

Q: 代码中的 transition_covariance 参数如何调整？
A: 这个参数控制了模型对新信息的敏感度。

• 如果你发现 Beta 变化太慢，跟不上市场节奏，可以调大该值（例如 1e-3）。
• 如果你发现 Beta 波动太剧烈，受噪音影响大，可以调小该值（例如 1e-5）。

Q: 回测时报错 pykalman 找不到怎么办？
A: PTrade 的标准研究和回测环境中通常包含此库。如果确实缺失，需要手动实现卡尔曼滤波的递推公式（基于 Numpy），但这涉及较复杂的矩阵运算代码。上述代码基于 PTrade 官方文档支持的三方库列表编写。