其他分享
首页 > 其他分享> > 多股回测(backtrader+quantstats+akshare)

多股回测(backtrader+quantstats+akshare)

作者:互联网

导包

#引入技术指标数据
from __future__ import (absolute_import ,division,print_function,unicode_literals)
import datetime #用于datetime对象操作
import os.path  #用于管理路径
import sys      #用于在argvTo[0]中找到脚本名称
import backtrader as bt #引入backtrader框架
%matplotlib inline

策略

#创建策略
class TestStrategy(bt.Strategy):
    params = (
        ('maperiod1',5),
        ('maperiod2',13),
        ('maperiod3',21),
        ('maperiod4',34),
        ('maperiod5',55),
        ('printlog',True),
        ('poneplot' , False),#是否打印到同一张图
        ('pstake' , 100000) #单笔交易股票数据
    )
    def log(self,txt,dt=None,doprint = False):
        dt = dt or self.datas[0].datetime.date(0)
        #print('%s,%s' % (dt.isoformat(),txt))

        """策略的日志函数"""
        if self.params.printlog or doprint:
            dt = dt or self.datas[0].datetime.date(0)
            print('%s,%s' % (dt.isoformat(),txt))
    def __init__(self):
        self.inds = dict()
        for i, d in enumerate(self.datas):
            self.inds[d] = dict()
            self.inds[d]['ma1'] = bt.indicators.SimpleMovingAverage( d.close,period = self.params.maperiod1)
            self.inds[d]['ma2'] = bt.indicators.SimpleMovingAverage( d.close,period = self.params.maperiod2)
            self.inds[d]['ma3'] = bt.indicators.SimpleMovingAverage( d.close,period = self.params.maperiod3)
            self.inds[d]['ma4'] = bt.indicators.SimpleMovingAverage( d.close,period = self.params.maperiod4)
            self.inds[d]['ma5'] = bt.indicators.SimpleMovingAverage( d.close,period = self.params.maperiod5)
            self.inds[d]['D1'] = bt.ind.CrossOver(self.inds[d]['ma5'],self.inds[d]['ma4']) #交叉信号
            self.inds[d]['A1'] = bt.ind.CrossOver(self.inds[d]['ma1'],self.inds[d]['ma2']) #交叉信号   
            self.inds[d]['C1'] = bt.ind.CrossOver(self.inds[d]['ma2'],self.inds[d]['ma3'])
            #跳过第一只股票data,第一只股票data作为主图数据
            if i > 0:
                if self.p.poneplot:
                    d.plotinfo.plotmaster = self.datas[0]
    def notify_trade(self,trade):
        if not trade.isclosed:
            return
        self.log('OPERATION PROFIT,GROSS %.2F,NET %.2F' %
                (trade.pnl,trade.pnlcomm))
    #多股回测时使用,数据读取。 
    def prenext(self):
        self.next()
    def next(self):
        # 获取当天日期
        date = self.datas[0].datetime.date(0)
        # 获取当天value
        value = self.broker.getvalue()
        for i , d in enumerate(self.datas):            
            dt,dn = self.datetime.date(),d._name             #获取时间及股票代码        
            pos = self.getposition(d).size 
            sig1 = ((self.inds[d]['D1'][-1]>0) and (self.inds[d]['A1'][0]>0)) and (self.inds[d]['ma2'][0] >self.inds[d]['ma4'][0])and (self.inds[d]['ma4'][0] >self.inds[d]['ma4'][-1])
            sig2 = ((self.inds[d]['D1'][-1]>0)  or (self.inds[d]['A1'][0]>0 ))and(self.inds[d]['ma2'][0] >self.inds[d]['ma2'][-1])and(d.close[0]/d.open[0]>1.05)and(d.volume[0] /d.volume[-1]>2)
            sig3 = ((self.inds[d]['D1'][-1]>0)  or (self.inds[d]['A1'][0]>0 ))and(self.inds[d]['ma2'][0] >self.inds[d]['ma3'][0] )and(self.inds[d]['ma3'][0] >self.inds[d]['ma4'][0] )and(self.inds[d]['ma4'][0] >self.inds[d]['ma4'][-1] )
            sig4 = self.inds[d]['C1'][0]<0
            #print('sig1',sig1)
            if not pos:                                      # 不在场内,则可以买入  vol成交量, ref日前
                if sig1 or sig2 and sig3: #如果金叉
                    self.buy(data =d,size =self.p.pstake)    #买
                    self.log('%s,BUY CREATE, %.2f ,%s' % (dt, d.close[0] ,d._name))
                    #self.order = self.buy()
            elif sig4:              #在场内。且死叉
                self.close(data = d)                     #卖
                self.log('%s,SELL CREATE,%.2f,%s' % (dt, d.close[0] ,d._name))
                #self.order = self.sell()

