【期货量化进阶】期货量化交易策略模型集成方法（Python量化）

一、前言

单一模型往往存在局限性，模型集成可以结合多个模型的优势，提高预测准确性和策略稳定性。模型集成是机器学习在量化交易中的重要应用。本文将介绍如何实现模型集成。

本文将介绍：

• 模型集成的基本原理
• 投票集成方法
• 加权集成方法
• 堆叠集成方法
• 动态集成方法

二、为什么选择天勤量化（TqSdk）

TqSdk模型集成支持：

安装方法：

pip install tqsdk pandas numpy scikit-learn xgboost

三、模型集成基础

3.1 集成方法

3.2 集成优势

四、投票集成

4.1 简单投票

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：简单投票集成
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import LinearRegression
from tqsdk import TqApi, TqAuth
from tqsdk.tafunc import ma, macd, rsi

api = TqApi(auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
klines = api.get_kline_serial(SYMBOL, 3600, 2000)

api.wait_update()

def build_features(df):
"""构建特征"""
features = pd.DataFrame()
features['return_5'] = df['close'].pct_change(5)
features['return_20'] = df['close'].pct_change(20)
features['ma5'] = ma(df['close'], 5)
features['ma20'] = ma(df['close'], 20)
features['close_ma20_ratio'] = df['close'] / features['ma20'] – 1

macd_data = macd(df['close'], 12, 26, 9)
features['macd'] = macd_data['macd']
features['rsi'] = rsi(df['close'], 14)

features['volatility'] = df['close'].pct_change().rolling(20).std()
return features

# 准备数据
features = build_features(klines)
forward_return = klines['close'].shift(–5) / klines['close'] – 1

data = pd.concat([features, pd.Series(forward_return, name='target')], axis=1)
data = data.dropna()

X = data.iloc[:, :–1].values
y = data['target'].values

# 划分训练集和测试集
split_idx = int(len(X) * 0.8)
X_train, X_test = X[:split_idx], X[split_idx:]
y_train, y_test = y[:split_idx], y[split_idx:]

# 训练多个模型
models = {
'rf': RandomForestRegressor(n_estimators=50, random_state=42),
'gb': GradientBoostingRegressor(n_estimators=50, random_state=42),
'lr': LinearRegression()
}

for name, model in models.items():
model.fit(X_train, y_train)

# 投票集成（回归问题用平均）
predictions = {}
for name, model in models.items():
predictions[name] = model.predict(X_test)

# 集成预测（简单平均）
ensemble_pred = np.mean([predictions[name] for name in models.keys()], axis=0)

# 评估
from sklearn.metrics import mean_squared_error, r2_score

print("模型集成结果:")
for name in models.keys():
mse = mean_squared_error(y_test, predictions[name])
print(f" {name} MSE: {mse:.6f}")

ensemble_mse = mean_squared_error(y_test, ensemble_pred)
print(f" 集成模型 MSE: {ensemble_mse:.6f}")

api.close()

五、加权集成

5.1 性能加权

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：性能加权集成
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
from tqsdk import TqApi, TqAuth

class WeightedEnsemble:
"""加权集成"""

def __init__(self, models):
"""
初始化

参数:
models: 模型字典
"""
self.models = models
self.weights = None

def fit(self, X_train, y_train, X_val, y_val):
"""训练模型并计算权重"""
# 训练所有模型
for name, model in self.models.items():
model.fit(X_train, y_train)

# 在验证集上评估
val_errors = {}
for name, model in self.models.items():
val_pred = model.predict(X_val)
mse = mean_squared_error(y_val, val_pred)
val_errors[name] = mse

# 根据误差计算权重（误差越小，权重越大）
total_inv_error = sum(1 / (error + 1e-6) for error in val_errors.values())
self.weights = {
name: (1 / (error + 1e-6)) / total_inv_error
for name, error in val_errors.items()
}

def predict(self, X):
"""加权预测"""
predictions = {}
for name, model in self.models.items():
predictions[name] = model.predict(X)

# 加权平均
ensemble_pred = np.zeros(len(X))
for name, weight in self.weights.items():
ensemble_pred += predictions[name] * weight

return ensemble_pred

# 使用示例
api = TqApi(auth=TqAuth("快期账户", "快期密码"))
SYMBOL = "SHFE.rb2510"
klines = api.get_kline_serial(SYMBOL, 3600, 2000)
api.wait_update()

from feature_builder import build_features

features = build_features(klines)
forward_return = klines['close'].shift(–5) / klines['close'] – 1

data = pd.concat([features, pd.Series(forward_return, name='target')], axis=1)
data = data.dropna()

X = data.iloc[:, :–1].values
y = data['target'].values

# 划分数据
split1 = int(len(X) * 0.6)
split2 = int(len(X) * 0.8)
X_train, X_val, X_test = X[:split1], X[split1:split2], X[split2:]
y_train, y_val, y_test = y[:split1], y[split1:split2], y[split2:]

# 创建集成模型
models = {
'rf': RandomForestRegressor(n_estimators=50, random_state=42),
'gb': GradientBoostingRegressor(n_estimators=50, random_state=42),
'lr': LinearRegression()
}

ensemble = WeightedEnsemble(models)
ensemble.fit(X_train, y_train, X_val, y_val)

# 预测
ensemble_pred = ensemble.predict(X_test)

# 评估
mse = mean_squared_error(y_test, ensemble_pred)
print(f"加权集成 MSE: {mse:.6f}")
print(f"\\n模型权重:")
for name, weight in ensemble.weights.items():
print(f" {name}: {weight:.4f}")

api.close()

六、堆叠集成

6.1 堆叠模型

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：堆叠集成
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import LinearRegression, Ridge
from sklearn.model_selection import KFold
from tqsdk import TqApi, TqAuth

class StackingEnsemble:
"""堆叠集成"""

def __init__(self, base_models, meta_model):
"""
初始化

参数:
base_models: 基础模型字典
meta_model: 元模型
"""
self.base_models = base_models
self.meta_model = meta_model

def fit(self, X, y, n_folds=5):
"""训练堆叠模型"""
kf = KFold(n_splits=n_folds, shuffle=False)

# 训练基础模型
for name, model in self.base_models.items():
model.fit(X, y)

# 生成元特征
meta_features = np.zeros((len(X), len(self.base_models)))

for fold_idx, (train_idx, val_idx) in enumerate(kf.split(X)):
X_train_fold, X_val_fold = X[train_idx], X[val_idx]
y_train_fold, y_val_fold = y[train_idx], y[val_idx]

for model_idx, (name, model) in enumerate(self.base_models.items()):
# 在fold上训练
model.fit(X_train_fold, y_train_fold)
# 在验证集上预测
val_pred = model.predict(X_val_fold)
meta_features[val_idx, model_idx] = val_pred

# 训练元模型
self.meta_model.fit(meta_features, y)

def predict(self, X):
"""预测"""
# 基础模型预测
base_predictions = np.zeros((len(X), len(self.base_models)))
for idx, (name, model) in enumerate(self.base_models.items()):
base_predictions[:, idx] = model.predict(X)

# 元模型预测
meta_pred = self.meta_model.predict(base_predictions)
return meta_pred

# 使用示例
api = TqApi(auth=TqAuth("快期账户", "快期密码"))
SYMBOL = "SHFE.rb2510"
klines = api.get_kline_serial(SYMBOL, 3600, 2000)
api.wait_update()

from feature_builder import build_features

features = build_features(klines)
forward_return = klines['close'].shift(–5) / klines['close'] – 1

data = pd.concat([features, pd.Series(forward_return, name='target')], axis=1)
data = data.dropna()

X = data.iloc[:, :–1].values
y = data['target'].values

# 划分数据
split_idx = int(len(X) * 0.8)
X_train, X_test = X[:split_idx], X[split_idx:]
y_train, y_test = y[:split_idx], y[split_idx:]

# 创建堆叠模型
base_models = {
'rf': RandomForestRegressor(n_estimators=50, random_state=42),
'gb': GradientBoostingRegressor(n_estimators=50, random_state=42),
'lr': LinearRegression()
}
meta_model = Ridge(alpha=1.0)

ensemble = StackingEnsemble(base_models, meta_model)
ensemble.fit(X_train, y_train, n_folds=5)

# 预测
ensemble_pred = ensemble.predict(X_test)

# 评估
from sklearn.metrics import mean_squared_error
mse = mean_squared_error(y_test, ensemble_pred)
print(f"堆叠集成 MSE: {mse:.6f}")

api.close()

七、动态集成

7.1 动态模型选择

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：动态模型选择
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
from tqsdk import TqApi, TqAuth

class DynamicEnsemble:
"""动态集成"""

def __init__(self, models, window=100):
"""
初始化

参数:
models: 模型字典
window: 评估窗口大小
"""
self.models = models
self.window = window
self.recent_errors = {name: [] for name in models.keys()}

def update_performance(self, X, y_true, y_pred_dict):
"""
更新模型性能

参数:
X: 特征
y_true: 真实值
y_pred_dict: 预测值字典
"""
for name, y_pred in y_pred_dict.items():
error = mean_squared_error([y_true], [y_pred])
self.recent_errors[name].append(error)

# 保持窗口大小
if len(self.recent_errors[name]) > self.window:
self.recent_errors[name].pop(0)

def select_best_model(self):
"""选择最佳模型"""
if not all(len(errors) > 0 for errors in self.recent_errors.values()):
# 如果还没有足够数据，返回第一个模型
return list(self.models.keys())[0]

# 计算平均误差
avg_errors = {
name: np.mean(errors)
for name, errors in self.recent_errors.items()
}

# 选择误差最小的模型
best_model = min(avg_errors.items(), key=lambda x: x[1])[0]
return best_model

def predict(self, X, use_ensemble=True):
"""
预测

参数:
X: 特征
use_ensemble: 是否使用集成（否则用最佳模型）
"""
predictions = {}
for name, model in self.models.items():
predictions[name] = model.predict(X.reshape(1, –1))[0]

if use_ensemble and all(len(errors) >= 10 for errors in self.recent_errors.values()):
# 使用加权集成
avg_errors = {
name: np.mean(errors) if len(errors) > 0 else 1.0
for name, errors in self.recent_errors.items()
}
total_inv_error = sum(1 / (error + 1e-6) for error in avg_errors.values())
weights = {
name: (1 / (error + 1e-6)) / total_inv_error
for name, error in avg_errors.items()
}

ensemble_pred = sum(predictions[name] * weights[name] for name in self.models.keys())
return ensemble_pred
else:
# 使用最佳模型
best_model = self.select_best_model()
return predictions[best_model]

# 使用示例
api = TqApi(auth=TqAuth("快期账户", "快期密码"))
SYMBOL = "SHFE.rb2510"
klines = api.get_kline_serial(SYMBOL, 3600, 2000)
api.wait_update()

from feature_builder import build_features

features = build_features(klines)
forward_return = klines['close'].shift(–5) / klines['close'] – 1

data = pd.concat([features, pd.Series(forward_return, name='target')], axis=1)
data = data.dropna()

X = data.iloc[:, :–1].values
y = data['target'].values

# 划分数据
split_idx = int(len(X) * 0.8)
X_train, X_test = X[:split_idx], X[split_idx:]
y_train, y_test = y[:split_idx], y[split_idx:]

# 训练模型
models = {
'rf': RandomForestRegressor(n_estimators=50, random_state=42),
'gb': GradientBoostingRegressor(n_estimators=50, random_state=42),
'lr': LinearRegression()
}

for name, model in models.items():
model.fit(X_train, y_train)

# 创建动态集成
ensemble = DynamicEnsemble(models, window=50)

# 在线更新和预测
predictions = []
for i in range(len(X_test)):
X_sample = X_test[i]
y_true = y_test[i]

# 预测
y_pred_dict = {name: model.predict(X_sample.reshape(1, –1))[0] for name, model in models.items()}
y_pred = ensemble.predict(X_sample, use_ensemble=True)
predictions.append(y_pred)

# 更新性能
ensemble.update_performance(X_sample, y_true, y_pred_dict)

# 评估
mse = mean_squared_error(y_test, predictions)
print(f"动态集成 MSE: {mse:.6f}")

api.close()

八、常见问题

Q1: 模型集成一定比单模型好吗？

A: 不一定，取决于：

• 基础模型是否互补
• 集成方法是否合适
• 是否过拟合
• 计算成本是否值得

Q2: 如何选择集成方法？

A: 建议：

• 简单场景：投票或加权
• 复杂场景：堆叠
• 实时应用：动态选择
• 需要解释：投票

Q3: 集成模型容易过拟合吗？

A: 可能，需要注意：

• 使用交叉验证
• 限制模型数量
• 使用正则化
• 样本外验证

九、总结

下一步学习建议：

免责声明：本文仅供学习交流使用，不构成任何投资建议。期货交易有风险，入市需谨慎。

更多资源：

• 天勤量化官网：https://www.shinnytech.com
• GitHub开源地址：https://github.com/shinnytech/tqsdk-python
• 官方文档：https://doc.shinnytech.com/tqsdk/latest