Python实现股票量化交易学习进阶第二篇之简单交易策略的定义实现

1、backtrader回测框架知识2、需求一自定义MACD指标3、需求二自定义实现KDJ指标4、需求三自定义CCI指标

1、backtrader回测框架知识

由于backtrader框架的中文文档比较少，这里首先推荐看两个文档，一个是中文文档：交易员之家—量化宗师之路 ,一个是英文文档，对于我来说收货最大的应该是英文文档，虽然看起来很费劲，但是结合代码例子去看，还是很快熟悉Indicator和Strategy的用法。

这里总结一下backtrader框架中常用的Cerebro、Strategy和Indicator三个类的作用：

Cerebro是回测引擎，在Cerebro中配置策略、添加经纪人账户初始金额、交易手续费、配置历史交易数据以及交易的分析器。Strategy交易策略类，在Strategy中使用交易指标(Indicator)，并设置交易执行buy和sell的条件，Strategy中可以读取到Cerebro中配置的交易数据，并可以自动传递给Indicator中。Indicator指标类，自定义继承后可以实现诸多的交易公式，并将计算结果写入lines中，方便Strategy中进行下标同步访问。当然backtrader中也提供了很多指标，具体可以参考英文文档中的Indicators条目。

以上三点是我学习后总结的，在没有自己实际去写代码测试后是很难立刻理解的，下面我用三个需求例子来具体讲解一下。

2、需求一自定义MACD指标

这里我想通过自己编写MACD的Indicator，因为上一篇内容中我写的macd计算方法去按照炒股软件公式进行了修正，talib中自带的macd公式与股票软件中计算结果存在一点点误差，所以这里我要自己定义一个MACD的Indicator。定义的Indicator会在自定义的Strategy类中进行指标判断，当MACD的柱状线由负转正然后连续两天大于0时买入，在柱状线值连续降低两天后卖出。

这里采用紫金矿业股票测试两个及两个周期看看投资效果。

首先给出自己定义的macd的Indicator类：

from __future__ import (absolute_import, division, print_function,unicode_literals)import backtrader as btimport tools as toimport pandas as pdimport array'''自定义的MACD计算指标类，继承backtrader的indicator'''class MACD(bt.Indicator):# 定义macd lines，实例中以及实例外可以使用self.lines.macd进行访问，也可以使用self.l.macd进行访问或self.macd# 注意以下self.lines.macd是linebuffer类型，在init中使用self.lines.macd，而在next中就需要使用# self.lines.macd[index]的形式# macd为DIFF线，macdsignal为DEA线,macdhist为MACD柱状线lines = ('macd', 'macdsignal', 'macdhist')# macd周期参数定义，可以使用self.params.fastperiod的形式访问，同样也可以按照self.p.fastperiod的形式访问# params = (('fastperiod', 12),('slowperiod', 26),('signalperiod', 9) )def __init__(self, **kwargs):super(MACD, self).__init__()# 将策略传入的周期参数进行赋值# 组装变量fastperiod = Noneslowperiod = Nonesignalperiod = Nonekeys = kwargs.keys()for k in keys:if k == 'fastperiod':fastperiod = kwargs[k]if k == 'slowperiod':slowperiod = kwargs[k]if k == 'signalperiod':signalperiod = kwargs[k]# Indicator中读取Cerebro中填充的数据形式与Strategy一致，# self.data与self.datas[0]等价，默认都指向收盘价close。如果# 要访问传入的具体列也可以使用self.data.close来指出列名# 将数据转成series类型计算macdcloseSeriesData = pd.Series(self.data.array)macd, macdsignal, macdhist =\self.calc_macd(close=closeSeriesData, fastperiod=fastperiod or 12, slowperiod=slowperiod or 26, signalperiod=signalperiod or 9)# 将计算结果赋值给lines的array,在Strategy中可以通过访问lines进行访问下面数据# macd为DIFF线，macdsignal为DEA线,macdhist为MACD柱状线self.lines.macd.array = array.array(str('d'),list(macd.values))self.lines.macdsignal.array = array.array(str('d'),list(macdsignal.values))self.lines.macdhist.array = array.array(str('d'),list(macdhist.values))def calc_macd(self,close, fastperiod=12, slowperiod=26, signalperiod=9):'''通过传入的收盘价close的series类型计算MACD，返回的macd是dif，macdsignal是dea，macdhist是macdfastperiod=12, slowperiod=26, signalperiod=9这三个参数分别对应股票软件上的三个日期参数close是收盘价列'''if not isinstance(close, pd.Series):raise Exception("传入的close参数不是Series类型!")ewma12 = close.ewm(span=fastperiod,adjust=False).mean() # 收盘价针对fastperiod参数求EMAewma26 = close.ewm(span=slowperiod,adjust=False).mean() # 收盘价针对slowperiod参数求EMAmacd = ewma12 - ewma26macdsignal = macd.ewm(span=signalperiod,adjust=False).mean() # macd针对signalperiod参数求EMAmacdhist = (macd - macdsignal) * 2return (macd, macdsignal, macdhist)

