AlphaNet
Stock factor mining with CNN and GRU.
Install / Use
/learn @jeremy-feng/AlphaNetREADME
思路框架
问题背景
传统的因子挖掘过程通常是由人工构造因子表达式，对多个单因子进行加权合成。面对大量的原始数据，人工基于投资经验，手动构造因子表达式生成单因子的过程是极其繁琐的。在因子合成阶段，通常是用ICIR加权平均等手段进行合成，这种简单的线性加权方式也限制了因子合成的多种可能性。
将卷积思想应用于因子挖掘
在卷积神经网络中，最关键的特征提取组件是卷积核。在图像识别领域，卷积核通过一个带有可优化的权重和偏置项的矩阵，对原始数据进行互相关操作。
我们可以将原始量价数据整理成一个二维矩阵，尝试使用卷积核对数据进行特征提取。
但是，如果完全采用传统的卷积操作，提取的特征就是：一定感受野范围内的特征的加权组合。这样的操作会有两个问题：
- 提取的特征只是某些特征数据的固定的加权组合，这极大地限制了因子表达式的可能性。
- 传统的卷积核只能感受局部范围内的数据，因此，我们输入的特征变量的上下顺序会影响提取出的特征。因此输入变量的顺序还需要人工干预。
由此看来，简单地套用卷积操作并不合适，但我们可以借鉴卷积核的“遍历操作”的思想，自定义运算符函数，实现类似“卷积层”的特征提取层。
具体的特征提取层将在后文介绍。经过特征提取后，可再添加批标准化层、池化层、全连接层，将原始数据转换为收益率的预测。
优化模型
对于上述将卷积思想应用于因子挖掘的方法，可以尝试对两个方向进行优化。
- 调整网络结构。添加更丰富的特征提取层，将池化层转换为可以记忆时序信息的循环神经网络。
- 调整标签值。将收益率值得预测转换为涨跌方向的预测和超额收益率方向的预测。
准备数据集
我们需要的特征均为量价数据，即open, high, low, close, vwap, volume, return1, turn, free_turn这9个量价指标在$t-29$到$t$时间段的$9\times 30$个特征。
Tushare提供了免费的量价数据接口，在程序中导入token，即可使用pro.daily()下载数据。
下面具体介绍获取数据的细节。
训练集和测试集包含的时间段
由于通过Tushare的免费接口获取数据的速度较慢（逐股票、逐日获取后再合并，而不是批量一次性获取，因此耗时较久），本文只截取了20220101至20220630这半年的数据作为训练集，20220930至20221231这一季度的数据作为测试集。没有用20220630至20221231的数据作为测试集，是因为希望训练集和验证集之间能够暂停一段时间，否则训练集的标签可能会包含未来信息，进而夸大测试集上的预测效果。
本项目下载数据的时间为2022年12月初，因此实际所用的验证集并不是完整的一季度。
采样的日期
如果对训练集和验证集包含的时间段中的每一个交易日均进行采样，会造成两个问题：
- 采样过于频繁，导致相邻日期的数据基本相近。
- 采样天数过多，下载数据的时间会非常久。
因此，本文使用间隔采样的方法，每间隔10个交易日进行一次采样。具体判断哪一天为采样日的函数为：
# 给定日期区间的端点，输出期间的定长采样交易日列表
def get_datelist(start: str, end: str, interval: int):
df = pro.index_daily(ts_code='399300.SZ', start_date=start, end_date=end)
date_list = list(df.iloc[::-1]['trade_date'])
sample_list = []
for i in range(len(date_list)):
if i % interval == 0:
sample_list.append(date_list[i])
return sample_list
其原理是基于沪深300指数（399300.SZ）的交易数据进行间隔采样。沪深300指数有数据的日期一定是交易日。
采样的股票
A股市场的股票数量近5000只，若对每一只股票均进行采样也将耗费大量时间。本文对每个采样日，获取前1000只股票的数据。具体判断对哪些股票进行采样的函数为：
# 给定一个交易日，返回该日满足条件的A股股票列表
def get_stocklist(date: str, num: int):
start = str(pd.to_datetime(date)-timedelta(30))
start = start[0:4]+start[5:7]+start[8:10]
df1 = pro.index_weight(index_code='000002.SH',
start_date=start, end_date=date) # 交易日当天的股票列表
codes = list(df1['con_code'])
codes = codes[0:1000] # 在每个截面期只选取1000只股票
return codes
其原理是基于A股指数（000002.SH）的前1000只成分股进行采样。
获取单个股票在单个交易日的数据
get_x_y()函数返回两个值，一个是前30个交易日的9个指标面板（9*30），一个是未来10天的收益率。
def get_x_y(code: str, date: str, pass_day: int, future_day: int, len1: int, len2: int):
start = str(pd.to_datetime(date)-timedelta(pass_day*2))
start = start[0:4]+start[5:7]+start[8:10]
end = str(pd.to_datetime(date)+timedelta(future_day*2))
end = end[0:4]+end[5:7]+end[8:10]
df_price = pro.daily(ts_code=code, # OHLC,pct_change,volume
start_date=start, end_date=date)
df_basic = pro.daily_basic(ts_code=code,
start_date=start, end_date=date)
df_return = pro.daily(ts_code=code,
start_date=date, end_date=end).iloc[::-1]['close']
if (df_price.shape[0] == df_basic.shape[0]) & (df_price.shape[0] == len1) & (df_return.shape[0] == len2): # 判断数据的完整性
df_price = df_price.iloc[0:pass_day, [2, 3, 4, 5, 8, 9]].fillna(0.1)
df_basic = df_basic.iloc[0:pass_day, [3, 4, 5]].fillna(0.1)
data = np.array(pd.merge(df_price, df_basic,
left_index=True, right_index=True).iloc[::-1].T)
# print(data.shape)
# 未来十个交易日的收益率
dfr = df_return.iloc[0:future_day]
ret = dfr.iloc[-1]/dfr.iloc[0]-1 # 后十个交易日的收益率
return data, ret
else:
return None, None # 数据缺失的预处理
舍弃缺失值
在获取单个股票在单个交易日的数据时，若某只股票的数据有缺失，则需舍弃它，否则在输入到神经网络时会带有缺失值。
基于沪深300指数，判断某日应有的数据长度的函数：
def get_length(date: str, pass_day: int, future_day: int):
start = str(pd.to_datetime(date)-timedelta(pass_day*2))
start = start[0:4]+start[5:7]+start[8:10]
end = str(pd.to_datetime(date)+timedelta(future_day*2))
end = end[0:4]+end[5:7]+end[8:10]
len_1 = pro.index_daily(ts_code='399300.SZ',
start_date=start, end_date=date).shape[0]
len_2 = pro.index_daily(ts_code='399300.SZ',
start_date=date, end_date=end).shape[0]
return len_1, len_2
在get_x_y()函数中，基于len_1和len_2判断了数据的完整性。若有缺失值则返回空值，不会计入数据集中。
获取数据集
筛选出哪一天、哪一只股票需要进行采样后，我们就可以获取数据了。
对每一个采样日、每一只股票进行循环。配合rich.progress可以展示下载数据的进度条。
rich.progress的使用示例可以参考这里。
def get_dataset(num: int, start: str, end: str, interval: int, pass_day: int, future_day: int):
X_train = []
y_train = []
trade_date_list = get_datelist(start, end, interval)
# 添加进度条
with Progress() as progress:
task_date = progress.add_task(
"[red]Date...", total=len(trade_date_list))
for date in trade_date_list:
# 更新进度条
progress.update(task_date, advance=1)
stock_list = get_stocklist(date, num)
len1, len2 = get_length(date, pass_day, future_day)
task_stock = progress.add_task(
"[green]Stock...", total=len(range(len(stock_list))))
for i in range(len(stock_list)):
# 更新进度条
progress.update(task_stock, advance=1)
code = stock_list[i]
x, y = get_x_y(code, date, pass_day, future_day, len1, len2)
try:
if (x.shape[0] == 9) & (x.shape[1] == pass_day):
X_train.append(x)
y_train.append(y)
except Exception:
continue
return X_train, y_train
数据示例：
保存.npy数据到本地
为了方便训练模型，可以将数据以.npy格式存储到本地。在训练模型时可以直接使用np.load('../data/X_train.npy')载入数据。
# 参数设定：使用过去30天的数据预测未来10天的收益率，回归问题
X_train, y_train = get_dataset(
num=1000, start='20220101', end='20220630', interval=10, pass_day=30, future_day=10)
X_test, y_test = get_dataset(num=1000, start='20220931',
end='20221231', interval=10, pass_day=30, future_day=10)
print("there are in total", len(X_train), "training samples")
print("there are in total", len(X_test), "testing samples")
# 将数据保存到本地供离线训练
Xa = np.array(X_train)
ya = np.array(y_train)
Xe = np.array(X_test)
ye = np.array(y_test)
np.save('./X_train.npy', Xa)
np.save('./y_train.npy', ya)
np.save('./X_test.npy', Xe)
np.save('./y_test.npy', ye)
整个获取数据的时间约为3个小时，共获取到11825条训练数据和4943条测试数据（数据量不为1000的整数倍，是因为舍弃了部分缺失值）。
-
特征数据为$9\times 30$的个股量价数据构成的矩阵。9行代表9个量价特征，30代表$t-29$至$t$这30天的数据。
-
标签数据为个股在某个交易日往后10个交易日的收益率。
搭建AlphaNet-V1
AlphaNet-V1的整体网络结构
下图展示了AlphaNet-V1的整体网络结构。它由7个平行的特征提取层、3个平行的池化层和1个全连接层组成。其中，特征提取层和池化层后都有一个批标准化层（Batch Normalization）。
输入数据是一个$9\times30$的个股量价“数据图片”，预测目标为个股从当日到10个交易日后的收益率数值。
特征提取层（类似卷积层）
AlphaNet的输入数据是一个$9\times30$的个股量价“数据图片”。如果简单地套用卷积神经网络处理图片像素数据的操作，则卷积操作只能在感受野内将若干日期的若干量价数据进行加权平均，经过卷积层得到的特征将变得很难解释，也不符合传统构造量价因子的方式。
因此，借鉴卷积神经网络CNN的思想，我们可以将多种运算符函数作为自定义网络层进行特征提取。本文实现了7种运算符，分别是ts_corr, ts_cov, ts_stddev, ts_zscore, ts_return, ts_decaylinear, ts_mean，它们的含义如下：
| 名称 | 定义 |
| ---------------- | ------------------------------------------------------------ |
| ts_corr | 过去 d 天 X 值构成的时序数列和 Y 值构成的时序数列的相关系数。 |
| ts_cov | 过去 d 天 X 值构成的时序数列和 Y 值构成的时序数列的协方差。 |
| ts_stddev | 过去 d 天 X 值构成的时序数列的标准差。 |
| ts_zscore | 过去 d 天 X 值构成的时序数列的平均值除以标准差。 |
| ts_return | (X - delay(X, d))/delay(X, d)-1, delay(X, d)为 X 在 d 天前的取值。 |
| ts_decaylinear | 过去 d 天 X 值构成的时序数列的加权平均值，权数为 d, d – 1, …, 1(权数之和应为 1，需进行归一化处理)，其中离现在越近的日子权数越大。 |
| ts_mean | 过去 d 天 X 值构成的时序数列的平均值。 |
这7个运算符函数中，ts_corr和ts_cov需要从9行数据中提取2行数据，并计算相关系数和协方差。其他5个运算符函数仅需针对某一行数据计算标准差、变化率等。下面针对这两种情况分别举例说明。
基于双变量的特征提取层——以ts_corr为例
我们的输入数据是$9\times30$的矩阵，每一行是某个量价指标在最近30个交易日的值。基于双变量进行特征提取的步骤为：
- 取出两行数据。
- 对于取出的两行数据，给定步长
stride，在时间维度上对两行数据进行遍历，计算两行数据的相关系数。例如，当$stride=3$时，下一次计算将在时间维度上往右步进3步，我们将进行$\frac{30}{3}=10$次运算。 - 将运算结果整理到新的矩阵，得到新的“特征图片”，作为后续池化层的输入。
从9行数据中任取2行，有$\tbinom{9}{2}=36$种取法。假设我们设定步长为10，则得到的新的“特征图片”的维数是$36\times3$。
基于双变量的特征提取层——代码实现
需要给定原始矩阵Matrix、两两组合的列表combination、反转的两两组合的列表combination_rev以及每次遍历运算的起始索引列表index_list。
- 生成
combination和combination_rev的代码为：
# 生成卷积操作时需要的两列数据的组合的列表
def generate_combination(N):
"""
args:
N: int, the number of rows of the matrix
return:
combination: list, the combination of two columns of the matrix
combination_rev: list, the combination of two rows of the matrix, which is the reverse of combination
"""
col = []
col_rev = []
for i in range(1,N):
for j in range(0,i):
col.append([i,j])
col_rev.append([j,i])
return col, col_rev
# 生成卷积操作时需要的两列数据的组合的列表
combination, combination_rev = generate_combination(9)
- 生成
index_list的代码为：
# 根据输入的矩阵和卷积操作的步长, 计算卷积操作的索引
def get_index_list(matrix, stride):
"""
args:
matrix: torch.tensor, the input matrix
stride: int, the str
