【backtrader股票策略】《151 trading strategies》中的收益动量(earnings momentum)

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

在上一讲分享了《151 trading strategies》中，我们测试了价格动量策略，在本文中，我们尝试分享一下收益动量策略。

从优矿上获取了股票的每个季度的每股财务指标，我们使用每股营业利润作为earings的代表，但是由于这个财务指标公布日期不知道，只知道统计截止日期每季度结束的日期，所以，在使用这些数据的时候，有可能存在利用未来数据的可能性，有可能会造成收益率虚高。

和原先的策略一样，本文也主要分为四个部分：策略逻辑描述、策略代码、策略绩效、策略简单分析

策略逻辑说明

相对于原先的价格动量策略，收益动量策略的唯一区别就是不在使用过去的收益率进行排序了，而是使用SUE指标进行排序。SUE通过计算现在的收益与四个季度前的收益的差值除以过去八个季度的收益的标准差。做多SUE高的一部分股票，做空SUE低的一部分股票。

1. 和前几个策略的资金、资金分配、交易手续费都是一样的。
2. 我们使用全市场的A股日数据进行测试，做多头，也做空头。多头和空头都占用资金。
3. 假设初始资金有1个亿，手续费为万分之二。
4. 在实际的测试中，如果可以得到财务报表的发布日期，使用发布日期，避免使用未来数据。由于暂时没有得到数据的发布日期，在本次测试中，暂时忽略这个。

策略代码

import backtrader as bt
import datetime
import pandas as pd
import numpy as np
import os,sys
import copy
import talib
import math 
import warnings
warnings.filterwarnings("ignore")
import pyfolio as pf

# 我们使用的时候，直接用我们新的类读取数据就可以了。
class test_two_ma_strategy(bt.Strategy):
   
    params = (('period',30),
              ('hold_percent',0.02)
             )

    def log(self, txt, dt=None):
        ''' Logging function fot this strategy'''
        dt = dt or self.datas[0].datetime.date(0)
        print('{}, {}'.format(dt.isoformat(), txt))

    def __init__(self):
        # Keep a reference to the "close" line in the data[0] dataseries
        self.bar_num=0    
        # 保存现有持仓的股票
        self.position_dict={}
        # 当前有交易的股票
        self.stock_dict={}
        # 加载每股收益的数据
        self.stock_info = pd.read_csv('/home/yun/data/每股财务指标数据/股票每股收益数据.csv')
        self.stock_info['endDate'] = pd.to_datetime(self.stock_info['endDate'])
       
    def prenext(self):
        
        self.next()
        
        
    def next(self):
        # 假设有100万资金，每次成份股调整，每个股票使用1万元
        self.bar_num+=1
        # 前一交易日和当前的交易日
        pre_date = self.datas[0].datetime.date(-1).strftime("%Y-%m-%d")
        current_date = self.datas[0].datetime.date(0).strftime("%Y-%m-%d")
        # 总的价值
        total_value = self.broker.get_value()
        total_cash  = self.broker.get_cash()
        self.log(f"total_value : {total_value}")
        # 第一个数据是指数，校正时间使用，不能用于交易
        # 循环所有的股票,计算股票的数目
        for data in self.datas[1:]:
            data_date = data.datetime.date(0).strftime("%Y-%m-%d")
            # 如果两个日期相等，说明股票在交易
            if current_date == data_date:
                stock_name = data._name
                if stock_name not in self.stock_dict:
                    self.stock_dict[stock_name]=1
        total_target_stock_num = len(self.stock_dict)
        # 现在持仓的股票数目
        total_holding_stock_num = len(self.position_dict)
        # 计算理论上的手数
        now_value = total_value/int(total_target_stock_num*self.p.hold_percent*2)
        # 如果今天是调仓日
        if self.bar_num%self.p.period == 0:
            
            # 循环股票，平掉所有的股票，计算现在可以交易的股票的累计收益率
            result = []
            for data in self.datas[1:]:
                    data_date = data.datetime.date(0).strftime("%Y-%m-%d")
                    size = self.getposition(data).size
                    # 如果有仓位
                    if size!=0:
                        self.close(data)
                        if data._name in self.position_dict:
                            self.position_dict.pop(data._name)

                    # 已经下单，但是订单没有成交
                    if data._name in self.position_dict and size==0:
                        order = self.position_dict[data._name]
                        self.cancel(order)
                        self.position_dict.pop(data._name) 
                    # 如果两个日期相等，说明股票在交易,就计算收益率，进行排序
                    if current_date == data_date:
                        new_stock_info = self.stock_info[self.stock_info['secID']==data._name]
                        new_stock_info = new_stock_info[new_stock_info['endDate']<=pd.to_datetime(current_date)]
                        new_stock_info = new_stock_info.sort_values('endDate')
                        # print(current_date,data._name,new_stock_info)
                        if len(new_stock_info)>=8:
                            opps = list(new_stock_info['opPS'])
                            sue = (opps[-1]-opps[-4])/np.std(opps[-8:])
                            # self.log(f"{data._name},{sue}")
                            result.append([data,sue])
            # 根据计算出来的累计收益率进行排序，选出前10%的股票做多，后10%的股票做空
            new_result = sorted(result,key=lambda x:x[1])
            num = int(self.p.hold_percent * total_target_stock_num)
            sell_list = new_result[:num]
            buy_list = new_result[-num:]
            # 根据计算出来的信号，买卖相应的股票
            for data,cumsum_rate in buy_list:
                lots = now_value/data.close[0]
                lots = int(lots/100)*100 # 计算能下的手数，取整数
                order = self.buy(data,size = lots)
                self.position_dict[data._name] = order
            for data,cumsum_rate in sell_list:
                lots = now_value/data.close[0]
                lots = int(lots/100)*100 # 计算能下的手数，取整数
                order = self.sell(data,size = lots)
                self.position_dict[data._name] = order
                
                
                        
        
    def notify_order(self, order):
        
        if order.status in [order.Submitted, order.Accepted]:
            return
        
        if order.status == order.Rejected:
            self.log(f"Rejected : order_ref:{order.ref}  data_name:{order.p.data._name}")
            
        if order.status == order.Margin:
            self.log(f"Margin : order_ref:{order.ref}  data_name:{order.p.data._name}")
            
        if order.status == order.Cancelled:
            self.log(f"Concelled : order_ref:{order.ref}  data_name:{order.p.data._name}")
            
        if order.status == order.Partial:
            self.log(f"Partial : order_ref:{order.ref}  data_name:{order.p.data._name}")
         
        if order.status == order.Completed:
            if order.isbuy():
                self.log(f" BUY : data_name:{order.p.data._name} price : {order.executed.price} , cost : {order.executed.value} , commission : {order.executed.comm}")

            else:  # Sell
                self.log(f" SELL : data_name:{order.p.data._name} price : {order.executed.price} , cost : {order.executed.value} , commission : {order.executed.comm}")
    
    def notify_trade(self, trade):
        # 一个trade结束的时候输出信息
        if trade.isclosed:
            self.log('closed symbol is : {} , total_profit : {} , net_profit : {}' .format(
                            trade.getdataname(),trade.pnl, trade.pnlcomm))
            # self.trade_list.append([self.datas[0].datetime.date(0),trade.getdataname(),trade.pnl,trade.pnlcomm])
            
        if trade.isopen:
            self.log('open symbol is : {} , price : {} ' .format(
                            trade.getdataname(),trade.price))
    def stop(self):
        
        pass 
                
        
