【原】backtrader:基于信号的策略开发及参数优化

AI量化实验室 2023-10-12 发布于北京

展开全文

原创文章第73篇，专注“个人成长与财富自由、世界运作的逻辑， AI量化投资”。

今天继续讲backtrader的交易。

bt在易用性上确实是下足了功夫，我们先来看一下“极简”的策略开发。

01 “基于信号的策略”。

它不需要写strategy。

直接定义信号即可，信号同自定义指标一样，比如多头信号是 close>sma(30)，退出信号是sma5<sma30。

我们只需要给大脑添加这两个信号：

# 添加交易信号1
cerebro.add_signal(bt.SIGNAL_LONG, MySignal, period=30)
# 添加交易信号2
cerebro.add_signal(bt.SIGNAL_LONGEXIT, SMAExitSignal, p1=5, p2=30)

注意这里信号的逻辑：

signal指标取值大于0 → 对应多头 long 信号；

signal指标取值小于0 → 对应空头 short 信号；

signal指标取值等于0 → 不发出指令；

bt.SIGNAL_LONGSHORT：

多头信号和空头信号都会作为开仓信号；

对于多头信号，如果之前有空头仓位，会先对空仓进行平仓close，再开多仓；

空头信号也类似，会在开空仓前对多仓进行平仓close。

bt.SIGNAL_LONG：

多头信号用于做多，空头信号用于平仓close；

如果系统中同时存在LONGEXIT 信号类型，SIGNAL_LONG 中的空头信号将不起作用，将会使用 LONGEXIT中的空头信号来平仓多头，如上面的多条交易信号的例子。

bt.SIGNAL_SHORT：

空头信号用于做空，多头信号用于平仓；如果系统中同时存在SHORTEXIT 信号类型，SIGNAL_SHORT 中的多头信号将不起作用，将会使用 SHORTEXIT中的多头信号来平仓空头。

import backtrader as bt


# 自定义信号指标
class MySignal(bt.Indicator):
    lines = ('signal',) # 声明 signal 线，交易信号放在 signal line 上
    params = (('period', 30),)

    def __init__(self):
        self.lines.signal = self.data - bt.indicators.SMA(period=self.p.period)


# 定义交易信号1
class SMACloseSignal(bt.Indicator):
    lines = ('signal',)
    params = (('period', 30),)

    def __init__(self):
        self.lines.signal = self.data - bt.indicators.SMA(period=self.p.period)


# 定义交易信号2
class SMAExitSignal(bt.Indicator):
    lines = ('signal',)
    params = (('p1', 5), ('p2', 30),)

    def __init__(self):
        sma1 = bt.indicators.SMA(period=self.p.p1)
        sma2 = bt.indicators.SMA(period=self.p.p2)
        self.lines.signal = sma1 - sma2


# 实例化大脑
cerebro = bt.Cerebro()

import pandas as pd
def load_data(name):
    df = pd.read_csv('../data/csv/{}.csv'.format(name))
    df['date'] = df['date'].apply(lambda x: str(x))
    df.set_index('date', inplace=True)
    df.sort_index(ascending=True, inplace=True)
    df.dropna(inplace=True)
    return df


df = load_data('000300.SH')
#df_spx = load_data('spx')
def to_backtrader_dataframe(df):
    df.index = pd.to_datetime(df.index)
    df['openinterest'] = 0
    if 'amount' not in df.columns:
        df['amount'] = 0
    df = df[['open', 'high', 'low', 'close', 'volume','amount', 'openinterest']]
    return df


df = to_backtrader_dataframe(df)
# 加载数据
cerebro.adddata(bt.feeds.PandasData(dataname=df))
# 添加交易信号1
cerebro.add_signal(bt.SIGNAL_LONG, MySignal, period=30)
# 添加交易信号2
cerebro.add_signal(bt.SIGNAL_LONGEXIT, SMAExitSignal, p1=5, p2=30)

# 添加分析指标
# 返回年初至年末的年度收益率
cerebro.addanalyzer(bt.analyzers.AnnualReturn, _name='_AnnualReturn')
# 计算最大回撤相关指标
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='_DrawDown')
# 计算年化收益：日度收益
cerebro.addanalyzer(bt.analyzers.Returns, _name='_Returns', tann=252)
# 计算年化夏普比率：日度收益
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='_SharpeRatio', timeframe=bt.TimeFrame.Days, annualize=True, riskfreerate=0) # 计算夏普比率
cerebro.addanalyzer(bt.analyzers.SharpeRatio_A, _name='_SharpeRatio_A')
# 返回收益率时序
cerebro.addanalyzer(bt.analyzers.TimeReturn, _name='_TimeReturn')