上面代码中定义了一个MACD的类，这个类要继承bt.Indicator，注意Indicator是首字母大写的类名，我刚开始书写时没注意写成小写的了，报错后整整调试了一下午才发现这个错误。

代码中在lines元组汇总定义，MACD指标会返回的line

lines = ('macd', 'macdsignal', 'macdhist')

MACD类中要返回三条line分别是’macd’、 ‘macdsignal’和 ‘macdhist’，当计算完数据后，计算结果将分别存储在’macd’、 'macdsignal’和 'macdhist’三条线的array中，让Strategy类中可以通过lines.macd的形式访问三条线中保存的计算结果。这里要强调一句，每条线中只能存储一列数据，所以macd的计算结果是三列，所以需要三条线分别进行存储。

类中我将计算过程放置在__init__方法中了，这样可以借助pandas的优势进行矩阵运算提高效率，如果由于特殊原因需要一行一行的去计算可以放在next方法中进行计算，但是这里需要强调两点，第一：next中计算时元素计算不是集合运算，所以需要读取下标(当然不写下标默认返回的还是当前行的值)；第二：因为next每执行一次是按照最小周期进行读取的，默认情况下最小周期设置时1，关于最小周期的设置我也没有搞懂，有想搞懂的朋友可以看英文文档或者看这个译文，所以这里强烈建议在__init__方法计算。

其他代码这里就不详细的去讲了看注释就可以了。

下面定义交易策略类：

from __future__ import (absolute_import, division, print_function,unicode_literals)import backtrader as btimport backtrader.indicators as biimport pandas as pdimport indmacd as idsimport indkdj as indkdjimport indcci as indcciimport tools as toimport backtrader.analyzers as btanalyzersfrom datetime import datetimeimport talib as ta# 策略逻辑：macdclass Calf(bt.Strategy):params = (('fastperiod', 12),("slowperiod", 26),("signalperiod", 9),("printlog", False),)def __init__(self):self.order = Noneself.buyprice = Noneself.buycomm = None# 计算macd，计算结果分别会存储在self.macd的lines中# self.macd.lines.macd是DIFF线，等效访问方式self.macd.l.macd或self.macd.macd# self.macd.lines.macdsignal为DEA线，等效访问方式self.macd.l.macdsignal或self.macd.macdsignal# self.macd.lines.macdhist为MACD柱状线，等效访问方式self.macd.l.macdhist或self.macd.macdhistself.macd = ids.MACD(fastperiod=self.p.fastperiod,slowperiod=self.p.slowperiod,signalperiod=self.p.signalperiod)# 制定交易策略的函数，策略模块最核心的部分def next(self):if self.order: # 检查是否有指令等待执行,return# 检查是否持仓if not self.position: # 没有持仓# MACD柱状线今天大于0 and 昨天大于0 and 前天小于0时买入if self.macd.macdhist[0] > 0 and self.macd.macdhist[-1] > 0 and self.macd.macdhist[-1]<0:# 执行买入self.order = self.buy()else:# MACD柱状线处于0轴上方，且连续两天下降卖出if self.macd.macdhist[0] > 0 and self.macd.macdhist[-1] > 0 and self.macd.macdhist[-1] > 0 and self.macd.macdhist[0] < self.macd.macdhist[-1] and self.macd.macdhist[-1] < self.macd.macdhist[-2]:# 执行卖出self.order = self.sell()# 输出交易记录def log(self, txt, dt=None, doprint=False):if self.params.printlog or doprint:dt = dt or self.datas[0].datetime.date(0)print('%s, %s' % (dt.isoformat(), txt))def notify_order(self, order):# 有交易提交/被接受，啥也不做if order.status in [order.Submitted, order.Accepted]:return# 交易完成，报告结果if order.status in [pleted]:if order.isbuy():self.log('执行买入, 价格: %.2f, 成本: %.2f, 手续费 %.2f' %(order.executed.price, order.executed.value,m))else:self.log('执行卖出, 价格: %.2f, 成本: %.2f, 手续费 %.2f' %(order.executed.price, order.executed.value,m))elif order.status in [order.Canceled, order.Margin, order.Rejected]:self.log("交易失败")self.order = None#记录交易收益情况（可省略，默认不输出结果）def notify_trade(self, trade):if not trade.isclosed:returnself.log(f'策略收益：\n毛收益 {trade.pnl:.2f}, 净收益 {trade.pnlcomm:.2f},现有持股数量{self.position.size}')# 测略结束时，多用于参数调优def stop(self):self.log('(均线周期 %2d)期末资金 %.2f' %(self.params.maperiod, self.broker.getvalue()))self.log('(均线周期 %2d)期末仓位 %.2f' %(self.params.maperiod, self.position.size))

