前文记录了双神因子的实现，本文记录候选因子的实现。当候选因子值为True时，股票就会进入候选股票池，后续监测是否买入。
主要代码分析
新建源文件，命名为data_center_v5.py，全部内容见文末，v5主要涉及5个方面的改动：
新增计算均线因子函数
def ma(df, n=5, factor='close'):
该函数用于计算均线因子，其中：
- 参数df为待计算扩展因子的DataFrame
- 参数n为待计算均线的周期，默认计算5日均线
- 参数factor为待计算均线的因子，默认为收盘价
- 返回值为包含扩展因子的DataFrame
name = '{}ma_{}'.format('' if 'close' == factor else factor + '_', n)
设置均线因子名称，例如，收盘价的5日均线名称为ma_5，成交量的5日均线名称为volume_ma_5。
s = pd.Series(df[factor], name=name, index=df.index)
取待计算均线的因子列。
s = s.rolling(center=False, window=n).mean()
利用rolling和mean计算均线数据。
df = df.join(s)
将均线数据添加到原始的DataFrame中。
df[name] = df[name].apply(lambda x: round(x + 0.001, 2))
均线数值保留两位小数。
return df
返回包含均线因子的DataFrame。
新增计算多条均线因子函数
def mas(df, ma_list=None, factor='close'):
该函数用于计算多条均线因子，内部调用ma计算单条均线，其中：
- 参数df为待计算扩展因子的DataFrame
- 参数ma_list: 待计算均线的周期列表，默认为None
- 参数factor为 待计算均线的因子，默认为收盘价
- 返回值为包含扩展因子的DataFrame
if ma_list is None:
ma_list = []
for i in ma_list:
df = ma(df, i, factor)
return df
循环调用函数ma，计算得到多条均线因子。
新增计算穿均线因子函数
def cross_mas(df, ma_list=None):
该函数用于计算穿均线因子，其中：
- 参数df为待计算扩展因子的DataFrame
- 参数ma_list为均线的周期列表，默认为None
- 返回值为包含扩展因子的DataFrame
若当日最低价不高于均线价格，且当日收盘价不低于均线价格，则当日穿均线因子值为True，否则为False。
if ma_list is None:
ma_list = []
for i in ma_list:
df['cross_{}'.format(i)] = (df['low'] <= df['ma_{}'.format(i)]) & (
df['ma_{}'.format(i)] <= df['close'])
return df
循环计算穿多条均线因子。
新增计算候选因子函数
def candidate(df):
该函数用于计算候选因子，其中：
- 参数df为待计算扩展因子的DataFrame
- 返回值为包含扩展因子的DataFrame
若同时满足以下3个条件，则股票当日作为候选，该因子值为True，否则为False
- 当日日线同时穿过5、10、20、30日均线
- 30日均线在60日均线上方
- 当日形成双神
ma_list = [5, 10, 20, 30, 60]
均线周期列表。
temp_df = mas(df, ma_list)
计算均线的因子，保存到临时的DataFrame中。
temp_df = cross_mas(temp_df, ma_list)
计算穿多线的因子，保存到临时的DataFrame中。
column_list = ['cross_{}'.format(x) for x in ma_list[:-1]]
穿多线因子的列名列表。
df['candidate'] = temp_df[column_list].all(axis=1) & (temp_df['ma_30'] >= temp_df['ma_60']) & df['ss']
计算候选因子，若为True则为候选。
return df
返回包含候选因子的DataFrame。以603358华达科技为例，在2021年11月24日的打印结果如下：
date open high low ... isST zt ss candidate
0 2017-01-25 12.280496 14.735282 12.280496 ... 0 True False False
1 2017-01-26 16.208810 16.208810 16.208810 ... 0 True False False
2 2017-02-03 17.830019 17.830019 17.830019 ... 0 True False False
3 2017-02-06 19.612037 19.612037 17.902219 ... 0 True False False
4 2017-02-07 19.474201 20.501405 18.968804 ... 0 False False False
... ... ... ... ... ... ... ... ... ...
1169 2021-11-18 21.300000 21.390000 20.680000 ... 0 False False False
1170 2021-11-19 20.660000 20.660000 19.060000 ... 0 False False False
1171 2021-11-22 19.940000 20.420000 19.920000 ... 0 False False False
1172 2021-11-23 19.800000 20.200000 19.650000 ... 0 False False False
1173 2021-11-24 19.870000 21.660000 19.390000 ... 0 True True True
[1174 rows x 20 columns]
可见在2021年11月24日，该股票的candidate因子为True，结合下面的K线图进行验证：
2021年11月24日，日线同时穿过5、10、20、30日均线， 30日均线在60日均线上方，且当日形成双神，符合策略的候选条件。
修改计算扩展因子函数
def extend_factor(df):
df = df.pipe(zt).pipe(ss, delta_days=30).pipe(candidate)
return df
使用pipe依次计算涨停、双神及是否为候选股票。
小结
至此，我们完成了主要的策略因子计算，后续可以利用这些因子进行策略胜率计算了。
到目前为止，创建的数据只是用于打印，未实现存储。因此只要确保程序能正常运行即可，不需要等待程序运行结束。在进行策略胜率计算和回测前，我们先在后续文章中记录多线程计算，以及数据保存到MySQL的过程。
data_center_v5.py的全部代码如下：
import baostock as bs
import datetime
import sys
import numpy as np
import pandas as pd
# 可用日线数量约束
g_available_days_limit = 250
# BaoStock日线数据字段
g_baostock_data_fields = 'date,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ, psTTM,pcfNcfTTM,isST'
def get_stock_codes(date=None):
"""
获取指定日期的A股代码列表
若参数date为空，则返回最近1个交易日的A股代码列表
若参数date不为空，且为交易日，则返回date当日的A股代码列表
若参数date不为空，但不为交易日，则打印提示非交易日信息，程序退出
:param date: 日期
:return: A股代码的列表
"""
# 登录baostock
bs.login()
# 从BaoStock查询股票数据
stock_df = bs.query_all_stock(date).get_data()
# 如果获取数据长度为0，表示日期date非交易日
if 0 == len(stock_df):
# 如果设置了参数date，则打印信息提示date为非交易日
if date is not None:
print('当前选择日期为非交易日或尚无交易数据，请设置date为历史某交易日日期')
sys.exit(0)
# 未设置参数date，则向历史查找最近的交易日，当获取股票数据长度非0时，即找到最近交易日
delta = 1
while 0 == len(stock_df):
stock_df = bs.query_all_stock(datetime.date.today() - datetime.timedelta(days=delta)).get_data()
delta += 1
# 注销登录
bs.logout()
# 筛选股票数据，上证和深证股票代码在sh.600000与sz.39900之间
stock_df = stock_df[(stock_df['code'] >= 'sh.600000') & (stock_df['code'] < 'sz.399000')]
# 返回股票列表
return stock_df['code'].tolist()
def create_data(stock_codes, from_date='1990-12-19', to_date=datetime.date.today().strftime('%Y-%m-%d'),
adjustflag='2'):
"""
下载指定日期内，指定股票的日线数据，计算扩展因子
:param stock_codes: 待下载数据的股票代码
:param from_date: 日线开始日期
:param to_date: 日线结束日期
:param adjustflag: 复权选项 1：后复权 2：前复权 3：不复权 默认为前复权
:return: None
"""
# 下载股票循环
for code in stock_codes:
print('正在下载{}...'.format(code))
# 登录BaoStock
bs.login()
# 下载日线数据
out_df = bs.query_history_k_data_plus(code, g_baostock_data_fields, start_date=from_date, end_date=to_date,
frequency='d', adjustflag=adjustflag).get_data()
# 注销登录
bs.logout()
# 剔除停盘数据
if out_df.shape[0]:
out_df = out_df[(out_df['volume'] != '0') & (out_df['volume'] != '')]
# 如果数据为空，则不创建
if not out_df.shape[0]:
continue
# 删除重复数据
out_df.drop_duplicates(['date'], inplace=True)
# 日线数据少于g_available_days_limit，则不创建
if out_df.shape[0] < g_available_days_limit:
continue
# 将数值数据转为float型，便于后续处理
convert_list = ['open', 'high', 'low', 'close', 'preclose', 'volume', 'amount', 'turn', 'pctChg']
out_df[convert_list] = out_df[convert_list].astype(float)
# 重置索引
out_df.reset_index(drop=True, inplace=True)
# 计算扩展因子
out_df = extend_factor(out_df)
print(out_df)
def extend_factor(df):
"""
计算扩展因子
:param df: 待计算扩展因子的DataFrame
:return: 包含扩展因子的DataFrame
"""
# 使用pipe依次计算涨停、双神及是否为候选股票
df = df.pipe(zt).pipe(ss, delta_days=30).pipe(candidate)
return df
def zt(df):
"""
计算涨停因子
若涨停，则因子为True，否则为False
以当日收盘价较前一日收盘价上涨9.8%及以上作为涨停判断标准
:param df: 待计算扩展因子的DataFrame
:return: 包含扩展因子的DataFrame
"""
df['zt'] = np.where((df['close'].values >= 1.098 * df['preclose'].values), True, False)
return df
def shift_i(df, factor_list, i, fill_value=0, suffix='a'):
"""
计算移动因子，用于获取前i日或者后i日的因子
:param df: 待计算扩展因子的DataFrame
:param factor_list: 待移动的因子列表
:param i: 移动的步数
:param fill_value: 用于填充NA的值，默认为0
:param suffix: 值为a(ago)时表示移动获得历史数据，用于计算指标；值为l(later)时表示获得未来数据，用于计算收益
:return: 包含扩展因子的DataFrame
"""
# 选取需要shift的列构成新的DataFrame，进行shift操作
shift_df = df[factor_list].shift(i, fill_value=fill_value)
# 对新的DataFrame列进行重命名
shift_df.rename(columns={x: '{}_{}{}'.format(x, i, suffix) for x in factor_list}, inplace=True)
# 将重命名后的DataFrame合并到原始DataFrame中
df = pd.concat([df, shift_df], axis=1)
return df
def shift_till_n(df, factor_list, n, fill_value=0, suffix='a'):
"""
计算范围移动因子
用于获取前/后n日内的相关因子，内部调用了shift_i
:param df: 待计算扩展因子的DataFrame
:param factor_list: 待移动的因子列表
:param n: 移动的步数范围
:param fill_value: 用于填充NA的值，默认为0
:param suffix: 值为a(ago)时表示移动获得历史数据，用于计算指标；值为l(later)时表示获得未来数据，用于计算收益
:return: 包含扩展因子的DataFrame
"""
for i in range(n):
df = shift_i(df, factor_list, i + 1, fill_value, suffix)
return df
def ss(df, delta_days=30):
"""
计算双神因子，即间隔的两个涨停
若当日形成双神，则因子为True，否则为False
:param df: 待计算扩展因子的DataFrame
:param delta_days: 两根涨停间隔的时间不能超过该值，否则不判定为双神，默认值为30
:return: 包含扩展因子的DataFrame
"""
# 移动涨停因子，求取近delta_days天内的涨停情况，保存在一个临时DataFrame中
temp_df = shift_till_n(df, ['zt'], delta_days, fill_value=False)
# 生成列表，用于后续检索第2天前至第delta_days天前是否有涨停出现
col_list = ['zt_{}a'.format(x) for x in range(2, delta_days + 1)]
# 计算双神，需同时满足3个条件：
# 1、第2天前至第delta_days天前，至少有1个涨停
# 2、1天前不是涨停（否则就是连续涨停，不是间隔的涨停）
# 3、当天是涨停
df['ss'] = temp_df[col_list].any(axis=1) & ~temp_df['zt_1a'] & temp_df['zt']
return df
def ma(df, n=5, factor='close'):
"""
计算均线因子
:param df: 待计算扩展因子的DataFrame
:param n: 待计算均线的周期，默认计算5日均线
:param factor: 待计算均线的因子，默认为收盘价
:return: 包含扩展因子的DataFrame
"""
# 均线名称，例如，收盘价的5日均线名称为ma_5，成交量的5日均线名称为volume_ma_5
name = '{}ma_{}'.format('' if 'close' == factor else factor + '_', n)
# 取待计算均线的因子列
s = pd.Series(df[factor], name=name, index=df.index)
# 利用rolling和mean计算均线数据
s = s.rolling(center=False, window=n).mean()
# 将均线数据添加到原始的DataFrame中
df = df.join(s)
# 均线数值保留两位小数
df[name] = df[name].apply(lambda x: round(x + 0.001, 2))
return df
def mas(df, ma_list=None, factor='close'):
"""
计算多条均线因子，内部调用ma计算单条均线
:param df: 待计算扩展因子的DataFrame
:param ma_list: 待计算均线的周期列表，默认为None
:param factor: 待计算均线的因子，默认为收盘价
:return: 包含扩展因子的DataFrame
"""
if ma_list is None:
ma_list = []
for i in ma_list:
df = ma(df, i, factor)
return df
def cross_mas(df, ma_list=None):
"""
计算穿均线因子
若当日最低价不高于均线价格
且当日收盘价不低于均线价格
则当日穿均线因子值为True，否则为False
:param df: 待计算扩展因子的DataFrame
:param ma_list: 均线的周期列表，默认为None
:return: 包含扩展因子的DataFrame
"""
if ma_list is None:
ma_list = []
for i in ma_list:
df['cross_{}'.format(i)] = (df['low'] <= df['ma_{}'.format(i)]) & (
df['ma_{}'.format(i)] <= df['close'])
return df
def candidate(df):
"""
计算是否为候选
若当日日线同时穿过5、10、20、30日均线
且30日均线在60日均线上方
且当日形成双神
则当日作为候选，该因子值为True，否则为False
:param df: 待计算扩展因子的DataFrame
:return: 包含扩展因子的DataFrame
"""
# 均线周期列表
ma_list = [5, 10, 20, 30, 60]
# 计算均线的因子，保存到临时的DataFrame中
temp_df = mas(df, ma_list)
# 计算穿多线的因子，保存到临时的DataFrame中
temp_df = cross_mas(temp_df, ma_list)
# 穿多线因子的列名列表
column_list = ['cross_{}'.format(x) for x in ma_list[:-1]]
# 计算是否为候选
df['candidate'] = temp_df[column_list].all(axis=1) & (temp_df['ma_30'] >= temp_df['ma_60']) & df['ss']
return df
if __name__ == '__main__':
stock_codes = get_stock_codes()
create_data(stock_codes)
博客内容只用于交流学习，不构成投资建议，盈亏自负！
个人博客：https://coderx.com.cn/（优先更新）
欢迎大家转发、留言。已建微信群用于学习交流，群1已满，群2已创建，感兴趣的读者请扫码加微信！
