你的位置:首页 > 信息动态 > 新闻中心
信息动态
联系我们

股票因子扩展3(候选因子计算)——从零到实盘6

2021-12-1 4:22:34

在这里插入图片描述

前文记录了双神因子的实现,本文记录候选因子的实现。当候选因子值为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已创建,感兴趣的读者请扫码加微信!