代码看注释吧。

下面进行回测代码编写，并进行回测:

if __name__ == "__main__":# 读取紫金矿业交易数据df = to.readStockDataFromCsv(code='601899.XSHG')if not df.empty:try:cerebro = bt.Cerebro()# 设置账户初始资金cash = 100000.0cerebro.broker.setcash(cash)# 设置手续费cerebro.broker.setcommission(commission=0.005)# 导入策略参数寻优,开启打印日志,使用optstrategy模式，cerebro.broker.getvalue()获取的值一直是初始值# 如果要计算策略执行前后的投资收益请使用addstrategy模式，这样cerebro.broker.getvalue()可以有效取值#cerebro.optstrategy(Calf, printlog=True)# 导入策略,开启打印日志cerebro.addstrategy(Calf, printlog=True)# 指定具体回测时间段start_date = datetime(, 1, 1) # 回测开始时间end_date = datetime(, 12, 31) # 回测结束时间data = bt.feeds.PandasData(dataname=df,fromdate=start_date,todate=end_date)cerebro.adddata(data)# 设定需要设定每次交易买入的股数(3000股)cerebro.addsizer(bt.sizers.FixedSize, stake=3000)# 添加分析器# 优化运行模式下，返回值是列表的列表,内列表只含一个元素，即策略实例# 分析夏普比率cerebro.addanalyzer(btanalyzers.SharpeRatio, _name='sharp')# 最大回撤cerebro.addanalyzer(btanalyzers.DrawDown, _name='draw')thestrats = cerebro.run(maxcpus=2)# 优化运行模式(optstrategy)下，返回值是列表的列表,内列表只含一个元素，即策略实例# cerebro.addstrategy模式返回值时列表，用thestrats[0]取值#thestrat = thestrats[0][0]thestrat = thestrats[0]sharp = thestrat.analyzers.sharp.get_analysis()draw = thestrat.analyzers.draw.get_analysis()profit = (cerebro.broker.getvalue() / cash) - 1print("收益率:", profit)print("夏普率:", '%.2f' % (sharp['sharperatio'] or 0))print("最大回撤率:", '%.2f' % draw['drawdown'])except Exception as e:print(e)

看回测结果数据:

-02-13, 执行买入, 价格: 2.95, 成本: 8850.00, 手续费 44.25-02-28, 执行卖出, 价格: 3.45, 成本: 8850.00, 手续费 51.75-02-28, 策略收益：毛收益 1500.00, 净收益 1404.00,现有持股数量0-04-08, 执行买入, 价格: 3.59, 成本: 10770.00, 手续费 53.85-04-11, 执行卖出, 价格: 3.50, 成本: 10770.00, 手续费 52.50-04-11, 策略收益：毛收益 -270.00, 净收益 -376.35,现有持股数量0-05-23, 执行买入, 价格: 3.07, 成本: 9210.00, 手续费 46.05-06-11, 执行卖出, 价格: 3.09, 成本: 9210.00, 手续费 46.35-06-11, 策略收益：毛收益 60.00, 净收益 -32.40,现有持股数量0-08-07, 执行买入, 价格: 3.68, 成本: 11040.00, 手续费 55.20-08-13, 执行卖出, 价格: 3.69, 成本: 11040.00, 手续费 55.35-08-13, 策略收益：毛收益 30.00, 净收益 -80.55,现有持股数量0-10-29, 执行买入, 价格: 3.27, 成本: 9810.00, 手续费 49.05-11-12, 执行卖出, 价格: 3.43, 成本: 9810.00, 手续费 51.45-11-12, 策略收益：毛收益 480.00, 净收益 379.50,现有持股数量0-11-25, 执行买入, 价格: 3.60, 成本: 10800.00, 手续费 54.00-11-29, 执行卖出, 价格: 3.47, 成本: 10800.00, 手续费 52.05-11-29, 策略收益：毛收益 -390.00, 净收益 -496.05,现有持股数量0-12-31, 期末资金 100798.15-12-31, 期末仓位 0.00收益率: 0.007981499999999864夏普率: 0.00最大回撤率: 1.00

