一个低PE的股票池--选股策略

whiteBear

EP因子即市盈率因子，常被投资者使用的几个估值因子之一。一般使用PE，即Price to Earning,维基百科上的解释：市盈率指每股市价除以每股收益盈利（Earning Per Share,EPS）,通常作为股票是便宜抑或昂贵的指标（通货膨胀会使每股收益虚增，从而扭曲市盈率的比较价值）。市盈率把企业的股价与其制造财富的能力联系起来。市盈率越低，代表投资者能够以相对低价格购入股票。

PE因子的计算

市盈率PE的定义：

\[PE = \frac{每股股价}{每股收益(EPS)}=\frac{市值}{净利润} \]

未来函数：

公式系统对数据的运算是基于一系列函数，函数必须满足时序不变形，即时间靠后的数据不对时间靠前的数据产生影响（判断是否是未来函数的以及）。对未来函数也可以理解为：某一量依赖另一量，如量A和量B，B变化使A变化，那么A是B的函数，如果B是稍后的量，A是稍早的量，A跟着B变，A是B的未来函数（来自：百度百科）
因此为了防止未来函数，每股收益以公告日期为准

计算PE

# 计算市盈率：compute_pe
# 市盈率 = 股价/每股收益（EPS）
# 市盈率 = 市值 / 净利润

import os
from stock_util import get_all_codes
from database import DB_CONN
from pymongo import DESCENDING,UpdateOne
from tqdm import tqdm