印花税

class stampDutyCommissionScheme(bt.CommInfoBase):
    params = (
        ('stamp_duty',0.005),#印花税率
        ('percabs',True),
    )
    def _gotcommission(self,size,price,pseudoexec):
        if size >0:#买入,不考虑印花税
            return size*price * self.p.commission
        elif size<0:#卖出,考虑印花税
            return -size*price*(self.p.stamp_duty + self.p.commission)
        else:
            return 0

开始回测

#创建cerebro实体
cerebro = bt.Cerebro()
#添加策略
cerebro.addstrategy(TestStrategy)

添加数据

#创建价格数据
import akshare as ak
import baostock as bs
import pandas as pd
import datetime

#获取股票池数据
from os import listdir
filename = listdir('D:/stock_data')
stk_pools = filename

for i in stk_pools[:]:
  
    try:
        datapath = 'D:/stock_data/'+i
        df = pd.read_csv('D:/stock_data/'+i)
        #将数据长度不足的股票删去
        if len(df)<55:
            pass
        else:
            try:
                data = bt.feeds.GenericCSVData(
                    dataname = datapath,
                    fromdate = datetime.datetime(2010,4,1),
                    todate = datetime.datetime(2021,7,8),
                    nullvalue = 0.0,
                    dtformat = ('%Y-%m-%d'),
                    datetime = 1,
                    open =2,
                    high = 3,
                    low = 4,
                    close = 5,
                    volume = 6,
                    openinterest = -1
                    )
                cerebro.adddata(data,name = i)
            except:
                continue 
    except:
        continue

 设置参数

#设置启动资金
cerebro.broker.setcash(len(stk_pools[:50])*10000)
#设置交易单位大小
cerebro.addsizer(bt.sizers.FixedSize,stake = 100)
#设置佣金为千分之一
comminfo = stampDutyCommissionScheme(stamp_duty=0.005,commission=0.001)
cerebro.broker.addcommissioninfo(comminfo)
#不显示曲线
for d in cerebro.datas:
    d.plotinfo.plot = False
#打印开始信息
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())

回测数据分析

#查看策略效果
cerebro.addanalyzer(bt.analyzers.PyFolio, _name='pyfolio')
back  = cerebro.run(maxcpus=12,exactbars=True,stdstats=False)


import warnings
warnings.filterwarnings('ignore')
strat = back[0]
portfolio_stats = strat.analyzers.getbyname('pyfolio')
returns, positions, transactions, gross_lev = portfolio_stats.get_pf_items()
returns.index = returns.index.tz_convert(None)



import quantstats
quantstats.reports.html(returns, output='stats.html', title='Stock Sentiment')



import webbrowser
f = webbrowser.open('stats.html')
#打印最后结果
print('Final Profolio Value : %.2f' %cerebro.broker.getvalue())

标签:bt,inds,self,cerebro,quantstats,回测,akshare,import,dt
来源: https://blog.csdn.net/qq_26742269/article/details/119216791