# 初始化cerebro,获得一个实例
cerebro = bt.Cerebro()
# cerebro.broker = bt.brokers.BackBroker(shortcash=True)  # 0.5%
data_root = "/home/yun/data/stock/day/"
file_list =sorted(os.listdir(data_root))
params=dict(
    
    fromdate = datetime.datetime(2009,1,4),
    todate = datetime.datetime(2020,7,31),
    timefr ame = bt.Timefr ame.Days,
    dtformat = ("%Y-%m-%d"),
    # compression = 1,
    datetime = 0,
    open = 1,
    high = 2,
    low =3,
    close =4,
    volume =5,
    openinterest=-1)

# 加载指数数据
df = pd.read_csv("/home/yun/data/stock/index.csv")
df.columns = ['datetime','open','high','low','close','volume','openinterest']
df.index = pd.to_datetime(df['datetime'])
df = df[['open','high','low','close','volume','openinterest']]
df = df[(df.index<=params['todate'])&(df.index>=params['fromdate'])]
# feed = bt.feeds.GenericCSVData(dataname = "/home/yun/data/stock/index.csv",**params)
feed = bt.feeds.PandasDirectData(dataname = df)
# 添加数据到cerebro
cerebro.adddata(feed, name = 'index')

# 读取数据
for file in file_list:
    df = pd.read_csv(data_root+file)
    df.columns = ['datetime','open','high','low','close','volume','openinterest']
    df.index = pd.to_datetime(df['datetime'])
    df = df[['open','high','low','close','volume','openinterest']]
    df = df[(df.index<=params['todate'])&(df.index>=params['fromdate'])]
    if len(df)==0:
        continue 
    # feed = bt.feeds.GenericCSVData(dataname = data_root+file,**params)
    feed = bt.feeds.PandasDirectData(dataname = df)
    # 添加数据到cerebro
    cerebro.adddata(feed, name = file[:-4])
print("加载数据完毕")
# 添加手续费，按照万分之二收取
cerebro.broker.setcommission(commission=0.0002,stocklike=True)
# 设置初始资金为100万
cerebro.broker.setcash(1_0000_0000)
# 添加策略
cerebro.addstrategy(test_two_ma_strategy)
cerebro.addanalyzer(bt.analyzers.TotalValue, _name='_TotalValue')
cerebro.addanalyzer(bt.analyzers.PyFolio)
# 运行回测
results = cerebro.run()
# 打印相关信息
pyfoliozer = results[0].analyzers.getbyname('pyfolio')
returns, positions, transactions, gross_lev = pyfoliozer.get_pf_items()
pf.create_full_tear_sheet(
    returns,
    positions=positions,
    transactions=transactions,
    # gross_lev=gross_lev,
    live_start_date='2019-01-01',
    )

策略绩效

策略评价

作为一个多空中性策略(long-short equity),这个收益动量因子能够达到每年收益率13%,夏普率接近1.7的水平，总体上挺不错的。但是考虑到回测使用的每股营业利润的数据没有公布日期，我们回测的时候可能存在使用未来数据的现象，造成策略的绩效虚高。实际使用当中，尤其是利用财务数据的时候，一定要注意，在当前，能够直接拿到的有哪些数据，不能使用未来数据。

本文部分转载自付费专栏：https://yunjinqi.blog.csdn.net/article/details/113195111

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏7 回帖

关键词：Strategies earnings Momentum earning Strateg

感谢楼主，哈哈，正在看这个书（或者是假装在看这个书）