关于程序员:Backtrader-数据篇

Backtrader 数据篇

上一篇曾经学习了整个框架的大略划分模块，对于 Backtrader 这框架有了一个大略的意识，这篇将学习这框架基石 – 数据流程。<br/>

上面将会一一解答对于这框架 数据方面的疑难 <br/>
怎么将数据填充到框架种？<br/>
数据组织模式是怎么样？<br/>
怎么拜访数据？<br/>
在回测的过程种，数据流是以怎么模式传递?<br/>
怎么扩大其余回测指标？<br/>

数据填充

在简介篇，曾经晓得所有的模块有大脑对立管制治理，数据模块也不例外，所以首先要实例化 Cerebro，而后往这 Cerebro 塞数据，再对立调度治理。

上面数据是以 dataframe 数据格式为例的介绍，当然 backtrader 也反对其余数据标准的格局，都差不多。

backtrader 把标准的数据包装成了 Data Feeds , 其实就是一个回测数据汇合治理类。数据能够增加一个或者多个股票数据。

如何填充数据？

只须要指定数据源，回测时间段，即可增加多个股票数据源。

示例代码

# 股票查问开始和完结工夫
start_query = '2022-01-01'
end_query = '2022-09-01'
# 回测开始和完结工夫
start_date = datetime.datetime(2022, 7, 10)
end_date = datetime.datetime(2022, 8, 10)
cerebro = bt.Cerebro()
# 增加几个股票数据
codes = [
    '300015',
    '300347',
    '300760',
    '603127',
    '600438'
]
# 增加多个股票回测数据
for code in codes:
    data = sdb.stock_daily(code, start_query, end_query)
    data.index.names = ['datetime']
    date_feed = bt.feeds.PandasData(dataname=data, fromdate=start_date, todate=end_date)
    cerebro.adddata(date_feed, name=code)
    print('增加股票数据：code: %s' % code)

留神：查问的数据时间段与回测时间段是离开的，最终数据流是以回测时间段测试数据。

参数 name 是这张表的表名。
self.datas[0]._name 能够打印表名。

存储形式

把每个股票数据看作为一张表，其实就是一张指标维度和工夫维度的表，self.datas 就是汇合了多个股票的数据集，造成了一个三维数据源，别离是：数据表维度、工夫维度、指标维度。

如上图所示，Data Feeds 种的 self.datas 数据类型是 list. 每个元素是一张蕴含工夫维度和指标维度的股票数据表。

数据表维度

该维度其实就是 list 汇合，汇合了所有增加进来了股票数据，每个股票数据都有工夫维度和指标维度形成的数据表。<br/>
self.datas[N]

指标维度

该维度是回测时应用的指标，除了罕用指标还能够自定义指标。<br/>
self.datas[N].lines.xxx[M]

参数

所有的指标

self.data.lines.getlinealiases()

控制台
>> lines ('close', 'low', 'high', 'open', 'volume', 'openinterest', 'datetime')

工夫维度

工夫维度其实是回测时间段，fromdate ~ todate 之间。<br/>
self.data[N].lines.datetime.date(M)

数据索引

self.datas 数据类型是 list<br/>

1. 下标索引：self.datas[N]，其中 N 为 0 到 N-1 时，是正向，N 为 -1 到 -N 时为反向
2. 缩写索引，self.dataN，留神不是 datas。N 为 0 到 N-1。
3. 表名索引，self.getdatabyname(‘name’)，其中 name 为导入数据时 adddata(date_feed, name=code) 时设置的表名。
4. 第一个数据集索引，self.datas[0] 等价 self.data0 等价 self.data<br/>

留神：self.datas 和 self.data 的区别，self.datas 是数据汇合 list. self.data 是第 0 个数据。self.dataN 的拜访是没有 s 的。

self.datas[N].lines

索引形式 <br/>
日期：self.datas[N].lines.datetime.date(N)
其余：self.datas[N].lines. 其余字段名(open、high、low、close、volume)[N]<br/>

留神：datetime 是以 float 数据类型存储，拜访是须要借助 xxx.date(N) 函数进行转换。

示例

# 拜访第一个数据表的 close 指标的索引 0
self.data[0].lines.close[0]

切片形式

