如何使用pykalman的filter_update实现价格数据流的递归卡尔曼回归

我之前用pykalman做实时递归卡尔曼回归的时候也踩过不少坑，看你的需求是用kf.filter_update()处理CAT和DOG的实时价格序列，大概率是初始化或者迭代逻辑没处理对。下面我结合你的示例数据，给你一套能跑通的代码，顺便说说常见的错误点。

先理清楚需求：卡尔曼回归的逻辑 #

我们要做的是用DOG的价格预测CAT的价格，也就是拟合模型 CAT = a*DOG + b，这里的a（斜率）和b（截距）就是卡尔曼滤波器要跟踪的状态变量。每来一条新的价格数据，我们就用filter_update()递归更新a和b的估计值。

完整可运行代码 #

第一步：导入库并加载示例数据 #

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

# 你的示例数据流
data = """DateTime,CAT,DOG
2015-01-02 09:01:00,1471.24,9868.76
2015-01-02 09:02:00,1471.75,9877.75
2015-01-02 09:03:00,1471.81,9867.70
2015-01-02 09:04:00,1471.59,9849.03
2015-01-02 09:05:00,1471.45,9840.15
2015-01-02 09:06:00,1471.16,9852.71
2015-01-02 09:07:00,1471.30,9860.24"""

# 转成DataFrame方便处理
df = pd.read_csv(pd.compat.StringIO(data), parse_dates=['DateTime'], index_col='DateTime')

第二步：初始化卡尔曼滤波器 #

这里的参数设置很关键，错一个维度就会报错：

# 状态维度：2个变量，分别是斜率a和截距b
n_dim_state = 2
# 观测维度：1个，就是CAT的价格
n_dim_obs = 1

# 转移矩阵：假设a和b不会随时间突变，所以用单位矩阵
transition_matrix = np.eye(n_dim_state)

# 初始状态估计：先给个粗略值，比如a=0，b用CAT的初始均值
initial_state_mean = np.array([0, df['CAT'].iloc[0]])

# 初始协方差：设置大一点，让滤波器有足够空间调整初始估计
initial_state_covariance = np.eye(n_dim_state) * 10

# 过程噪声：控制状态的变化幅度，数值越小说明我们认为a和b越稳定
process_covariance = np.eye(n_dim_state) * 1e-5

# 观测噪声：对应CAT价格的波动，根据数据量级调整
observation_covariance = np.array([[1e-2]])

# 初始化滤波器，观测矩阵后面每步动态更新
kf = KalmanFilter(
    transition_matrices=transition_matrix,
    observation_matrices=np.zeros((n_dim_obs, n_dim_state)),  # 先占位
    initial_state_mean=initial_state_mean,
    initial_state_covariance=initial_state_covariance,
    process_covariance=process_covariance,
    observation_covariance=observation_covariance
)

第三步：递归处理实时数据流 #

这一步的核心是手动维护每一步的状态均值和协方差，因为filter_update()不会自动记住上一步的结果：

# 初始化迭代用的状态和协方差
current_state_mean = initial_state_mean
current_state_covariance = initial_state_covariance

# 存储每一步的估计结果
estimates = []

# 模拟实时数据流：逐行处理每一条新数据
for timestamp, row in df.iterrows():
    dog_price = row['DOG']
    cat_price = row['CAT']
    
    # 关键！更新观测矩阵：当前观测模型是 CAT = a*DOG + b，所以观测矩阵是 [DOG, 1]
    kf.observation_matrices = np.array([[dog_price, 1]])
    
    # 执行递归更新
    current_state_mean, current_state_covariance = kf.filter_update(
        current_state_mean,
        current_state_covariance,
        observation=cat_price
    )
    
    # 保存当前的估计值和预测值
    estimates.append({
        'datetime': timestamp,
        'estimated_slope': current_state_mean[0],
        'estimated_intercept': current_state_mean[1],
        'observed_cat': cat_price,
        'predicted_cat': current_state_mean[0] * dog_price + current_state_mean[1]
    })

# 转成DataFrame查看结果
estimates_df = pd.DataFrame(estimates).set_index('datetime')
print(estimates_df[['estimated_slope', 'estimated_intercept', 'observed_cat', 'predicted_cat']].head())

你可能踩的坑 #

• 观测矩阵维度不匹配：如果忘记每步更新观测矩阵，或者矩阵形状不是(1,2)，直接会报维度错误。因为每一步的DOG价格不一样，观测矩阵必须跟着变。
• 没维护迭代状态：很多人会犯的错是每次调用filter_update()都用初始的initial_state_mean，而不是上一步返回的current_state_mean，这样相当于没有递归更新，结果完全不对。
• 噪声参数设置不合理：如果过程噪声太大，估计值会波动剧烈；如果观测噪声太小，滤波器会过度信任观测值，失去平滑效果。可以根据你的数据量级调整这两个参数。
• 数据类型错误：确保传入observation的是单个标量，而不是DataFrame的行或者数组，不然会触发维度错误。

内容的提问来源于stack exchange，提问作者Pman70