关于后端:Backtrader-策略回测初探

36次阅读

共计 8310 个字符,预计需要花费 21 分钟才能阅读完成。

Backtrader 策略回测初探

这篇介绍简略的回测流程,次要的内容如下:

  • 回测函数介绍
  • 单股回测
  • 多股回测

回测函数

回测策略类很简洁,间接继承 bt.Strategy,复写父类的办法,最初把回测策略类增加到大脑即可。

回测参数
回测类参数增加通过属性变量 params 记录,能够是元组模式,也能够是字典模式。

  • 定义参数 <br/>

    # 元组模式,留神最初一个,逗号别删除
    params = (('maperiod', 20),
      )
    
    或
    # 字典模式
    params = {'maperiod': 20}
  • 应用 <br/>
    通过 self.p.maperiod 拜访提取。

    bt.ind.SMA(self.data, period=self.p.maperiod)
  • 传参 <br/>
    将策略类传入大脑时,传入参数

    cerebro.addstrategy(TestStrategy, maperiod=5)

函数 <br/>
因为继承了策略类 bt.Strategy,运行策略时会回调这些函数,须要搞的事件是在相应的函数调用交易逻辑即可。<br/>
先说下 lines 线对象概念,每个指标都是一条 line 线对象,贯通所有的回测日期。bar 概念是每个日期对应所有的指标。简略了解就是 excel 表的中的列和行,line 线对象相当于 excel 的列,bar 概念相当于 excel 表的行。

函数名 函数阐明 调用机会
__init__() 初始化时调用,只调用一次 初始化
next() 每条 bar 调用一次,<br/> 直到所有 bar 回测结束,<br/> 无效的 bar 每条 bar 都回调
prenext() 不在 next()函数调用的 bar,<br/> 就调用到这函数 不在 next()回调的 bar
notify_order() 当执行 self.buy()、self.order_target_percent()、<br/>self.sell()等时触发回调 订单状态发生变化回调
notify_trade() 交易扭转是 交易产生时回调
log() 日志打印函数,打印一些交易信息 本人调用

还有其余一些函数,这里不一一介绍了,不是很重要的。

指标简介

指标在代码层面上的示意模式就是以 line 线对象的形式存在,贯通整个回测周期,个别在测类类的 __init__ 函数外面构建好 <br/>
在数据篇,有展现过在 feeds.Data 上间接扩大指标,上面介绍的是通过 backtrader 的指标对象来构建新的指标,新的指标也是 line 对象,贯通整个回测周期。

上面例子构建 20 日挪动平均线 SMA:<br/>

def __init__(self):
    self.sma = bt.ind.SMA(self.data, period=20)

留神: 这种模式构建的指标是基于数据篇外面所说的第一个表数据的指标,并不是针对所有表的。而且会影响到 next 回调机会。20 日挪动平均线的指标,前 20 日个回测 bar 是有效的,不会在 next() 函数回调,但回调到 prenext() 函数。

下图所示:<br/>

对于指标在这里不细说,前面会写一篇具体点的介绍。

回测繁难配置

只是做一个简略回测测试,所以繁难配置下经纪商,setcash() 配置了一小指标资产 1 亿,设置佣金 setcommission() 千分之一,addstrategy() 增加策略并设置自定义的参数。

# 设置资产
cerebro.broker.setcash(100000000.0)
# 设置佣金
cerebro.broker.setcommission(commission=0.001)
# 增加回测策略,设置自定义参数数值
cerebro.addstrategy(SingleTestStrategy, maperiod=20)
# 执行策略
cerebro.run()
# 画图
cerebro.plot()

这里不细说这配置了,前面再具体说一篇。

单股回测

上面进入主题,搞一个策略,回测下能不能赚钱,这里只是介绍应用办法,理论利用必定不能只靠一个指标。

策略
这策略很简略,当当日收盘价高于 20 日平均线时买买买!当当日收盘价低于 20 日挪动平均线时卖卖卖。

留神:

  • 该策略是以收盘价下单,以下单后的下一日开盘价作为交割价。
  • 最初一日不做任何的交易。所以在回测的最初一日的前一天必须下单卖出股票,以便在最初一根 bar 的开盘价做为交割价卖出,不然呈现将来函数。

交易罕用函数:

  1. self.order_target_percent(secu_data, target_pct, name=secu)
  2. self.order_target_value(secu_data, target_val, name=secu)
  3. self.buy(secu_data, order_amount, name=secu)
  4. self.sell(secu_data, order_amount, name=secu)

代码胜过一言半语,间接上残缺代码:<br/>

import datetime

import backtrader as bt
import pandas as pd

import stock_db as sdb