self.data1.lines.close.get(ago=N, size=M))

返回值：array 数组 [close[N-(M-1)],…,close[N-1],close[N]]<br/>
如果 N – (M – 1) <1，返回空数组 []。<br/>
列如：N = 3，M = 4，返回 [].

策略中的数据流

回测长度：N = self.data.buflen()<br/>

索引下标 0 的含意

索引下标 0 在 init() 函数和 next() 函数是不一样的，在 init() 函数中索引 0 是代表回测工夫 todate，在 next() 函数中 索引 0 是代表以后回测的工夫。

init()函数中的数据流

在 init() 函数中，索引 0 是 todate，索引 1 是 fromdate. 反对 正向和反向拜访的两种形式。

• 正向索引的索引下标为 1、2 … N
• 反向索引下标为 0、-1、-2 … -(N-1).

其中，对应下标值是雷同 <br/>

next()函数中的数据流

在 next() 函数中，索引 0 永远是以后的工夫节点，索引 0 随着以工夫维度的循环，不停的挪动。backtrader 是曾经回测过的了，forward 是还没有回测到的。

自定义读取数据

如果你感觉每次都要设置这么多参数来告知指标地位很麻烦，那你也能够从新自定义数据读取函数，自定义的形式就是继承数据加载类 GenericCSVData、PandasData 再构建一个新的类，而后在新的类里对立设置参数：

class My_PandasData(bt.feeds.PandasData):
    params = (('fromdate', datetime.datetime(2019, 1, 2)),
        ('todate', datetime.datetime(2021, 1, 28)),
        ('nullvalue', 0.0),
        ('dtformat', ('%Y-%m-%d')),
        ('datetime', 0),
        ('time', -1),
        ('high', 3),
        ('low', 4),
        ('open', 2),
        ('close', 5),
        ('volume', 6),
        ('openinterest', -1)
    )

新增指标

backtrader 除了提供根本的指标外(‘close’, ‘low’, ‘high’, ‘open’, ‘volume’, ‘openinterest’, ‘datetime’)，还提供给用户自定义扩大。例如：macd,pe,pb 等等指标。<br/>

如何扩大？<br/>

1. 继承 bt.feeds.PandasData 类，定义新指标。
2. 数据源扩大新指标对应的列，并初始化新指标值。

示例

# 继承
class PandasData_more(bt.feeds.PandasData):
    lines = ('pe', 'pb',)  # 要增加的线
    # 设置 line 在数据源上的列地位
    params = (('pe', -1),
        ('pb', -1),
    )

# 数据源新增列
# 股票查问开始和完结工夫
start_query = '2022-01-01'
end_query = '2022-09-01'
# 回测开始和完结工夫
start_date = datetime.datetime(2022, 7, 11)
end_date = datetime.datetime(2022, 8, 10)
cerebro = bt.Cerebro()
# 增加几个股票数据
codes = [
    '300015',
    '300347',
    '300760',
    '603127',
    '600438'
]
# 增加多个股票回测数据
for code in codes:
    data = sdb.stock_daily(code, start_query, end_query)
    data.index.names = ['datetime']
    # 新增指标列
    data['pe'] = 3
    data['pb'] = 4
    date_feed = PandasData_more(dataname=data, fromdate=start_date, todate=end_date)
    cerebro.adddata(date_feed, name=code)
    print('增加股票数据：code: %s' % code)

cerebro.addstrategy(TestStrategy)
cerebro.run()

罕用函数手册

测试示例

import datetime

import backtrader as bt

import stock_db as sdb


class PandasData_more(bt.feeds.PandasData):
    lines = ('pe', 'pb',)  # 要增加的线
    # 设置 line 在数据源上的列地位
    params = (('pe', -1),
        ('pb', -1),
    )


class My_PandasData(bt.feeds.PandasData):
    params = (('fromdate', datetime.datetime(2019, 1, 2)),
        ('todate', datetime.datetime(2021, 1, 28)),
        ('nullvalue', 0.0),
        ('dtformat', ('%Y-%m-%d')),
        ('datetime', 0),
        ('time', -1),
        ('high', 3),
        ('low', 4),
        ('open', 2),
        ('close', 5),
        ('volume', 6),
        ('openinterest', -1)
    )


