Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
latency between system and bt clock
import datetime
import backtrader as bt
import pytz
import pandas as pd
from utils.constants import getIBFX
class TickerStrat(bt.Strategy):
params = (
('tickNameEndswith', None),
)
def notify_data(self, data, status, *args, **kwargs):
print('*' * 5, datetime.datetime.utcnow(), 'DATA NOTIF:', data._getstatusname(status), data._name, *args)
def notify_store(self, msg, *args, **kwargs):
print('*' * 5, datetime.datetime.utcnow(), 'STORE NOTIF:', msg)
def __init__(self):
# self.data_length = [len(data) for data in self.datas]
self.data_length = [0 for data in self.datas]
self.dt = [None for i in self.datas]
return
def printAllTicks(self):
print(datetime.datetime.utcnow())
for i, x in enumerate(self.datas):
if self.p.tickNameEndswith not in self.datas[i]._name:
continue
if self.dt[i] != self.datas[i].datetime.datetime():
marker = '---'
self.dt[i] = self.datas[i].datetime.datetime()
else:
marker = '***'
print(self.datas[i]._name, self.datas[i].datetime.datetime(), self.datas[i].close[0], marker)
print()
def next(self):
self.printAllTicks()
def main():
cerebro = bt.Cerebro(stdstats=False)
ibstore = bt.stores.IBStore(
host='127.0.0.1', port=4002, clientId=42, reconnect=-1)
broker = ibstore.getbroker()
cerebro.setbroker(broker)
tickNameEndswith = '_tick'
instrumentNames = getIBFX()
instrumentNames = instrumentNames[:3]
for instrumentName in instrumentNames:
data = ibstore.getdata(dataname=instrumentName, tz=pytz.timezone('UTC'))
cerebro.adddata(data, name='{}{}'.format(instrumentName, tickNameEndswith))
cerebro.addstrategy(TickerStrat, tickNameEndswith=tickNameEndswith)
cerebro.run(tradehistory=True)
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment