Backtrader - 自订 datafeeds

我们之前在喂历史资料，都是先用 shioaji 下载下来，然後再用 padas 转成 dataframe，最後再喂给 backtrader，在 backtrader 里的 datafeeds 也是可以自订义的，有几个步骤:

1. 定义一个 datafeeds 的 class 继承 backtrader.feed.DataBase
2. 定义需要的 params 参数：
   这边要注意的是，backtrader.feed.DataBase 有自己的参数，要避免使用一样的参数名称，以下是 DataBase 的参数名称

       params = (('dataname', None),
        ('fromdate', datetime.datetime.min),
        ('todate', datetime.datetime.max),
        ('name', ''),
        ('compression', 1),
        ('timeframe', TimeFrame.Days),
        ('sessionend', None))
   

3. 覆写初始化的方法 __init__(self) 或是 start(self)，也可以两个都用
4. 如果在结束时，有需要清除的程序，可以放在 stop(self)
5. 覆写主要载入资料的 _load(self)

以下我就用 Shioaji 取得资料来示范

import backtrader as bt
import backtrader.feeds as btfeeds
from backtrader import date2num
import shioaji as sj
from datetime import datetime


class ShioajiFeeds(bt.feed.DataBase):
    params = (
        ("person_id", "身份证字号"),
        ("passwd", "永丰金证券密码"),
        ("start", datetime.now().strftime("%Y-%m-%d")),
        ("end", datetime.now().strftime("%Y-%m-%d")),
        ("stock", "2330"),
        # 这个变数是 DataBase 的
        # 因为 Shioaji 取得的资料是每分钟，所以设成 TimeFrame.Minutes
        ("timeframe", bt.TimeFrame.Minutes),
    )

    def start(self):
        self.api = sj.Shioaji()
        self.api.login(person_id = self.p.person_id, passwd = self.p.passwd)
        contract = self.api.Contracts.Stocks[self.p.stock]

        self.kbar = self.api.kbars(contract, start = self.p.start, end = self.p.end)

        self.idx = 0
        self.dataLength = len(self.kbar.ts) # 纪录我们取得的资料有多少笔


    def _load(self):
        # 资料会一笔一笔读进来，如果读到最後一笔，就跳出 False
        if (self.idx == self.dataLength):
            return False
            
        dt = datetime.utcfromtimestamp(self.kbar.ts[self.idx] / 10**9)
        self.lines.datetime[0] = date2num(dt)
        
        self.lines.open[0] = self.kbar.Open[self.idx]
        self.lines.high[0] = self.kbar.High[self.idx]
        self.lines.low[0] = self.kbar.Low[self.idx]
        self.lines.close[0] = self.kbar.Close[self.idx]
        self.lines.volume[0] = self.kbar.Volume[self.idx]
                
        self.idx += 1
        
        return True


    def stop(self):
        # 结束的时候进行登出
        self.api.logout()
        self.api = None
        

# 使用自订的 datafeed
cerebro = bt.Cerebro()

data = ShioajiFeeds(start='2021-10-01', end='2021-10-08', stock='2303')

cerebro.resampledata(data, timeframe=bt.TimeFrame.Days)

cerebro.addstrategy(bt.Strategy)

cerebro.run()

cerebro.plot()

那如果我们要载入多笔股票的资料，以这个写法，需要进行登入登出多次，所以可以稍微改一下，把 api 当成一个参数去执行

class ShioajiFeeds(bt.feed.DataBase):
    params = (
        ("person_id", "身份证字号"),
        ("passwd", "永丰金证券密码"),
        ("start", datetime.now().strftime("%Y-%m-%d")),
        ("end", datetime.now().strftime("%Y-%m-%d")),
        ("stock", "2330"),
        ("timeframe", bt.TimeFrame.Minutes),
        ("api", None), # 新增一个 api 参数
    )

    def start(self):
        if self.p.api is None:
            self.api = sj.Shioaji()
            self.api.login(person_id = self.p.person_id, passwd = self.p.passwd)
        else:
            self.api = self.p.api
    # ...略
    
    def stop(self):
        if self.p.api is None:
            self.api.logout()
            
        self.api = None

在执行的时候，我们就可以先初始化 api，然後把 api 当参数传入，就可以不用一直登入登出了

api = sj.Shioaji()
api.login(person_id = "帐号", passwd = "密码")

data = ShioajiFeeds(start = '2021-10-01', end = '2021-10-08', stock = '2330', api = api)

# ... 略

# 结束後，记得登出
api.logout()