11.自定义模型开发
本文详细介绍了在 QLib 量化投资平台中开发自定义模型的完整流程，包括模型接口规范、实现步骤、保存加载方法以及集成策略，帮助开发者将自己的预测模型无缝整合到 QLib 的工作流中。
模型接口规范
在 QLib 中开发自定义模型，首先需要了解平台的模型接口规范。所有模型都必须继承自 qlib.model.base.Model 基类，并实现核心方法。这个设计确保了不同模型能够以统一的方式在 QLib 的工作流中运行，无论是训练、预测还是评估。
核心接口定义
Model 基类定义了以下必须实现的接口：
__init__(self, **kwargs): 初始化模型参数fit(self, dataset): 训练模型，接收数据集对象predict(self, dataset): 预测方法，返回预测结果
可选实现的接口包括：
finetune(self, dataset): 模型微调方法，继承自ModelFT基类save(self, path): 模型保存方法load(self, path): 模型加载方法
这些接口的设计遵循了机器学习的基本流程，同时考虑了量化投资中的特殊需求，比如数据集的处理方式和预测结果的格式要求。
数据集接口
自定义模型的 fit 和 predict 方法都接收 DatasetH 对象作为参数。这个对象封装了特征数据和标签数据，提供了统一的数据访问接口。通过 dataset.prepare() 方法可以获取不同阶段（训练、验证、测试）的数据：
# 准备训练和验证数据
df_train, df_valid = dataset.prepare(
["train", "valid"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
)
x_train, y_train = df_train["feature"], df_train["label"]
x_valid, y_valid = df_valid["feature"], df_valid["label"]
这种数据准备方式确保了不同模型能够使用统一格式的输入数据，降低了集成难度。
自定义模型实现步骤
开发自定义模型通常需要以下几个步骤：定义模型类、实现核心方法、配置模型参数和测试模型功能。下面以一个简单的线性回归模型为例，详细说明每个步骤的具体实现。
定义模型类
首先创建一个继承自 qlib.model.base.Model 的类，并初始化必要的参数。在 __init__ 方法中，我们可以接收模型的超参数，并进行必要的初始化：
import numpy as np
import pandas as pd
from sklearn.linear_model import LinearRegression
from qlib.model.base import Model
from qlib.data.dataset import DatasetH
from qlib.data.handler import DataHandlerLP
class LinearModel(Model):
def __init__(self, fit_intercept=True, normalize=False, **kwargs):
# 初始化超参数
self.fit_intercept = fit_intercept
self.normalize = normalize
self.model = None
# 存储其他参数
self.kwargs = kwargs
def fit(self, dataset: DatasetH, **kwargs):
# 准备训练数据
df_train, df_valid = dataset.prepare(
["train", "valid"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
)
x_train, y_train = df_train["feature"], df_train["label"]
x_valid, y_valid = df_valid["feature"], df_valid["label"]
# 处理标签维度
if y_train.values.ndim == 2 and y_train.values.shape[1] == 1:
y_train = np.squeeze(y_train.values)
y_valid = np.squeeze(y_valid.values)
# 初始化并训练模型
self.model = LinearRegression(
fit_intercept=self.fit_intercept,
normalize=self.normalize
)
self.model.fit(x_train.values, y_train)
# 可以添加模型评估逻辑
train_score = self.model.score(x_train.values, y_train)
valid_score = self.model.score(x_valid.values, y_valid)
print(f"Train R^2 score: {train_score:.4f}")
print(f"Valid R^2 score: {valid_score:.4f}")
def predict(self, dataset: DatasetH, **kwargs) -> pd.Series:
if self.model is None:
raise ValueError("模型尚未训练，请先调用fit方法")
# 准备测试数据
x_test = dataset.prepare("test", col_set="feature", data_key=DataHandlerLP.DK_I)
# 进行预测
preds = self.model.predict(x_test.values)
# 返回Series格式的预测结果，索引与输入数据一致
return pd.Series(preds, index=x_test.index)
这段代码实现了一个基于 scikit-learn 的线性回归模型，遵循了 QLib 的模型接口规范。在 fit 方法中，我们从 DatasetH 对象中获取训练和验证数据，然后使用这些数据训练模型。在 predict 方法中，我们获取测试数据并返回预测结果。
实现模型微调（可选）
如果需要实现模型微调功能，可以继承 ModelFT 基类并实现 finetune 方法。以下是一个简单的示例：
from qlib.model.base import ModelFT
class LinearModelFT(ModelFT):
# 继承前面定义的LinearModel
def finetune(self, dataset: DatasetH, epochs=10, **kwargs):
# 获取微调数据
df_train, _ = dataset.prepare(
["train"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
)
x_train, y_train = df_train["feature"], df_train["label"]
if y_train.values.ndim == 2 and y_train.values.shape[1] == 1:
y_train = np.squeeze(y_train.values)
# 简单的微调策略：增加训练轮次
for i in range(epochs):
self.model.fit(x_train.values, y_train)
score = self.model.score(x_train.values, y_train)
if (i+1) % 5 == 0:
print(f"Finetune epoch {i+1}, R^2 score: {score:.4f}")
这个微调方法通过多轮训练来优化模型参数，适用于需要增量更新的场景。
模型保存与加载
在实际应用中，训练好的模型需要保存到磁盘，以便后续使用。QLib 并没有强制规定模型的保存格式，开发者可以根据自己的需求选择合适的方法。常用的模型保存方法包括使用 pickle 模块或特定框架的保存接口。
实现模型保存功能
我们可以为自定义模型添加 save 和 load 方法，用于模型的持久化：
import pickle
import os
class LinearModel(Model):
# ... 前面的代码 ...
def save(self, path: str):
"""保存模型到指定路径"
if self.model is None:
raise ValueError("模型尚未训练，无法保存")
# 创建目录（如果不存在）
os.makedirs(os.path.dirname(path), exist_ok=True)
# 保存模型参数
with open(path, 'wb') as f:
pickle.dump({
'fit_intercept': self.fit_intercept,
'normalize': self.normalize,
'model': self.model,
'kwargs': self.kwargs
}, f)
def load(self, path: str):
"""从指定路径加载模型"
with open(path, 'rb') as f:
data = pickle.load(f)
self.fit_intercept = data['fit_intercept']
self.normalize = data['normalize']
self.model = data['model']
self.kwargs = data['kwargs']
return self
这段代码使用 Python 的 pickle 模块来序列化模型对象。在保存时，我们将模型的超参数和训练好的模型一起保存；在加载时，我们恢复这些参数并重建模型。
模型保存与加载示例
使用上述方法保存和加载模型的示例代码如下：
# 训练模型
dataset = ... # 准备数据集\mmodel = LinearModel(fit_intercept=True, normalize=False)
model.fit(dataset)
# 保存模型
model.save('./models/linear_model.pkl')
# 加载模型
loaded_model = LinearModel().load('./models/linear_model.pkl')
# 使用加载的模型进行预测
preds = loaded_model.predict(dataset)
这种方法简单有效，适用于大多数基于 scikit-learn 的模型。对于深度学习模型，可能需要使用框架特定的保存方法，如 PyTorch 的 torch.save() 或 TensorFlow 的 model.save()。
模型集成与融合策略
在量化投资中，单一模型的预测能力往往有限。通过模型集成或融合，可以结合多个模型的优势，提高预测性能。QLib 提供了灵活的接口来支持模型集成，开发者可以根据自己的需求实现不同的集成策略。
简单加权集成
最简单的集成方法是对多个模型的预测结果进行加权平均。以下是一个实现示例：
class EnsembleModel(Model):
def __init__(self, models, weights=None, **kwargs):
"""
初始化集成模型
models: 模型列表
weights: 权重列表，若为None则等权重
"""
self.models = models
self.weights = weights if weights is not None else [1/len(models)]*len(models)
self.kwargs = kwargs
def fit(self, dataset: DatasetH, **kwargs):
# 训练所有模型
for model in self.models:
model.fit(dataset, **kwargs)
def predict(self, dataset: DatasetH, **kwargs) -> pd.Series:
# 获取所有模型的预测结果
predictions = []
for model in self.models:
pred = model.predict(dataset, **kwargs)
predictions.append(pred)
# 加权平均
result = pd.DataFrame(predictions).T.dot(self.weights)
return result
使用这个集成模型的示例代码如下：
# 创建基础模型
model1 = LinearModel(fit_intercept=True)
model2 = LGBModel(loss='mse') # 假设已经实现了LGBModel
# 创建集成模型
ensemble = EnsembleModel(models=[model1, model2], weights=[0.4, 0.6])
# 训练和预测
ensemble.fit(dataset)
preds = ensemble.predict(dataset)
stacking 集成
Stacking 是一种更复杂的集成方法，它使用元模型来学习如何组合基础模型的预测结果。以下是一个简单的 stacking 实现：
class StackingModel(Model):
def __init__(self, base_models, meta_model, **kwargs):
self.base_models = base_models
self.meta_model = meta_model
self.kwargs = kwargs
def fit(self, dataset: DatasetH, **kwargs):
# 准备训练数据
df_train, df_valid = dataset.prepare(
["train", "valid"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
)
# 训练基础模型并生成元特征
meta_features = []
for model in self.base_models:
model.fit(dataset)
pred = model.predict(dataset)
meta_features.append(pred)
# 训练元模型
X_meta = pd.DataFrame(meta_features).T
y_meta = df_valid['label'] # 使用验证集标签训练元模型
self.meta_model.fit(X_meta, y_meta)
def predict(self, dataset: DatasetH, **kwargs) -> pd.Series:
# 生成基础模型预测
meta_features = []
for model in self.base_models:
pred = model.predict(dataset)
meta_features.append(pred)
# 元模型预测
X_meta = pd.DataFrame(meta_features).T
return pd.Series(self.meta_model.predict(X_meta), index=X_meta.index)
Stacking 通常能获得比简单加权更好的性能，但实现复杂度和计算成本也更高。在实际应用中，需要根据数据特点和计算资源选择合适的集成策略。
配置文件编写
要将自定义模型集成到 QLib 的工作流中，需要编写相应的配置文件。配置文件描述了模型的类路径、参数以及数据集等信息，使 QLib 能够自动加载和运行模型。
配置文件结构
一个典型的模型配置文件如下所示：
model:
class: LinearModel
module_path: mymodels.linear # 模型类所在的模块路径
args:
fit_intercept: true
normalize: false
dataset:
class: DatasetH
module_path: qlib.data.dataset
kwargs:
handler:
class: Alpha158
module_path: qlib.contrib.data.handler
kwargs:
start_time: "2008-01-01"
end_time: "2020-08-01"
fit_start_time: "2008-01-01"
fit_end_time: "2014-12-31"
instruments: csi300
segments:
train: ("2008-01-01", "2014-12-31")
valid: ("2015-01-01", "2016-12-31")
test: ("2017-01-01", "2020-08-01")
在这个配置中，model 部分指定了自定义模型的类名和模块路径，以及模型的超参数。dataset 部分定义了数据集的配置，包括数据处理器、时间区间和标的范围等。
使用 qrun 命令运行模型
编写完配置文件后，可以使用 QLib 提供的 qrun 命令来运行模型：
cd examples
qrun my_config.yaml
这条命令会自动加载配置文件中指定的模型和数据集，执行训练和预测，并生成相应的记录和报告。
模型测试与验证
开发自定义模型后，需要进行充分的测试和验证，确保其在 QLib 中能够正常工作。测试通常包括单元测试、集成测试和性能测试三个层面。
单元测试
单元测试主要验证模型的各个方法是否正常工作。以下是一个使用 pytest 的单元测试示例：
import pytest
from qlib.init import init
from qlib.data.dataset import DatasetH
from mymodels.linear import LinearModel
@pytest.fixture(scope="module")
def init_qlib():
init(provider_uri="~/.qlib/qlib_data/cn_data")
@pytest.fixture
def dataset():
handler_config = {
"start_time": "2008-01-01",
"end_time": "2020-08-01",
"fit_start_time": "2008-01-01",
"fit_end_time": "2014-12-31",
"instruments": "csi300",
}
dataset_config = {
"class": "DatasetH",
"module_path": "qlib.data.dataset",
"kwargs": {
"handler": {
"class": "Alpha158",
"module_path": "qlib.contrib.data.handler",
"kwargs": handler_config,
},
"segments": {
"train": ("2008-01-01", "2014-12-31"),
"valid": ("2015-01-01", "2016-12-31"),
"test": ("2017-01-01", "2020-08-01"),
},
},
}
return DatasetH(** dataset_config["kwargs"])
def test_linear_model(init_qlib, dataset):
model = LinearModel(fit_intercept=True, normalize=False)
# 测试训练
model.fit(dataset)
assert model.model is not None
# 测试预测
preds = model.predict(dataset)
assert isinstance(preds, pd.Series)
assert len(preds) > 0
# 测试保存和加载
model.save("./test_model.pkl")
loaded_model = LinearModel().load("./test_model.pkl")
loaded_preds = loaded_model.predict(dataset)
pd.testing.assert_series_equal(preds, loaded_preds, check_names=False)
这个测试用例验证了模型的训练、预测、保存和加载功能是否正常工作。
集成测试
集成测试验证模型是否能够融入 QLib 的工作流。可以通过运行完整的工作流来进行测试：
from qlib.workflow import R
from qlib.utils import init_instance_by_config
# 加载配置
config = {
"model": {
"class": "LinearModel",
"module_path": "mymodels.linear",
"args": {
"fit_intercept": True,
"normalize": False
}
},
"dataset": dataset_config # 与前面定义的dataset_config相同
}
# 初始化模型和数据集
model = init_instance_by_config(config["model"])
dataset = init_instance_by_config(config["dataset"])
# 执行工作流
with R.start(experiment_name="linear_model_test"):
model.fit(dataset)
recorder = R.get_recorder()
sr = SignalRecord(model, dataset, recorder)
sr.generate()
# 验证记录是否生成
assert "pred_score" in recorder.list_records()
这个测试验证了模型能否在 QLib 的工作流中正常训练并生成预测记录。
常见问题与解决方案
在开发自定义模型的过程中，可能会遇到各种问题。以下是一些常见问题及其解决方案：
数据格式不匹配
问题：模型训练时出现数据格式错误，如特征和标签的维度不匹配。
解决方案：使用 dataset.prepare() 方法获取正确格式的数据，并确保特征和标签的维度一致。可以使用 print(x_train.shape, y_train.shape) 来检查数据维度。
模型无法保存或加载
问题：使用 pickle 保存包含大型模型的对象时失败，或加载后模型无法正常工作。
解决方案：对于大型模型，可以考虑只保存模型参数而不是整个模型对象。例如，对于 PyTorch 模型，可以只保存 state_dict：
# 保存PyTorch模型参数
torch.save(self.model.state_dict(), path)
# 加载模型参数
self.model.load_state_dict(torch.load(path))
性能问题
问题：模型训练或预测速度慢，无法处理大规模数据。
解决方案：
- 使用数据缓存机制，避免重复数据处理
- 优化特征工程，减少特征数量
- 使用并行计算或 GPU 加速
- 考虑使用更高效的算法或框架
总结
本章详细介绍了在 QLib 中开发自定义模型的全过程，包括接口规范、实现步骤、保存加载、集成策略、配置编写和测试验证。通过遵循这些步骤，开发者可以将自己的模型无缝集成到 QLib 的量化投资工作流中，充分利用 QLib 的数据处理和回测功能。
自定义模型开发是量化投资研究中的重要环节，它允许开发者将最新的机器学习算法应用到实际的投资策略中。随着 AI 技术的不断发展，相信会有越来越多的创新模型被应用到量化投资领域，为投资者带来更好的收益。
在实际开发中，建议先从简单模型开始，逐步尝试更复杂的算法和集成策略。同时，要注重模型的可解释性和稳定性，这在量化投资中同样重要。最后，持续关注 QLib 的更新，以便及时利用平台提供的新功能和优化。