result = cerebro.run()

# 提取结果
print("--------------- AnnualReturn -----------------")
print(result[0].analyzers._AnnualReturn.get_analysis())
print("--------------- DrawDown -----------------")
print(result[0].analyzers._DrawDown.get_analysis())
print("--------------- Returns -----------------")
print(result[0].analyzers._Returns.get_analysis())
print("--------------- SharpeRatio -----------------")
print(result[0].analyzers._SharpeRatio.get_analysis())
print("--------------- SharpeRatio_A -----------------")
print(result[0].analyzers._SharpeRatio_A.get_analysis())

02 策略分析器

策略结果怎么样？风险、收益如何评价。

bt框架里专门负责回测收益评价指标计算的模块analyzers，大家可以将其称为“策略分析器”。

第一步：通过addanalyzer(ancls, _name, *args, **kwargs) 方法将分析器添加给大脑，ancls 对应内置的分析器类；

第二步：分别基于results = cerebro.run() 返回的各个对象results[x]，提取该对象analyzers属性下的各个分析器的计算结果，并通过get_analysis() 来获取具体值。

由于结果是保存在OrderedDict里，需要显示的话，可以自行根据需要的字段提取。

03 自定义分析器

Backtrader支持继承bt.Analyzer，然后实现自己的分析器。

最基础只需要实现stop和get_analysia两个函数即可。

# 官方提供的 SharpeRatio 例子
class SharpeRatio(Analyzer):
    params = (('timeframe', TimeFrame.Years), ('riskfreerate', 0.01),)

    def __init__(self):
        super(SharpeRatio, self).__init__()
        self.anret = AnnualReturn()

    def start(self):
        # Not needed ... but could be used
        pass

    def next(self):
        # Not needed ... but could be used
        pass

    def stop(self):
        retfree = [self.p.riskfreerate] * len(self.anret.rets)
        retavg = average(list(map(operator.sub, self.anret.rets, retfree)))
        retdev = standarddev(self.anret.rets)
        self.ratio = retavg / retdev

    def get_analysis(self):
        return dict(sharperatio=self.ratio)

04 优化参数

把add_strategy改为optstrategy，即可以对参数区间进行批量回测：

# 添加优化器
cerebro1.optstrategy(TestStrategy, period1=range(5, 25, 5), period2=range(10, 41, 10))

# 添加分析指标
# 返回年初至年末的年度收益率
cerebro1.addanalyzer(bt.analyzers.AnnualReturn, _name='_AnnualReturn')
# 计算最大回撤相关指标
cerebro1.addanalyzer(bt.analyzers.DrawDown, _name='_DrawDown')
# 计算年化收益
cerebro1.addanalyzer(bt.analyzers.Returns, _name='_Returns', tann=252)
# 计算年化夏普比率
cerebro1.addanalyzer(bt.analyzers.SharpeRatio_A, _name='_SharpeRatio_A')
# 返回收益率时序
cerebro1.addanalyzer(bt.analyzers.TimeReturn, _name='_TimeReturn')

# 启动回测
result = cerebro1.run()


# 打印结果
def get_my_analyzer(result):
    analyzer = {}
    # 返回参数
    analyzer['period1'] = result.params.period1
    analyzer['period2'] = result.params.period2
    # 提取年化收益
    analyzer['年化收益率'] = result.analyzers._Returns.get_analysis()['rnorm']
    analyzer['年化收益率（%）'] = result.analyzers._Returns.get_analysis()['rnorm100']
    # 提取最大回撤(习惯用负的做大回撤，所以加了负号)
    analyzer['最大回撤（%）'] = result.analyzers._DrawDown.get_analysis()['max']['drawdown'] * (-1)
    # 提取夏普比率
    analyzer['年化夏普比率'] = result.analyzers._SharpeRatio_A.get_analysis()['sharperatio']

    return analyzer


ret = []
for i in result:
    ret.append(get_my_analyzer(i[0]))

pd.DataFrame(ret)