看结果这个策略在度基本失败。

在回测代码中需要有基点注意事项：

cerebro.optstrategy和cerebro.addstrategy的使用区别

使用cerebro.optstrategy策略参数寻优方法导入策略后，cerebro.broker.getvalue()方法调用结果只能读取到broker.setcash设置的初始金额，无法读取到策略执行后的期末金额，因为策略参数寻优方法允许将参数作为可迭代对象进行设置，所以策略执行完无法在cerebro.broker.getvalue()中判断出来需要读取哪一个参数值的期末资金。cerebro.optstrategy和cerebro.addstrategy对分析器的影响

使用cerebro.optstrategy时，当添加了分析器后(如下面代码添加了夏普率及最大回撤分析器)

# 添加夏普率cerebro.addanalyzer(btanalyzers.SharpeRatio, _name='sharp')# 最大回撤cerebro.addanalyzer(btanalyzers.DrawDown, _name='draw')thestrats = cerebro.run(maxcpus=2)

cerebro.run(maxcpus=2)的执行结果thestrats返回的是一个列表，列表中只有一个为列表的值，要读取分析器的结果需要使用thestrats[0][0]的形式，参见下面代码：

thestrat = thestrats[0][0]sharp = thestrat.analyzers.sharp.get_analysis()draw = thestrat.analyzers.draw.get_analysis()profit = (cerebro.broker.getvalue() / cash) - 1print("收益率:", profit)print("夏普率:", '%.2f' % (sharp['sharperatio'] or 0))print("最大回撤率:", '%.2f' % draw['drawdown'])

如果要是使用迭代器参数，则多个参数值的运行结果需要使用thestrat = thestrats[0][0]、thestrat = thestrats[1][0]…

那么当使用cerebro.addstrategy时，添加分析器后由于只有一个运行结果所以，cerebro.run(maxcpus=2)的执行结果thestrats返回的是一个列表，列表中的值就是分析器的实例，如：

thestrat = thestrats[0]sharp = thestrat.analyzers.sharp.get_analysis()draw = thestrat.analyzers.draw.get_analysis()profit = (cerebro.broker.getvalue() / cash) - 1print("收益率:", profit)print("夏普率:", '%.2f' % (sharp['sharperatio'] or 0))print("最大回撤率:", '%.2f' % draw['drawdown'])

更多分析器类型详见下图，使用时用btanalyzers.SharpeRatio形式引用

# 替换btanalyzers.SharpeRatio为btanalyzers.Returns计算收益率cerebro.addanalyzer(btanalyzers.SharpeRatio, _name='sharp')

入门量化回测最强神器backtrader（三）

3、需求二自定义实现KDJ指标

