10.高级数据处理
本文详细介绍 QLib 量化投资平台中的高级数据处理技术，包括数据缓存机制、数据复用与性能优化方法、自定义数据处理器的实现，以及高频数据处理的关键技巧。通过本章内容，读者将能够掌握大规模金融数据的高效处理方法，提升量化策略研发的效率和性能。
数据缓存机制
在量化研究中，数据处理往往是整个工作流中最耗时的环节之一。特别是当我们需要反复进行特征计算、数据转换或模型训练时，重复处理相同的数据会严重影响研发效率。QLib 提供了完善的数据缓存机制，能够有效减少重复计算，显著提升数据访问速度。
QLib 缓存系统架构
QLib 的缓存系统主要由三个核心组件构成：内存缓存（MemCache）、表达式缓存（ExpressionCache）和数据集缓存（DatasetCache）。这三个组件协同工作，形成了一个多层次的缓存架构，能够满足不同场景下的数据缓存需求。
flowchart TD
A[数据源] --> B[内存缓存 MemCache]
B --> C[表达式缓存 ExpressionCache]
C --> D[数据集缓存 DatasetCache]
D --> E[模型训练/回测]
B --> F{数据是否存在?}
F -->|是| G[直接返回]
F -->|否| H[从磁盘加载]
内存缓存（MemCache）是缓存系统的第一层，主要用于缓存频繁访问的基础数据，如交易日历、股票列表和基础特征。它由三个 MemCacheUnit 实例组成，分别缓存 Calendar（交易日历）、Instruments（股票列表）和 Features（特征数据）。我们可以通过全局变量 H 来访问这些缓存单元：
from qlib.data.cache import H
# 获取交易日历缓存
calendar_cache = H['c']
# 获取股票列表缓存
instruments_cache = H['i']
# 获取特征数据缓存
features_cache = H['f']
表达式缓存（ExpressionCache）用于缓存通过表达式计算得到的衍生特征，如移动平均、收益率等。QLib 提供了 DiskExpressionCache 实现，将计算结果存储在磁盘上，避免重复计算复杂的表达式。
数据集缓存（DatasetCache）则是针对整个数据集的缓存机制，它根据股票池配置、特征表达式列表、时间范围和频率等信息生成唯一的缓存标识，将预处理好的数据集存储起来，供后续模型训练和回测使用。
缓存配置与使用
在 QLib 中，我们可以通过初始化配置来控制缓存行为。例如，在使用 qrun 命令运行工作流时，可以在配置文件中设置缓存相关参数：
qlib_init:
expression_cache: disk
dataset_cache: disk
cache_path: ~/.qlib/cache
上述配置将表达式缓存和数据集缓存都设置为磁盘缓存，并指定了缓存文件的存储路径。这样，当我们第一次运行工作流时，QLib 会将计算得到的表达式结果和数据集保存到缓存中，后续运行时则会直接从缓存加载，大大提高效率。
除了通过配置文件，我们还可以在代码中直接使用缓存功能。例如，使用 DataHandlerLP 的 to_pickle 方法将数据处理器及其处理后的数据保存到磁盘：
from qlib.contrib.data.handler import Alpha158
data_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",
}
handler = Alpha158(**data_handler_config)
# 将数据处理器保存到磁盘
handler.to_pickle("handler.pkl", dump_all=True)
之后，我们可以直接从磁盘加载保存的数据处理器，而无需重新进行数据预处理：
import pickle
with open("handler.pkl", "rb") as f:
handler = pickle.load(f)
# 直接使用加载后的处理器获取数据
features = handler.fetch(col_set="feature")
labels = handler.fetch(col_set="label")
缓存失效与更新策略
虽然缓存能够显著提高数据访问速度，但也带来了缓存失效的问题。当原始数据发生变化或处理逻辑更新时，缓存中的数据可能会过时，导致后续分析结果不准确。因此，我们需要合理的缓存失效与更新策略。
QLib 采用了基于哈希的缓存失效机制。对于表达式缓存，QLib 会根据表达式内容、时间范围、频率等信息生成唯一的哈希值作为缓存键。当这些参数发生变化时，哈希值也会相应变化，从而自动使用新的缓存。
对于数据集缓存，QLib 会综合考虑股票池配置、特征列表、处理器链等因素生成缓存键。当任何一个因素发生变化时，都会触发新的缓存生成。
在实际应用中，我们可以通过以下几种方式手动控制缓存：
- 删除指定的缓存文件：当我们确定某些缓存已经过时，可以直接删除对应的缓存文件，QLib 会在下次访问时重新生成缓存。
- 调整缓存路径：通过修改配置文件中的 cache_path 参数，可以让 QLib 使用新的缓存目录，从而避免使用旧的缓存数据。
- 禁用缓存：在开发和调试阶段，我们可以通过将 expression_cache 和 dataset_cache 设置为 None 来禁用缓存，确保每次都使用最新的数据和处理逻辑。
数据复用与性能优化
在量化研究中，数据复用是提升效率的关键手段之一。通过合理复用已经处理好的数据，我们可以避免重复劳动，将更多精力集中在策略设计和模型优化上。QLib 提供了多种数据复用机制，并针对大规模数据处理进行了性能优化。
内存数据复用
QLib 的数据处理器（DataHandler）和数据集（Dataset）都是可序列化的，这意味着我们可以将它们保存到内存中，在不同的任务之间共享和复用。这种内存数据复用方式特别适用于需要多次使用相同数据的场景，如参数调优、模型比较等。
下面是一个内存数据复用的示例：
from qlib.data.dataset.handler import DataHandlerLP
from qlib.utils import init_instance_by_config
from qlib.model.trainer import task_train
# 初始化数据处理器
handler_config = {
"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",
}
}
handler = init_instance_by_config(handler_config)
# 复用数据处理器进行多次训练
for i in range(5):
task = {
"dataset": {
"class": "DatasetH",
"module_path": "qlib.data.dataset",
"kwargs": {
"handler": handler, # 复用handler
"segments": {
"train": ("2008-01-01", "2014-12-31"),
"test": ("2015-01-01", "2020-08-01"),
}
}
},
"model": {
"class": "LightGBMModel",
"module_path": "qlib.contrib.model.gbdt",
"kwargs": {
"learning_rate": 0.05 * (i+1), # 不同的学习率
"n_estimators": 100,
"num_leaves": 63,
}
}
}
task_train(task, experiment_name=f"exp_{i}")
在这个示例中，我们只初始化了一次数据处理器，然后在多次模型训练中复用这个处理器。这样可以避免每次训练都重新加载和处理数据，显著节省时间。
磁盘数据复用
除了内存复用，QLib 还支持将处理好的数据保存到磁盘，以便在不同的会话或工作流中复用。这种方式特别适用于需要长期保存数据处理结果，或在多个实验之间共享数据的场景。
我们可以使用 DataHandlerLP 的 to_pickle 方法将数据处理器及其处理后的数据保存到磁盘：
# 将数据处理器保存到磁盘
handler.to_pickle("alpha158_handler.pkl", dump_all=True)
之后，在其他工作流或会话中，我们可以直接加载这个保存好的处理器：
import pickle
# 从磁盘加载数据处理器
with open("alpha158_handler.pkl", "rb") as f:
handler = pickle.load(f)
# 使用加载的处理器创建数据集
dataset = DatasetH(handler=handler, segments={"train": ("2008-01-01", "2014-12-31")})
QLib 还提供了通过配置文件引用磁盘缓存的处理器的功能。例如，在工作流配置文件中，我们可以直接指定处理器的文件路径：
task:
dataset:
class: DatasetH
module_path: qlib.data.dataset
kwargs:
handler: file:///path/to/alpha158_handler.pkl
segments:
train: [2008-01-01, 2014-12-31]
test: [2015-01-01, 2020-08-01]
这种方式使得我们可以在不同的工作流之间共享预处理数据，避免重复劳动。
性能优化技巧
除了数据复用，QLib 还提供了多种性能优化技巧，帮助用户更高效地处理大规模金融数据。
- 数据类型优化
金融数据通常包含大量的数值型特征，使用合适的数据类型可以显著减少内存占用，提高处理速度。QLib 默认会将数据转换为合适的类型，但我们也可以手动指定数据类型：
# 在数据处理器配置中指定数据类型
handler_config = {
"class": "Alpha158",
"module_path": "qlib.contrib.data.handler",
"kwargs": {
# 其他配置...
"dtype": "float32", # 使用float32代替默认的float64
}
}
- 并行计算
QLib 的许多数据处理操作都支持并行计算，可以通过设置 n_jobs 参数来启用：
# 启用并行计算
handler_config = {
"class": "Alpha158",
"module_path": "qlib.contrib.data.handler",
"kwargs": {
# 其他配置...
"n_jobs": 8, # 使用8个进程并行处理
}
}
- 按需加载
对于大规模数据集，我们可以采用按需加载的方式，只加载当前需要的数据，而不是一次性加载全部数据到内存：
# 按需加载数据
dataset = DatasetH(handler=handler, segments={"train": ("2008-01-01", "2014-12-31")})
# 迭代获取批次数据
for batch in dataset.batch_iterator(batch_size=1024):
x, y = batch
# 处理批次数据...
- 数据预处理优化
合理设计数据预处理流程也可以显著提升性能。例如，将耗时的预处理步骤（如特征标准化）放在数据处理器中，利用缓存机制避免重复计算。
自定义数据处理器
虽然 QLib 提供了丰富的内置数据处理器，但在实际量化研究中，我们经常需要根据特定的策略需求实现自定义的数据处理逻辑。QLib 的 Processor 接口设计灵活，使得用户可以方便地扩展和定制数据处理流程。
Processor 接口详解
QLib 中的 Processor 是实现数据处理逻辑的基本单元。所有的处理器都继承自基类 qlib.data.dataset.processor.Processor，该基类定义了以下核心方法：
- fit(data): 拟合处理器，根据输入数据学习处理参数（如标准化的均值和标准差）。
- process(data): 对数据进行处理，返回处理后的数据。
- fit_process(data): 先拟合后处理，等价于 fit(data) + process(data)。
- get_params(): 获取处理器的参数。
- set_params(params): 设置处理器的参数。
根据处理逻辑是否需要拟合参数，Processor 可以分为无状态处理器和有状态处理器。无状态处理器（如 Fillna）不需要拟合，直接对数据进行处理；有状态处理器（如 ZscoreNorm）则需要先拟合数据，学习处理参数，然后才能应用于新数据。
自定义处理器实现步骤
实现自定义数据处理器通常需要以下几个步骤：
- 继承 Processor 基类
- 根据需要实现 fit 和 process 方法
- 注册自定义处理器（可选）
- 在数据处理器配置中使用自定义处理器
下面我们以一个简单的自定义处理器为例，详细说明实现过程。假设我们需要实现一个将特征值缩放到[0, 1]区间的 MinMax 缩放处理器。
import numpy as np
from qlib.data.dataset.processor import Processor
class MinMaxScaler(Processor):
def __init__(self, feature_group="feature"):
self.feature_group = feature_group
self.min_val = None
self.max_val = None
def fit(self, data):
# 只对指定的特征组进行拟合
if self.feature_group in data:
features = data[self.feature_group]
self.min_val = features.min()
self.max_val = features.max()
return self
def process(self, data):
if self.min_val is None or self.max_val is None:
raise ValueError("Processor has not been fitted. Call fit() first.")
# 对数据进行深拷贝，避免修改原始数据
data = data.copy()
if self.feature_group in data:
features = data[self.feature_group]
# 处理除零情况
if self.max_val == self.min_val:
data[self.feature_group] = 0
else:
data[self.feature_group] = (features - self.min_val) / (self.max_val - self.min_val)
return data
在这个示例中，我们实现了一个 MinMaxScaler 处理器，它会将指定特征组的特征值缩放到[0, 1]区间。fit 方法计算特征的最小值和最大值，process 方法则使用这些值对数据进行缩放。
处理器链的构建与使用
在实际应用中，我们通常需要将多个处理器组合起来形成处理器链，按顺序对数据进行多步处理。QLib 支持通过配置文件或代码方式构建处理器链。
通过配置文件构建处理器链：
data_handler_config:
start_time: 2008-01-01
end_time: 2020-08-01
instruments: csi300
learn_processors:
- class: DropnaLabel
- class: MinMaxScaler
module_path: my_processors # 自定义处理器所在的模块
kwargs:
feature_group: feature
infer_processors:
- class: MinMaxScaler
module_path: my_processors
kwargs:
feature_group: feature
- class: Fillna
kwargs:
feature_group: feature
fill_value: 0
在这个配置中，learn_processors 定义了训练阶段使用的处理器链，包括删除缺失标签和 MinMax 缩放；infer_processors 定义了推理阶段使用的处理器链，包括 MinMax 缩放和填充缺失值。
通过代码方式构建处理器链：
from qlib.data.dataset.processor import ProcessorChain
from qlib.contrib.data.processor import DropnaLabel, Fillna
from my_processors import MinMaxScaler
# 创建处理器链
learn_processors = ProcessorChain([
DropnaLabel(),
MinMaxScaler(feature_group="feature")
])
infer_processors = ProcessorChain([
MinMaxScaler(feature_group="feature"),
Fillna(feature_group="feature", fill_value=0)
])
# 在数据处理器中使用处理器链
handler_config = {
"class": "Alpha158",
"module_path": "qlib.contrib.data.handler",
"kwargs": {
"start_time": "2008-01-01",
"end_time": "2020-08-01",
"instruments": "csi300",
"learn_processors": learn_processors,
"infer_processors": infer_processors,
}
}
处理器链中的处理器会按顺序执行，前一个处理器的输出会作为后一个处理器的输入。这种方式使得数据处理逻辑更加模块化，便于维护和扩展。
高级自定义处理器示例
下面我们介绍一个更复杂的自定义处理器示例——基于市场状态的动态标准化处理器。这个处理器会根据市场状态（如牛市、熊市）分别对特征进行标准化，以适应不同市场环境下的特征分布变化。
import numpy as np
import pandas as pd
from qlib.data.dataset.processor import Processor
class MarketStateNorm(Processor):
def __init__(self, feature_group="feature", index_code="000300.SH", window=20):
self.feature_group = feature_group
self.index_code = index_code
self.window = window
self.bull_params = {}
self.bear_params = {}
def _get_market_state(self, data):
# 假设数据中包含指数收益率
index_ret = data["$close/{}".format(self.index_code)].pct_change()
# 计算市场趋势
market_trend = index_ret.rolling(window=self.window).mean()
# 牛市：趋势为正；熊市：趋势为负
return market_trend > 0
def fit(self, data):
if self.feature_group not in data:
return self
features = data[self.feature_group]
market_state = self._get_market_state(data)
# 分别计算牛市和熊市的均值和标准差
for col in features.columns:
bull_mask = market_state & ~features[col].isna()
bear_mask = ~market_state & ~features[col].isna()
if bull_mask.sum() > 0:
self.bull_params[col] = {
"mean": features.loc[bull_mask, col].mean(),
"std": features.loc[bull_mask, col].std()
}
if bear_mask.sum() > 0:
self.bear_params[col] = {
"mean": features.loc[bear_mask, col].mean(),
"std": features.loc[bear_mask, col].std()
}
return self
def process(self, data):
if not self.bull_params or not self.bear_params:
raise ValueError("Processor has not been fitted. Call fit() first.")
data = data.copy()
if self.feature_group not in data:
return data
features = data[self.feature_group]
market_state = self._get_market_state(data)
for col in features.columns:
if col not in self.bull_params or col not in self.bear_params:
continue
# 对牛市和熊市数据分别标准化
bull_mask = market_state & ~features[col].isna()
bear_mask = ~market_state & ~features[col].isna()
# 牛市标准化
if bull_mask.sum() > 0:
mean = self.bull_params[col]["mean"]
std = self.bull_params[col]["std"] if self.bull_params[col]["std"] != 0 else 1e-8
features.loc[bull_mask, col] = (features.loc[bull_mask, col] - mean) / std
# 熊市标准化
if bear_mask.sum() > 0:
mean = self.bear_params[col]["mean"]
std = self.bear_params[col]["std"] if self.bear_params[col]["std"] != 0 else 1e-8
features.loc[bear_mask, col] = (features.loc[bear_mask, col] - mean) / std
data[self.feature_group] = features
return data
这个高级处理器示例展示了如何根据市场状态动态调整数据处理逻辑。它首先通过指数收益率的滚动均值判断市场状态（牛市或熊市），然后分别计算不同市场状态下的特征均值和标准差，最后根据当前市场状态应用相应的标准化参数。
高频数据处理技巧
随着量化投资的发展，高频数据在策略研发中的应用越来越广泛。与日线数据相比，高频数据具有数据量大、噪声多、包含更多微观结构信息等特点，因此需要特殊的处理方法。QLib 提供了专门的高频数据处理框架，支持分钟级甚至 tick 级数据的处理和分析。
高频数据结构与存储
QLib 采用了高效的二进制格式存储高频数据，与日线数据类似，但针对高频场景进行了优化。高频数据的存储结构如下：
- data/
- calendars/
- 1min.txt # 分钟级交易日历
- instruments/
- all.txt
- csi300.txt
- features/
- sh600000/
- open.1min.bin
- close.1min.bin
- high.1min.bin
- low.1min.bin
- volume.1min.bin
- ...
我们可以使用 QLib 提供的脚本下载和转换高频数据：
# 下载1分钟高频数据
python scripts/get_data.py qlib_data --target_dir ~/.qlib/qlib_data/cn_data_1min --region cn --interval 1min
初始化高频数据环境：
import qlib
from qlib.constant import REG_CN
from qlib.config import HIGH_FREQ_CONFIG
qlib.init(provider_uri="~/.qlib/qlib_data/cn_data_1min", region=REG_CN, **HIGH_FREQ_CONFIG)
高频数据处理特殊考量
处理高频数据时，需要考虑以下几个特殊问题：
- 数据完整性：高频数据可能存在缺失、异常值等问题，需要进行严格的数据清洗。
- 时间对齐：不同股票的交易时间可能不完全一致，需要进行时间对齐处理。
- 计算效率：高频数据量巨大，常规处理方法可能效率低下，需要优化计算逻辑。
- 特征工程：高频数据的特征工程需要考虑市场微观结构，如买卖价差、订单流等。
QLib 提供了专门的高频数据处理器 HighFreqHandler 来解决这些问题：
from qlib.data.dataset.handler import DataHandlerLP
class HighFreqHandler(DataHandlerLP):
def __init__(self, **kwargs):
# 添加高频特有的处理器
kwargs.setdefault("infer_processors", [])
kwargs["infer_processors"].extend([
{
"class": "HighFreqNorm",
"module_path": "highfreq_processor",
},
{
"class": "Fillna",
"kwargs": {
"fields_group": "feature",
"fill_value": 0,
},
},
])
super().__init__(**kwargs)
def get_feature_config(self):
# 定义高频特征
return {
"feature": [
"$high - $low", # 日内波动
"$volume / $amount", # 平均成交价
"Ref($close, 1) / $open - 1", # 开盘收益率
# 更多高频特征...
],
"label": [
"Ref($close, -2) / Ref($close, -1) - 1", # 未来2分钟收益率
],
}
高频数据特征工程
高频数据的特征工程与日线数据有很大不同，需要结合市场微观结构和交易行为进行设计。常见的高频特征包括：
- 价格波动特征：如日内最高价最低价差、价格波动率等。
- 成交量特征：如成交量加权平均价、成交量变化率等。
- 订单流特征：如买卖订单不平衡、订单深度变化等。
- 流动性特征：如买卖价差、市场深度等。
- 微观结构特征：如价格冲击、交易强度等。
QLib 提供了丰富的高频特征计算算子，如 DayLast（取当日最后一个值）、FFillNan（前向填充缺失值）等：
from qlib.data.ops import Operators
from highfreq_ops import DayLast, FFillNan, BFillNan
# 定义高频特征表达式
feature_exprs = [
# 当日累计成交量
"Sum($volume, 240)", # 假设一天有240分钟
# 当日均价
"Sum($close * $volume, 240) / Sum($volume, 240)",
# 最后30分钟价格变化
"DayLast($close) / DayLast(Ref($close, 30)) - 1",
# 前向填充缺失值
"FFillNan($open)",
]
高频策略回测框架
QLib 提供了专门的高频策略回测框架——嵌套决策执行框架（Nested Decision Execution Framework）。该框架支持多级别交易策略的联合回测，如日线级别的选股策略和分钟级别的执行策略。
flowchart TB
A[日线策略] -->|生成目标持仓| B[分钟级执行策略]
B -->|生成下单指令| C[订单执行模拟器]
C -->|返回执行结果| B
B -->|更新实际持仓| A
使用 QLib 进行高频策略回测的配置示例：
qlib_init:
provider_uri: "~/.qlib/qlib_data/cn_data_1min"
region: cn
market: &market 'csi300'
start_time: &start_time "2020-09-15 00:00:00"
end_time: &end_time "2021-01-18 16:00:00"
data_handler_config: &data_handler_config
start_time: *start_time
end_time: *end_time
instruments: *market
freq: '1min'
infer_processors:
- class: 'RobustZScoreNorm'
kwargs:
fields_group: 'feature'
- class: "Fillna"
kwargs:
fields_group: 'feature'
label: ["Ref($close, -2) / Ref($close, -1) - 1"]
task:
model:
class: "HFLGBModel"
module_path: "qlib.contrib.model.highfreq_gdbt_model"
kwargs:
objective: 'binary'
metric: ['binary_logloss','auc']
learning_rate: 0.01
max_depth: 8
dataset:
class: "DatasetH"
module_path: "qlib.data.dataset"
kwargs:
handler:
class: "Alpha158"
module_path: "qlib.contrib.data.handler"
kwargs: *data_handler_config
segments:
train: [*start_time, "2020-11-15 16:00:00"]
test: ["2020-12-01 00:00:00", *end_time]
在高频策略实现中，我们需要特别注意以下几点：
- 交易成本模型：高频交易的成本结构复杂，需要精确建模佣金、滑点等成本。
- 流动性考虑：高频策略往往对流动性较为敏感，需要考虑订单对价格的冲击。
- 市场微观结构：高频策略需要充分利用市场微观结构信息，如订单簿数据。
- 计算效率：高频策略回测需要处理大量数据，需要优化计算逻辑以提高效率。
QLib 的高频交易框架提供了灵活的接口，可以满足这些需求，帮助用户快速实现和验证高频交易策略。