class SingleTestStrategy(bt.Strategy):
    params = (('maperiod', 20),
    )

    def __init__(self):
        self.order = None
        self.sma = bt.ind.SMA(self.data, period=self.p.maperiod)
        pass

    def downcast(self, amount, lot):
        return abs(amount // lot * lot)

    # 能够不要,但如果你数据未对齐,须要在这里测验
    def prenext(self):
        print('prenext 执行', self.datetime.date(), self.getdatabyname('300015')._name
              , self.getdatabyname('300015').close[0])
        pass

    def next(self):
        # 查看是否有指令执行,如果有则不执行这 bar
        if self.order:
            return
        # 回测如果是最初一天,则不进行交易
        if pd.Timestamp(self.data.datetime.date(0)) == end_date:
            return
        if not self.position:  # 没有持仓
            # 执行买入条件判断:收盘价格上涨冲破 20 日均线;# 不要在股票剔除日前一天进行买入
            if self.datas[0].close > self.sma and pd.Timestamp(self.data.datetime.date(1)) < end_date:
                # 永远不要满仓买入某只股票
                order_value = self.broker.getvalue() * 0.98
                order_amount = self.downcast(order_value / self.datas[0].close[0], 100)
                self.order = self.buy(self.datas[0], order_amount, name=self.datas[0]._name)

        else:
            # 执行卖出条件判断:收盘价格跌破 20 日均线,或者股票剔除
            if self.datas[0].close < self.sma or pd.Timestamp(self.data.datetime.date(1)) >= end_date:
                # 执行卖出
                self.order = self.order_target_percent(self.datas[0], 0, name=self.datas[0]._name)
                self.log(f'卖{self.datas[0]._name},price:{self.datas[0].close[0]:.2f},pct: 0')
        pass

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            # Buy/Sell order submitted/accepted to/by broker - Nothing to do
            return

        # Check if an order has been completed
        # Attention: broker could reject order if not enough cash
        if order.status in [order.Completed, order.Canceled, order.Margin]:
            if order.isbuy():
                self.log(f"买入{order.info['name']}, 成交量{order.executed.size},成交价{order.executed.price:.2f} 订单状态:{order.status}")
                self.log('买入后以后资产:%.2f 元' % self.broker.getvalue())
            elif order.issell():
                self.log(f"卖出{order.info['name']}, 成交量{order.executed.size},成交价{order.executed.price:.2f} 订单状态:{order.status}")
                self.log('卖出后以后资产:%.2f 元' % self.broker.getvalue())
            self.bar_executed = len(self)

        # Write down: no pending order
        self.order = None

    def log(self, txt, dt=None):
        """
        输入日期
        :param txt:
        :param dt:
        :return:
        """
        dt = dt or self.datetime.date(0)  # 当初的日期
        print('%s , %s' % (dt.isoformat(), txt))

    pass

    def notify_trade(self, trade):
        '''可选,打印交易信息'''
        pass


# 开始查问工夫
start_query = '2019-01-01'
end_query = '2022-09-01'

# 开始回测工夫
from_date = datetime.datetime(2022, 1, 1)
to_date = datetime.datetime(2022, 10, 10)
cerebro = bt.Cerebro()
# 增加几个股票数据
codes = [
    '300015',
    # '300347',
    # '300760',
    # '603127',
    # '600438'
]

# 增加多个股票回测数据
end_date = 0
for code in codes:
    data = sdb.stock_daily(code, start_query, end_query)
    data.index.names = ['datetime']
    data_feed = bt.feeds.PandasData(dataname=data,
                                    fromdate=from_date,
                                    todate=to_date)
    cerebro.adddata(data_feed, name=code)
    end_date = data.index[-1]  # 股票剔除日
    print('增加股票数据:code: %s' % code)

cerebro.broker.setcash(100000000.0)
cerebro.broker.setcommission(commission=0.001)
cerebro.addstrategy(SingleTestStrategy, maperiod=20)
cerebro.run()
cerebro.plot()

if __name__ == '__main__':
    pass

后果:

em em … 一个亿的资产,亏了靠近 1000w,策略失败!!!!

来看看回测图,backtrader 的回测图确实有点丑哈,前面会有重构可视化篇的。<br/>

最下面是资产剖析图,行情数据区域的绿色三角形是买入,红色三角形是卖出。

多股回测

单股回测,下面曾经介绍了,那如何多个股同时回测呢?在这个问题上,咱们首先要解决的是多个股的指标计算并存储起来。

策略逻辑和下面的雷同。计算均线的时候用了 dict 循环计算每只股票的指标。

  1. self.getdatanames()按程序返回所有股票的名称 list
  2. self.getdatabyname(secu_name): 返回该股票的 data

所以,在给大脑塞数据时,须要指定 feedData 的 name , 对立用股票代码赋值,这样不便前面的索引。

间接上图说下整个流程逻辑:

代码只是再单股回测的根底下增加多股指标和多股持仓交易判断,策略和单股雷同。<br/>
代码如下:<br/>

import datetime

import backtrader as bt
import pandas as pd

import stock_db as sdb


class MultiTestStrategy(bt.Strategy):
    params = (('maperiod', 20),
    )

    def prenext(self):
        pass

    def downcast(self, amount, lot):
        return abs(amount // lot * lot)

    def __init__(self):
        # 初始化交易指令
        self.order = None
        self.buy_list = []
        # 增加挪动平均线指标,循环计算每个股票的指标
        self.sma = {x: bt.ind.SMA(self.getdatabyname(x), period=self.p.maperiod) for x in self.getdatanames()}

    def next(self):
        if self.order:  # 查看是否有指令期待执行
            return
        # 如果是最初一天,不进行交易
        if pd.Timestamp(self.datas[0].datetime.date(0)) == end_dates[self.datas[0]._name]:
            return
        # 是否持仓
        if len(self.buy_list) < 2:  # 没有持仓
            # 没有购买的票
            for secu in set(self.getdatanames()) - set(self.buy_list):
                data = self.getdatabyname(secu)
                # 如果冲破 20 日均线买买买,不要在最初一根 bar 的前一天买
                if data.close > self.sma[secu] and pd.Timestamp(data.datetime.date(1)) < end_dates[secu]:
                    # 买买买
                    order_value = self.broker.getvalue() * 0.48
                    order_amount = self.downcast(order_value / data.close[0], 100)
                    self.order = self.buy(data, size=order_amount, name=secu)
                    self.log(f"买{secu}, price:{data.close[0]:.2f}, amout: {order_amount}")
                    self.buy_list.append(secu)
        elif self.position:
            now_lst = []
            for secu in self.buy_list:
                data = self.getdatabyname(secu)
                # 执行卖出条件判断:收盘价格跌破 20 日均线,或者股票最初一根 bar 的前一天之剔除日
                if data.close[0] < self.sma[secu] or pd.Timestamp(data.datetime.date(1)) >= end_dates[secu]:
                    # 卖卖卖
                    self.order = self.order_target_percent(data, 0, name=secu)
                    self.log(f"卖{secu}, price:{data.close[0]:.2f}, pct: 0")
                    continue
                now_lst.append(secu)
            self.buy_list = now_lst

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            # Buy/Sell order submitted/accepted to/by broker - Nothing to do
            return

        # Check if an order has been completed
        # Attention: broker could reject order if not enough cash
        if order.status in [order.Completed, order.Canceled, order.Margin]:
            if order.isbuy():
                self.log(f"""买入{order.info['name']}, 成交量{order.executed.size},成交价{order.executed.price:.2f}""")
                self.log(f'资产:{self.broker.getvalue():.2f} 持仓:{[(x, self.getpositionbyname(x).size) for x in self.buy_list]}')
            elif order.issell():
                self.log(f"""卖出{order.info['name']}, 成交量{order.executed.size},成交价{order.executed.price:.2f}""")
                self.log(f'资产:{self.broker.getvalue():.2f} 持仓:{[(x, self.getpositionbyname(x).size) for x in self.buy_list]}')
            self.bar_executed = len(self)

        # Write down: no pending order
        self.order = None

    def log(self, txt, dt=None):
        """
        输入日期
        :param txt:
        :param dt:
        :return:
        """
        dt = dt or self.datetime.date(0)  # 当初的日期
        print('%s , %s' % (dt.isoformat(), txt))


# 开始查问工夫
start_query = '2019-01-01'
end_query = '2022-09-01'

# 开始回测工夫
from_date = datetime.datetime(2022, 1, 1)
to_date = datetime.datetime(2022, 10, 10)
cerebro = bt.Cerebro()
# 增加几个股票数据
codes = [
    '300015',
    '300347',
    # '300760',
    # '603127',
    # '600438'
]

# 增加多个股票回测数据
end_dates = {}
end_date = 0
for code in codes:
    data = sdb.stock_daily(code, start_query, end_query)
    data.index.names = ['datetime']
    data_feed = bt.feeds.PandasData(dataname=data,
                                    fromdate=from_date,
                                    todate=to_date)
    cerebro.adddata(data_feed, name=code)
    end_dates = data.index[-1]  # 股票剔除日
    end_date = data.index[-1]  # 股票剔除日
    print('增加股票数据:code: %s' % code)

cerebro.broker.setcash(100000000.0)
cerebro.broker.setcommission(commission=0.001)
cerebro.addstrategy(MultiTestStrategy, maperiod=20)
cerebro.run()
# 获取回测完结后的总资金
portvalue = cerebro.broker.getvalue()
# 打印后果
print(f'完结资金: {round(portvalue, 2)}')
cerebro.plot()

if __name__ == '__main__':
    pass

看下日志:

enen,竟然赚钱了,小赚靠近 1000w.

看下 backtracder 的交易点:

前面日期如同没触发交易逻辑,这前面再看看是咋回事。

写于 2022 年 10 月 23 日 10:27:29

本文由 mdnice 多平台公布

正文完
 0