22、曲面形态预测:时间序列模型、状态空间模型、深度学习预测
波动率曲面不是静态的。它每分每秒都在变化。
做期权交易这么多年,我最大的体会就是:预测曲面形态的变化,比预测曲面本身更重要。你想想看,如果你能提前知道曲面会怎么变,那交易信号自然就出来了。
这一章,我们就来聊聊怎么预测曲面形态。我会从三个角度切入:传统的时间序列模型、更灵活的状态空间模型,以及现在很火的深度学习方法。
22.1 时间序列模型:从ARIMA到VAR
先说最基础的。时间序列模型,说白了就是「用过去预测未来」。
我个人习惯把曲面形态拆成三个维度:水平(Level)、斜率(Slope)、曲率(Curvature)。这三个因子,每个都可以单独建模。
22.1.1 单因子ARIMA建模
假设我们提取了曲面水平因子(比如ATM波动率的均值),那就可以用ARIMA来预测。
import pandas as pd
import numpy as np
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
# 加载曲面因子数据
# 假设 df 包含 'level', 'slope', 'curvature' 三列
df = pd.read_csv('surface_factors.csv', index_col=0, parse_dates=True)
# 检查平稳性
from statsmodels.tsa.stattools import adfuller
result = adfuller(df['level'].dropna())
print(f'ADF Statistic: {result[0]:.4f}')
print(f'p-value: {result[1]:.4f}')
# 如果非平稳,做一阶差分
if result[1] > 0.05:
df['level_diff'] = df['level'].diff()
# 用差分后的数据建模
series = df['level_diff'].dropna()
else:
series = df['level'].dropna()
# 看ACF和PACF图,确定p和q
# plot_acf(series)
# plot_pacf(series)
# 拟合ARIMA模型
model = ARIMA(series, order=(2, 1, 2)) # p=2, d=1, q=2
model_fit = model.fit()
print(model_fit.summary())
# 预测未来5个时间步
forecast = model_fit.forecast(steps=5)
print(f'预测值: {forecast}')
22.1.2 多因子VAR模型
单因子模型有个问题:它假设三个因子是独立的。但实际中,水平、斜率、曲率是相互影响的。
举个例子:当水平因子上升时,斜率因子往往会变平(即远期升水减少)。这就是所谓的「联动效应」。
这时候,VAR(向量自回归)模型就派上用场了。
from statsmodels.tsa.api import VAR
# 准备多因子数据
data = df[['level', 'slope', 'curvature']].dropna()
# 检查平稳性
for col in data.columns:
result = adfuller(data[col])
print(f'{col}: p-value = {result[1]:.4f}')
# 如果非平稳,做差分
if any(result[1] > 0.05 for result in [adfuller(data[col]) for col in data.columns]):
data_diff = data.diff().dropna()
else:
data_diff = data
# 选择滞后阶数
model = VAR(data_diff)
results = model.select_order(maxlags=10)
print(results.summary())
# 拟合VAR模型
var_model = model.fit(maxlags=3) # 根据AIC/BIC选择
print(var_model.summary())
# 预测
lag_order = var_model.k_ar
forecast = var_model.forecast(data_diff.values[-lag_order:], steps=5)
print(f'预测的因子变化: {forecast}')
22.2 状态空间模型:卡尔曼滤波的妙用
时间序列模型有个硬伤:它假设参数是固定的。但市场在变,参数也在变。
状态空间模型就不一样了。它把曲面因子看作「隐藏状态」,通过观测数据来实时更新。说白了,就是让模型自己「学习」市场的变化。
22.2.1 动态因子模型
我常用的一个框架是:
- 状态方程:因子随时间演化,比如随机游走或均值回复
- 观测方程:观测到的波动率 = 因子载荷 × 因子值 + 噪声
import numpy as np
from pykalman import KalmanFilter
# 假设我们有10个期限的波动率数据
# 观测矩阵:10个观测值,3个因子
n_obs = 10
n_factors = 3
# 初始化卡尔曼滤波器
kf = KalmanFilter(
n_dim_obs=n_obs,
n_dim_state=n_factors,
initial_state_mean=np.zeros(n_factors),
initial_state_covariance=np.eye(n_factors) * 0.1,
transition_matrices=np.eye(n_factors), # 随机游走假设
observation_matrices=None, # 需要估计
observation_covariance=np.eye(n_obs) * 0.01,
transition_covariance=np.eye(n_factors) * 0.001
)
# 模拟数据:10个期限,500个时间步
np.random.seed(42)
n_steps = 500
true_factors = np.random.randn(n_steps, n_factors) * 0.1
# 假设因子载荷是已知的(实际中需要估计)
loadings = np.random.randn(n_obs, n_factors) * 0.5
observations = true_factors @ loadings.T + np.random.randn(n_steps, n_obs) * 0.05
# 运行卡尔曼滤波
state_means, state_covariances = kf.filter(observations)
# 提取估计的因子
estimated_factors = state_means
print(f'估计的因子形状: {estimated_factors.shape}')
print(f'最后5个时间步的因子估计:')
print(estimated_factors[-5:])
22.2.2 时变参数模型
有时候,因子载荷本身也会随时间变化。比如,短期波动率对曲面的影响可能在某个时期变大。
这时候,我们可以让载荷也随时间变化:
# 扩展状态空间:让载荷也随时间变化
# 状态向量 = [因子1, 因子2, 因子3, 载荷1_1, 载荷1_2, ...]
# 这会让状态维度变得很大,需要小心
# 简化版本:使用滚动窗口估计载荷
def estimate_time_varying_loadings(obs, window=60):
n_steps, n_obs = obs.shape
n_factors = 3
loadings_history = []
for t in range(window, n_steps):
# 用过去window天的数据估计载荷
obs_window = obs[t-window:t]
# 用PCA估计因子和载荷
from sklearn.decomposition import PCA
pca = PCA(n_components=n_factors)
factors = pca.fit_transform(obs_window)
loadings = pca.components_.T
loadings_history.append(loadings)
return np.array(loadings_history)
# 使用示例
loadings_series = estimate_time_varying_loadings(observations, window=60)
print(f'时变载荷形状: {loadings_series.shape}')
22.3 深度学习预测:LSTM与Transformer
传统模型有个天花板:它们假设因子之间的关系是线性的。但市场行为往往是非线性的。
深度学习模型,特别是LSTM和Transformer,可以捕捉这些非线性关系。
22.3.1 LSTM预测曲面形态
LSTM适合处理时间序列数据。它可以通过「门控机制」记住长期依赖关系。
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
class SurfaceLSTM(nn.Module):
def __init__(self, input_size=10, hidden_size=64, num_layers=2, output_size=10):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=0.2
)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# x: (batch, seq_len, input_size)
lstm_out, (h_n, c_n) = self.lstm(x)
# 取最后一个时间步的输出
last_out = lstm_out[:, -1, :]
output = self.fc(last_out)
return output
# 准备数据
def prepare_sequence_data(obs, seq_len=20, pred_len=5):
X, y = [], []
for i in range(len(obs) - seq_len - pred_len):
X.append(obs[i:i+seq_len])
y.append(obs[i+seq_len:i+seq_len+pred_len])
return np.array(X), np.array(y)
# 假设 obs 是 (n_steps, n_tenors) 的波动率曲面数据
seq_len = 20
pred_len = 5
X, y = prepare_sequence_data(observations, seq_len, pred_len)
# 转换为PyTorch张量
X_tensor = torch.FloatTensor(X)
y_tensor = torch.FloatTensor(y)
# 创建数据加载器
dataset = TensorDataset(X_tensor, y_tensor)
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
# 初始化模型
model = SurfaceLSTM(input_size=10, hidden_size=64, num_layers=2, output_size=10*pred_len)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# 训练
n_epochs = 100
for epoch in range(n_epochs):
for batch_X, batch_y in dataloader:
optimizer.zero_grad()
outputs = model(batch_X)
# 重塑输出
outputs = outputs.view(-1, pred_len, 10)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
if (epoch+1) % 20 == 0:
print(f'Epoch [{epoch+1}/{n_epochs}], Loss: {loss.item():.6f}')
22.3.2 Transformer:捕捉全局依赖
LSTM有个问题:它是按顺序处理的,无法并行。而且对于长序列,它容易遗忘早期信息。
Transformer通过自注意力机制,可以同时关注序列中所有位置的信息。
import torch.nn.functional as F
class SurfaceTransformer(nn.Module):
def __init__(self, input_size=10, d_model=64, nhead=4, num_layers=2):
super().__init__()
self.input_proj = nn.Linear(input_size, d_model)
self.pos_encoder = nn.Parameter(torch.randn(1, 100, d_model) * 0.1)
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model,
nhead=nhead,
dim_feedforward=256,
dropout=0.1,
batch_first=True
)
self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
self.output_proj = nn.Linear(d_model, input_size)
def forward(self, x):
# x: (batch, seq_len, input_size)
batch_size, seq_len, _ = x.shape
# 投影到d_model维度
x = self.input_proj(x)
# 加位置编码
x = x + self.pos_encoder[:, :seq_len, :]
# Transformer编码
x = self.transformer(x)
# 取最后一个时间步
x = x[:, -1, :]
# 投影回原始维度
x = self.output_proj(x)
return x
# 使用示例
transformer_model = SurfaceTransformer(input_size=10, d_model=64, nhead=4, num_layers=2)
# 训练过程与LSTM类似,这里省略
22.4 模型对比与选择
说了这么多,到底该用哪个?
我个人的选择逻辑是这样的:
| 模型 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| ARIMA/VAR | 简单、可解释性强、计算快 | 线性假设、参数固定 | 低频交易、因子分析 |
| 状态空间模型 | 时变参数、实时更新、鲁棒性好 | 实现复杂、需要调参 | 中高频交易、做市 |
| LSTM | 非线性、捕捉长期依赖 | 数据需求大、可解释性差 | 高频交易、模式识别 |
| Transformer | 并行计算、全局依赖 | 小数据过拟合、计算资源高 | 大数据量、复杂形态 |
22.5 实战:构建预测信号
最后,我们把这些模型串起来,构建一个完整的预测信号流程。
class SurfacePredictor:
def __init__(self, model_type='lstm'):
self.model_type = model_type
self.model = None
self.factors = None
def extract_factors(self, surface_data):
"""从曲面数据中提取因子"""
# 使用PCA或Nelson-Siegel分解
from sklearn.decomposition import PCA
pca = PCA(n_components=3)
factors = pca.fit_transform(surface_data)
return factors, pca
def predict(self, surface_data, horizon=5):
"""预测未来horizon个时间步的曲面"""
if self.model_type == 'arima':
# 用VAR预测因子
factors, pca = self.extract_factors(surface_data)
# ... 拟合VAR模型并预测
predicted_factors = self._var_predict(factors, horizon)
# 重构曲面
predicted_surface = pca.inverse_transform(predicted_factors)
elif self.model_type == 'lstm':
# 直接用LSTM预测曲面
# ... 加载训练好的LSTM模型
predicted_surface = self._lstm_predict(surface_data, horizon)
return predicted_surface
def generate_signal(self, current_surface, predicted_surface):
"""生成交易信号"""
# 计算预测变化
delta = predicted_surface - current_surface
# 寻找套利机会
# 比如:预测曲面变陡,则做多短期波动率,做空长期波动率
signals = []
if delta.mean() > 0.01: # 整体波动率上升
signals.append('buy_vol')
if delta[:, -1].mean() - delta[:, 0].mean() > 0.02: # 斜率变陡
signals.append('steepener')
return signals
# 使用示例
predictor = SurfacePredictor(model_type='lstm')
current_surface = observations[-20:] # 最近20个时间步
predicted = predictor.predict(current_surface, horizon=5)
signals = predictor.generate_signal(current_surface[-1], predicted[-1])
print(f'生成的信号: {signals}')
好了,这一章的内容就到这里。曲面形态预测是个大话题,我尽量把三个主流方向都覆盖到了。记住,没有最好的模型,只有最适合当前市场的模型。多试试,多对比,找到你自己的「屠龙刀」。