Custom live data feeding problem

[frank.zhang]

I want to do live data feed through Redis message queue, but got error message. The code:

class LiveDataTest(bt.feed.DataBase):        
    
    def __init__(self):
        super(LiveDataTest, self).__init__()
        
    
    def start(self):
        self.redis = redis.StrictRedis(host='localhost', port=6379, db=0)
        self.ps = self.redis.pubsub()
        self.ps.subscribe('marketdata')
        for item in self.ps.listen():
            if item['type'] == 'message':
                self.d = item['data']
                self._load()
    
    def _load(self):
        d = json.loads(self.d)
        print(d)
        self.lines.datetime[0] = d['date']
        self.lines.open[0] = d['open']
        self.lines.close[0] = d['close']
        self.lines.high[0] = d['high']
        self.lines.low[0] = d['low']
        self.lines.volume[0] = d['volume']
        return True
    
    def stop(self):
        self.redis.close()


class MyStrategy(bt.Strategy):
    
    def __init__(self):
        pass
    
    def next(self):
        print('date:{},close:{}'.format(self.data.datetime.date(0), self.data.close[0]))


cerebro = bt.Cerebro()
data = LiveDataTest()
cerebro.addstrategy(MyStrategy)
cerebro.adddata(data)
cerebro.run()

And the error message is as follow:

{'date': '2021-07-21', 'close': 1234.03, 'open': 1234.0, 'high': 1234.0, 'low': 1234.0, 'volume': 12345}
---------------------------------------------------------------------------
IndexError                                Traceback (most recent call last)
<ipython-input-7-6e26a703b6c1> in <module>
      3 cerebro.addstrategy(MyStrategy)
      4 cerebro.adddata(data)
----> 5 cerebro.run()

~/.conda/envs/backtrader/lib/python3.7/site-packages/backtrader/cerebro.py in run(self, **kwargs)
   1125             # let's skip process "spawning"
   1126             for iterstrat in iterstrats:
-> 1127                 runstrat = self.runstrategies(iterstrat)
   1128                 self.runstrats.append(runstrat)
   1129                 if self._dooptimize:

~/.conda/envs/backtrader/lib/python3.7/site-packages/backtrader/cerebro.py in runstrategies(self, iterstrat, predata)
   1208                 if self._exactbars < 1:  # datas can be full length
   1209                     data.extend(size=self.params.lookahead)
-> 1210                 data._start()
   1211                 if self._dopreload:
   1212                     data.preload()

~/.conda/envs/backtrader/lib/python3.7/site-packages/backtrader/feed.py in _start(self)
    201 
    202     def _start(self):
--> 203         self.start()
    204 
    205         if not self._started:

<ipython-input-5-6ce1b88d9462> in start(self)
     12             if item['type'] == 'message':
     13                 self.d = item['data']
---> 14                 self._load()
     15 
     16     def _load(self):

<ipython-input-5-6ce1b88d9462> in _load(self)
     17         d = json.loads(self.d)
     18         print(d)
---> 19         self.lines.datetime[0] = d['date']
     20         self.lines.open[0] = d['open']
     21         self.lines.close[0] = d['close']

~/.conda/envs/backtrader/lib/python3.7/site-packages/backtrader/linebuffer.py in __setitem__(self, ago, value)
    220             value (variable): value to be set
    221         '''
--> 222         self.array[self.idx + ago] = value
    223         for binding in self.bindings:
    224             binding[ago] = value

IndexError: array assignment index out of range

Anyone could help? thanks

[frank.zhang]

After some research, the problem has been resolved.

1. data feed should be in _load() rather than in start()
2. add islive() function
   here is the modified code:

class LiveDataTest(bt.feed.DataBase):

    def __init__(self):
        super(LiveDataTest, self).__init__()

    def start(self):
        self.redis = redis.StrictRedis(host='localhost', port=6379, db=0)
        self.ps = self.redis.pubsub()
        self.ps.subscribe('marketdata')

    def _load(self):
        for item in self.ps.listen():
            if item['type'] == 'message':
                self.d = item['data']

                bar = json.loads(self.d)
                print(bar)
                self.lines.datetime[0] = bt.date2num(pd.to_datetime(bar['date']))
                self.lines.open[0] = bar['open']
                self.lines.close[0] = bar['close']
                self.lines.high[0] = bar['high']
                self.lines.low[0] = bar['low']
                self.lines.volume[0] = bar['volume']
                return True

    def stop(self):
        self.redis.close()

    def islive(self):
        return True