from __future__ import (absolute_import, division, print_function,unicode_literals)import backtrader as btimport pandas as pdimport array'''自定义的KDJ计算指标类，继承backtrader的Indicator(注意：不是indicator)'''class KDJ(bt.Indicator):# 定义KDJ lines，类中可以使用self.lines.K进行访问，也可以使用self.l.K进行访问# 注意以下self.lines.K是linebuffer类型，在init中使用self.lines.K，而在next中就需要使用# self.lines.K[index]的形式lines = ('K', 'D', 'J')# kdj周期参数定义，可以使用self.params.fastk_period的形式访问，同样也可以按照self.p.fastk_period的形式访问# params = (('fastk_period', 9),('slowk_period', 3),('slowd_period', 3) )def __init__(self, **kwargs):super(KDJ, self).__init__()# 将策略传入的周期参数进行赋值# self.params.fastperiod=fastperiod# self.params.slowperiod=slowperiod# self.params.signalperiod=signalperiod# 组装变量fastk_period = Noneslowk_period = Noneslowd_period = Nonekeys = kwargs.keys()for k in keys:if k == 'fastk_period':fastk_period = kwargs[k]if k == 'slowk_period':slowk_period = kwargs[k]if k == 'slowd_period':slowd_period = kwargs[k]# 将数据转成DataFrame类型，且包含最低价、最高价及收盘价df = pd.DataFrame(pd.Series(self.data.close.array), columns=['close'])df['high'] = pd.Series(self.data.high.array)df['low'] = pd.Series(self.data.low.array)# 计算kdjk, d, j = self.calc_kdj(df,fastk_period=fastk_period or 9,slowk_period=slowk_period or 3,slowd_period=slowd_period or 3)# 将序列赋值给lines的array,在Strategy中可以通过访问lines进行访问下面数据self.lines.K.array = array.array(str('d'), list(k.values))self.lines.D.array = array.array(str('d'), list(d.values))self.lines.J.array = array.array(str('d'), list(j.values))def calc_kdj(self,df, fastk_period=9, slowk_period=3, slowd_period=3, fillna=False):'''根据传入的最高价、最低价、收盘价计算KDJ指标参数:df:pandas的DataFrame类型，需要包含最低价、最高价及收盘价fastk_period:RSV中日期间隔 int 类型。默认为9日slowk_period:K线指标日期间隔 int类型。默认为3天slowd_period:D线指标日期间隔 int类型。默认为3天fillna:bool类型，默认为False。为True时，在计算RSV的最高价(或最低价)的最大值(或最小值)过程中如果存在Nan数据将被传入的df中的最大值或最小值填充'''# 检查传入的参数是否是pandas的DataFrame类型if not isinstance(df, pd.DataFrame):raise Exception("传入的参数不是pandas的DataFrame类型!")# 检查传入的df是否存在high、low、close三列，不存在报错if ('high' not in df.columns) or ('low' not in df.columns) or ('close' not in df.columns):# 抛出异常raise Exception("传入的参数不存在最高价、最低价、收盘价中的一个或几个!")# 计算指定日期间隔内的最低价的最小值low_list = df['low'].rolling(fastk_period, min_periods=fastk_period).min()# 将NAN填充成现有数据中的最小值if fillna is True:low_list.fillna(value=df['low'].expanding().min(), inplace=True)# 计算指定日期间隔的最高阶的最大值high_list = df['high'].rolling(9, min_periods=9).max()# 将NAN填充成现有数据中的最大值if fillna is True:high_list.fillna(value=df['high'].expanding().max(), inplace=True)# 计算RSV （国泰君安中的RSV公式RSV:=(CLOSE-LLV(LOW,N))/(HHV(HIGH,N)-LLV(LOW,N))*100;）# RSV赋值:(收盘价-N日内最低价的最低值)/(N日内最高价的最高值-N日内最低价的最低值)*100rsv = (df['close'] - low_list) / (high_list - low_list) * 100k = pd.DataFrame(rsv).ewm(com=slowk_period - 1,adjust=False).mean() # rsv针对slowk_period参数求移动权重平均数d = k.ewm(com=slowd_period - 1, adjust=False).mean() # k针对slowd_period参数求移动权重平均数当,adjust为False时，以递归方式计算加权平均值j = 3 * k - 2 * dreturn (k, d, j)

交易策略自己可以按照自己的规则设定，然后进行回测。

4、需求三自定义CCI指标

这里需要说明一下，由于talib中CCI指标计算结果与炒股软件中一致，所以我的自定义类中使用了talib库计算CCI，当然也可以使用backtrader提供的CCI指标或者backtrader.talib中的CCI指标进行计算，我给出自己写出的类：

from __future__ import (absolute_import, division, print_function,unicode_literals)import backtrader as btimport pandas as pdimport arrayimport talib as ta'''自定义的CCI计算指标类，继承backtrader的Indicator(注意：不是indicator)'''class CCI(bt.Indicator):# 定义CCI lines，类中可以使用I进行访问，也可以使用I进行访问# 注意以下I是linebuffer类型，在init中使用I，而在next中就需要使用# I[index]的形式lines = ('CCI',)# kdj周期参数定义，可以使用self.params.timeperiod的形式访问，同样也可以按照self.p.timeperiod的形式访问# params = (('timeperiod', 14))def __init__(self, **kwargs):super(CCI, self).__init__()# 将策略传入的周期参数进行赋值# self.params.fastperiod=fastperiod# self.params.slowperiod=slowperiod# self.params.signalperiod=signalperiod# 组装变量timeperiod = Nonekeys = kwargs.keys()for k in keys:if k == 'timeperiod':timeperiod = kwargs[k]# 将数据转成Series类型，且包含最低价、最高价及收盘价# 计算CCIcci = I(pd.Series(self.data.high.array),pd.Series(self.data.low.array),pd.Series(self.data.close.array),timeperiod=timeperiod or 14)# 将序列赋值给lines的array,在Strategy中可以通过访问lines进行访问下面数据I.array = array.array(str('d'), list(cci.values))

如果觉得《Python实现股票量化交易学习进阶(二)之简单交易策略的定义实现》对你有帮助，请点赞、收藏，并留下你的观点哦！