finance_report_collection = DB_CONN[\'finance_report\']
daliy_collection = DB_CONN[\'daily\']

def compute_pe():
    """
    计算股票在某只的市盈率
    """
    #获取所有的股票列表
    codes = get_all_codes()
    print(codes )
    #计算市盈率
    for code in tqdm(codes):
        print(\'计算市盈率,%s\' %code)
        daily_cursor = daliy_collection.find(
            {\'code\':code},
            projection={\'date\':True,\'close\':True})
        update_requests = []
        for daily in daily_cursor:
            _date = daily[\'date\']

            #判断是否已经存在pe
            pe_exists = daliy_collection.find(
                {\'code\':code,\'date\':_date,\'pe\':{\'$exists\':True}}
                )
            if pe_exists:
                continue
            #找到该股票距离当前日期最近的年报，通过公告日期查询，防止未来函数
            finance_report = finance_report_collection.find_one(
                {\'code\':code,\'report_date\':{\'$regex\':\'\d{4}-12-31\'},\'annouced_date\':{\'$lte\':_date}},
                sort=[(\'annouced_date\',DESCENDING)]
            )
            if finance_report is None:
                continue

            #计算滚动市盈率并保存到daily中
            eps = 0
            if finance_report[\'eps\'] !=\'-\':
                eps = finance_report[\'eps\']
            
            #计算pe
            if eps != 0:
                update_requests.append(UpdateOne(
                    {\'code\':code,\'date\':_date},
                    {\'$set\':{\'pe\':round(daily[\'close\']/eps,4)}}))
        if len(update_requests)>0:
            update_result = daliy_collection.bulk_write(update_requests,ordered=False)    
            print(\'更新PE,%s,更新：%d\' %(code,update_result.modified_count))


if __name__ == "__main__":
    compute_pe()
    # os.system(\'shutdown -s -t 60\')

将每只股票所计算得到的市盈率PE值，更新到daily数据库中，与此同时，通过tqdm库实现股票代码列表进度条可视化展示，判断之前是否计算过pe值，防止重复计算，减少无用功，并运行结果图：


构建低PE股票池

条件

  • 0< PE < 30
  • PE从小到大排序，剔除停牌，取前100只
  • 再平衡周期：7个交易日

流程图如下：

代码实现：

""""
实现股票池，条件0<PE<30,按照PE正序排列，最多取100只股票
再平衡周期为7个交易日
"""
from stock_util import get_trading_date
from database import DB_CONN
from pymongo import ASCENDING,DESCENDING
import pandas as pd
import matplotlib.pyplot as plt

daily = DB_CONN[\'daily\']
daily_hfq = DB_CONN[\'daily_hfq\']

def stock_pool(begin_date,end_date):
    """
    股票池的选股逻辑
    :param begin_date:开始日期
    :param end_date:结束日期
    :return：tuple:所有调整日，以及调整日和代码列表对应的dict
    """
    """
    设置股票池参数
    """
    #调整周期为7个交易日，可以改变参数
    adjust_interval = 7
    #PE的范围
    pe_range = (0,30)
    #PE 的排序方式，ASCENDING升序，DESCENDING降序
    sort = ASCENDING
    #股票池内的股票数量
    pool_size = 100


    #返回值:调整日和当前股票代码列表
    adjust_date_codes_dict = dict()
    #返回值：所有的调整日列表
    all_adjust_dates = []

    #获取所有交易日列表,获取指定时间范围内的所有交易日列表，按照日期正序排列
    all_dates = get_trading_date(begin_date=begin_date,end_date=end_date)
    #上一期的所有股票代码
    last_phase_codes = []
    #在调整日调整股票池
    for _index in range(0,len(all_dates),adjust_interval):
        print(_index)
        #保存调整日
        adjust_date = all_dates[_index]
        all_adjust_dates.append(adjust_date)
        print(\'调整日期：%s\' % adjust_date,flush=True)

        #查询出调整当日，0 < PE < 30，且非停牌的股票
        #最重要的一点是：按照pe正序排列，只取前100只
        daily_cursor = daily.find(
            {\'date\':adjust_date,\'pe\':{\'$lt\':pe_range[1],\'$gt\':pe_range[0]},\'is_trading\':True}, 
            sort=[(\'pe\',sort)],
            projection={\'code\':True},
            limit=pool_size
            )
        #拿到所有的股票代码
        codes = [x[\'code\'] for x in daily_cursor]
        print(adjust_date)
        print(codes)

        #本期股票列表
        this_phase_codes =[]
        #如果上期股票代码列表不为空，则查询出上次股票池中正在停牌的股票
        ##如果不是第一次设立的股票池，可以把股票池全部清掉，重新设立新的股票池，但如果有股票停牌了，需要保存
        ##也就是股票池中为何要将停牌股票存入股票池的初始化呢???大大的问号
      
        if len(last_phase_codes)>0:
            suspension_cursor = daily.find(
                # 查询是股票代码、日期和是否为交易，这里is_trading=False
                {\'code\':{\'$in\':last_phase_codes},\'date\':adjust_date,\'is_trading\':False},
                #只需要股票代码
                projection={\'code\':True}
            )
            #获得股票代码
            suspension_codes = [x[\'code\'] for x in suspension_cursor]

            #保留股票池中正在停盘的股票
            this_phase_codes = suspension_codes
        #打印所有停盘的股票代码
        print(\'上期停盘\',flush=True)
        print(this_phase_codes,flush=True)

        #用新的股票将剩余位置补齐
        this_phase_codes += codes[0:pool_size - len(this_phase_codes)]
       
        #将本次股票设为下次运行时的上次股票
        last_phase_codes = this_phase_codes
        # print(last_phase_codes)

        #建立该调整日和股票列表的对应关系  
        adjust_date_codes_dict[adjust_date]= this_phase_codes
        print(\'最终出票\',flush=True)
        print(this_phase_codes,flush=True)
    #返回结果
    return all_adjust_dates,adjust_date_codes_dict
def find_out_stocks(last_phase_codes,this_phase_codes):
    """
    找到上期入选，本期被调出的股票，这些股票必须卖出
    :param last_phase_codes:上期股票列表
    :param this_phase_codes：本期股票列表
    :return :被调出的股票列表
    """
    out_stocks = []
     
    for code in last_phase_codes:
        if code not in this_phase_codes:
            out_stocks.append(code)
    return out_stocks

def statistic_stock_pool():
    """
    统计股票池的收益
    """
     # 找到指定时间范围的股票池数据，这里的时间范围可以改变
    adjust_dates,codes_dict = stock_pool(\'2019-01-02\',\'2019-12-31\')
    print(codes_dict)
    #用DataFrame保存收益，profit是股票池的收益，hs300是用来对比沪深300的涨跌幅
    df_profit = pd.DataFrame(columns=[\'profit\',\'hs300\'])
    #统计开始的第一天，股票池的收益和沪深300的涨跌幅都为0
    df_profit.loc[adjust_dates[0]] = {\'profit\':0,\'hs300\':0}
    #找到沪深300第一天的值，后面的累计涨跌幅都要和它比较
    hs300_begin_value = daily.find_one({\'code\':\'000300\',\'index\':True,\'date\':adjust_dates[0]})[\'close\']
    """
    通过净值的方式计算累计收益
    累计收益 = 期末净值 - 1
    第N期净值的计算方法：
    net_value(n) = net_value(n-1) + net_value(n-1)*profit(n)
                =net_value(n-1)*(1 + profit(n))
    """
    #设定初始净值为1
    net_value = 1
    #在所有调整日上统计收益，循环时从1开始，因为每次计算要用到当期和上期
    for _index in range(1,len(adjust_dates)-1):

        print(_index)
        #上一期的调整日
        last_adjust_date = adjust_dates[_index-1]
        #当前调整日
        current_adjust_date = adjust_dates[_index]

        #上一期的股票代码
        codes = codes_dict[last_adjust_date]
        print(codes)
        #构建股票代码和后复权买入价格的股票
        buy_daily_cursor = daily_hfq.find(
            {\'code\':{\'$in\':codes},\'date\':last_adjust_date},
            projection={\'close\':True,\'code\':True}
            )
        """
        >>>dict()                        # 创建空字典
        {}
        >>> dict(a=\'a\', b=\'b\', t=\'t\')     # 传入关键字
        {\'a\': \'a\', \'b\': \'b\', \'t\': \'t\'}
        >>> dict(zip([\'one\', \'two\', \'three\'], [1, 2, 3]))   # 映射函数方式来构造字典
        {\'three\': 3, \'two\': 2, \'one\': 1} 
        >>> dict([(\'one\', 1), (\'two\', 2), (\'three\', 3)])    # 可迭代对象方式来构造字典
        {\'three\': 3, \'two\': 2, \'one\': 1}
        >>>
        """
        #使用迭代对象方式来构造字典
        code_buy_close_dict = dict([(buy_daily[\'code\'],buy_daily[\'close\'] )for buy_daily in buy_daily_cursor])

        # 找到上期股票的在当前调整日时的收盘价
        # 1、这里用的是后复权的价格，保持价格的连续性
        # 2、当前的调整日，也就是上期的结束日
        sell_daily_cursor = daily_hfq.find(
            {\'code\':{\'$in\':codes},\'date\':current_adjust_date},
            # 只需要用到收盘价来计算收益
            projection={\'code\':True,\'close\':True}
        )

        #初始化所有股票的收益之和
        profit_sum = 0
        #参与收益统计的股票数量
        count= 0
        # 循环累加所有股票的收益
        for sell_daily in sell_daily_cursor:
            print(sell_daily)
            # 股票代码
            code = sell_daily[\'code\']

            #如果该股票存在股票池开始时的收盘价，则参与收益统计
            if code in code_buy_close_dict:
                #进入股票池的价格
                buy_close = code_buy_close_dict[code]
                #当前价格
                sell_close = sell_daily[\'close\']
                # 累加所有股票的收益
                profit_sum += (sell_close - buy_close) / buy_close
                #参与收益计算的股票数加1
                count+=1
          
        #如果股票数量大于0，才统计当前收益
        if count>0:
            
            #计算平均收益
            profit = round(profit_sum/count,4)
            #当前沪深300的值
            hs300_close = daily.find_one(
                {\'code\':\'000300\',\'index\':True,\'date\':current_adjust_date}
            )[\'close\']
            print(current_adjust_date)
            print(hs300_close)

            #计算净值和累计收益，放到DataFrame中
            net_value = net_value * (1 + profit)
            df_profit.loc[current_adjust_date] = {
                #乘100，改为百分比
                \'profit\':round((net_value -1) * 100,4),
                \'hs300\':round((hs300_close - hs300_begin_value)* 100 / hs300_begin_value,4)
            }
            print(df_profit)

    #绘制曲线
    df_profit.plot(title=\'Stock Pool Evaluation Result\',kind=\'line\')
    plt.show()

#股票池入口函数
if __name__ == "__main__":
    #统计股票池收益
    statistic_stock_pool()

股票收益统计效果展示：

由于之前对数据字段\'is_trading\'等未全部添加，导致数据的不完整。
鉴于上，存在的几个疑问：

  1. 为何将上期股票池中的停牌股票列表作为当期股票池的初始列表？？？

那是因为股票池中的股票已经买入，第二天它停盘了，则需要保留，无法卖出

  1. “再平衡周期：7个交易日”，这个平衡周期作用是什么？？？

一个简单的理解，假如有一批闲置资金,50%投资了股票指数基金,50%投资了低风险固定收益产品,每隔固定一段时间(再平衡周期),对资产进行再平衡,使股票资产和固定资产的比例恢复到50%对50%.
拿本讲股票池为例，除上期股票列表中的停牌股票以外，需要全部更新新的股票列表为本期股票列表，换句话说，卖出除上期股票列表中的停牌股票以外的所有股票，再买入更新过后的股票。当然这只是非常简单的策略。
这样做的优点:

  1. 再平衡可以提高实现资产长期回报率目标的概率。
  2. 再平衡有利于风险控制。
  3. 再平衡可以减少非理性交易。

缺点：

1、频繁操作容易带来很高的交易成本。
2、投资者很容易设定偏颇的资产配置比例。
3、遭遇单边行情。
4、许多投资者并不会真正理性地执行再平衡。
5、没有人真正知道该以什么样的频次做再平衡。
参考：再平衡的优缺点,再平衡周期对投资组合长期收益率影响大吗？——资产配置系列3

总结：
本股票池策略的逻辑：
根据PE指标按下列条件筛选出股票，并放入股票池中：

1、 0< PE < 30，并对筛选出来的股票按PE正序排序
2、 再平衡周期，更新股票池
此外，在统计股票池收益时，通过净值的方式计算累计收益，最后与沪深300收益进行比较

分类：

技术点：

相关文章：

猜你喜欢
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode