39、【backtrader股票策略】在A股中使用基于PB指标的价值投资策略可以赚钱吗

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

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

立即领取

感谢您参与论坛问题回答

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

+2 论坛币

和前几个策略一样，策略思路依然来自于《151 trading strategies》中，本文将会分析价值策略在A股中这些年的表现情况。

注：当你做策略的时候，你就会发现，用backtrader写策略真简单。这个策略写好，花了我7分钟的时间，运行花了一个多小时。

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

策略逻辑说明

相对于原先的收益动量策略，价值投资策略的思路唯一改变的地方就是选择股票的标准不一样，通过做多低P/B的一组股票，做空高P/B的一组股票，以期待获得相对稳定的收益率。

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

策略代码

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['nAssetPS'])
                            if opps[-1]!=0:
                                sue = data.close[0]/opps[-1]
                                # 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)
            # 做多低pb的股票，做空高pb的股票
            buy_list = new_result[:num]
            sell_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',
    )

策略回测结果

策略简单分析

从策略的回测结果上来看，总体上是盈利的，但是表现没有收益动量因子好，这可能是因为PB是大家常用的用于衡量是否被高估低估的因子，造成这个因子的失效一些。并不能得出结论说，价值投资策略不如收益动量策略，这可能跟我们选择的因子有关。

感兴趣的话，可以尝试一些其他代表价值的因子，网上有很多类似的。

本文转载自付费文章：https://yunjinqi.blog.csdn.net/article/details/113272017 文章中有相应的数据

---

智慧、心灵、财富，总要有一个在路上，愿我们能在人生的道路上，不断成长、不断成熟～～～

感兴趣可以关注我的专栏：

my_quant_study_note：分享一些关于量化投资、量化交易相关的思考

backtrader量化投资回测与交易：本专栏免费，分享backtrader相关的内容。

量化投资神器-backtrader源码解析-从入门到精通：本专栏目前收费99元，预计更新100篇策略+20篇backtrader讲解+80篇源代码分析。

扫码加我 拉你入群

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

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

分享0 收藏2 回帖

关键词：trader Trade 投资策略 可以赚钱 价值投资

Thanks for sharing.

个人建议：
1. 大A的投资标的，不建议使用基本面的变量，作为策略的输入参量。
2. 另外，大A的数据量太小，作为回测的数据源，不足以规避一些隐含的过度拟合。建议增加海外更大的数据作为回测数据源。