導包

#引入技術指標數據
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:returnself.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)# 獲取當天valuevalue = 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.commissionelif 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 = filenamefor i in stk_pools[:]:try:datapath = 'D:/stock_data/'+idf = pd.read_csv('D:/stock_data/'+i)#將數據長度不足的股票刪去if len(df)<55:passelse: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())