class TestStrategy(bt.Strategy):
    # 可选，设置回测的可变参数：如挪动均线的周期
    # params = (#     (..., ...),  # 最初一个“,”最好别删！# )

    def log(self, txt, dt=None):
        """
        可选，构建策略打印日志的函数：可用于打印订单记录或交易记录等
        :param txt:
        :param dt:
        :return:
        """
        dt = dt or self.datas[0].datetime.date(0)
        print('%s,%s' % (dt.isoformat(), txt))

    def __init__(self):
        """必选，初始化属性、计算指标等"""

        print(type(self.datas[0].lines.datetime[0]))
        print(type(self.datas[0].lines.volume[0]))
        self.count = 0
        print("------------- init 中的索引地位 -------------")
        print('lines', self.data1.lines.getlinealiases())
        print('datas type', type(self.datas))
        print("0 索引：", 'datetime', self.datas[1].lines.datetime.date(0), 'close', self.data1.lines.close[0])
        print('-1 索引：', 'datetime', self.data1.lines.datetime.date(-1), 'close', self.data1.lines.close[-1])
        print('-2 索引：', 'datetime', self.data1.lines.datetime.date(-2), 'close', self.data1.lines.close[-2])
        print('1 索引：', 'datetime', self.data1.lines.datetime.date(1), 'close', self.data1.lines.close[1])
        print('2 索引：', 'datetime', self.data1.lines.datetime.date(2), 'close', self.data1.lines.close[2])
        print('从 0 开始往前取 3 天的收盘价：', self.data1.lines.close.get(ago=0, size=3))
        print("从 - 1 开始往前取 3 天的收盘价：", self.data1.lines.close.get(ago=3, size=3))
        print("从 - 2 开始往前取 3 天的收盘价：", self.data1.lines.close.get(ago=-2, size=3))
        print("回测的总长度：", self.data.buflen())
        pass

    def notify_order(self, order):
        """
        可选，打印订单信息
        :param order:
        :return:
        """
        pass

    def notify_trade(self, trade):
        """
        可选，打印交易信息
        :param trade:
        :return:
        """
        pass

    def next(self):
        """
        必选，编写交易策略逻辑
        :return:
        """print(f"------------- next 的第 {self.count + 1} 次循环 --------------")
        print('以后节点(今日):', 'datetime', self.data1.lines.datetime.date(0), 'close', self.data1.lines.close[0])
        print("往前推 1 天（昨日）：", 'datetime', self.data1.lines.datetime.date(-1), 'close', self.data1.lines.close[-1])
        print("往前推 2 天（前日）", 'datetime', self.data1.lines.datetime.date(-2), 'close', self.data1.lines.close[-2])
        print("前日、昨日、今日的收盘价：", self.data1.lines.close.get(ago=0, size=3))
        # print("往后推 1 天（明日）：", 'datetime', self.data1.lines.datetime.date(1), 'close', self.data1.lines.close[1])
        # print("往后推 2 天（明后日）", 'datetime', self.data1.lines.datetime.date(2), 'close', self.data1.lines.close[2])
        print("已解决的数据点：", len(self.data))
        print("回测的总长度：", self.data.buflen())
        self.count += 1
        pass


# 股票查问开始和完结工夫
start_query = '2022-01-01'
end_query = '2022-09-01'
# 回测开始和完结工夫
start_date = datetime.datetime(2022, 7, 11)
end_date = datetime.datetime(2022, 8, 10)
cerebro = bt.Cerebro()
# 增加几个股票数据
codes = [
    '300015',
    '300347',
    '300760',
    '603127',
    '600438'
]
# 增加多个股票回测数据
for code in codes:
    data = sdb.stock_daily(code, start_query, end_query)
    data.index.names = ['datetime']
    # 新增指标列
    data['pe'] = 3
    data['pb'] = 4
    date_feed = PandasData_more(dataname=data, fromdate=start_date, todate=end_date)
    cerebro.adddata(date_feed, name=code)
    print('增加股票数据：code: %s' % code)

cerebro.addstrategy(TestStrategy)
cerebro.run()
# cerebro.plot()

if __name__ == '__main__':
    pass

写于 2022 年 10 月 05 日 14:24:59

本文由 mdnice 多平台公布