My guess is that threading.Thread is a constructor only that doesn't start the thread until .start(). And start() looks to be triggered by a timer.
However when downloading many symbols (37) I also receive the 'Max rate of messages per second exceeded' error. A puzzle.
No idea why I cannot edit a post or delete old posts, but found another bug which would wipe out the historical data cache under certain downloads:
from ib.opt import ibConnection, message
from ib.ext.Contract import Contract
from time import time, strftime, sleep
from datetime import datetime, timedelta
import pandas as pd
from pathlib import Path
from random import randrange #for random connection ID only
class IBDataCache_FlatFile(object):
def _reset_data(self, host='127.0.0.1', port=4001, client_id=None):
#Initialize connection as long as it's not already connected:
if (not hasattr(self, '_conn')) or (not self._conn.isConnected()):
self._conn = ibConnection(host, port, client_id)
self._conn.enableLogging()
# Register the response callback function and type of data to be returned
self._conn.register(self._error_handler, message.Error)
self._conn.register(self._historical_data_handler, message.historicalData)
self._conn.register(self._save_order_id, 'NextValidId')
self._conn.register(self._nextValidId_handler, message.nextValidId)
self._conn.connect()
def _save_order_id(self, msg):
self._next_valid_id = msg.orderId
print('save order id', msg.orderId)
def _nextValidId_handler(self, msg):
print("nextValidId_handler: ", msg)
'''
self.inner(con_Id=self.req.m_conId, sec_type=self.req.m_secType, symbol=self.req.m_symbol, currency=self.req.m_currency, exchange=self.req.m_exchange, \
primaryExchange=self.req.m_primaryExch, endtime=self.req.endtime, duration=self.req.duration, \
bar_size=self.req.bar_size, what_to_show=self.req.what_to_show, use_rth=self.req.use_rth)
'''
def _error_handler(self, msg):
#print("_error_handler error: ", msg)
if not msg:
print('disconnecting', self._conn.disconnect())
elif 'No security definition has been found' in str(msg):
print ('ibcachedata: triggering timout')
self.timeout = time() # trigger timeout
def __init__(self, data_path='/docker_stocks/data', date_format='%Y%m%d %H:%M:%S', host='127.0.0.1', port=4001, client_id=None):
self._data_path = data_path
self._date_format = date_format
self._next_valid_id = 1
self._host = host
self._port = port
self._client_id = client_id
def _historical_data_handler(self, msg):
"""
Define historical data handler for IB - this will populate our pandas data frame
"""
# print (msg.reqId, msg.date, msg.open, msg.close, msg.high, msg.low)
#print("historical_data_handler: ", msg)
if not 'finished' in str(msg.date):
try:
dateidx = datetime.strptime(msg.date, self._date_format)
self._s = ([msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
except:
#for dates only with no time in str:
dateidx = datetime.strptime(msg.date, "%Y%m%d")
self._s = ([msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
self._df.loc[dateidx] = self._s
else:
print("ibcachedata: historical_data_handler: Received finish")
if not self._df.empty:
self._df.sort_index(inplace=True)
self._df.to_csv(self.filename)
self._conn.disconnect()
def setAllWithKwArgs(self, **kwargs):
#set all attributes to the kwargs to pass along
for key, value in kwargs.items():
setattr(self, key, value)
def inner(self,
con_Id=0,
sec_type='',
symbol='',
currency='USD',
exchange='',
primaryExchange='',
endtime='',
duration='250 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth='1'):
print ("called inner... setting up req.")
self.req = Contract()
self.req.m_conId = con_Id
self.req.m_secType = sec_type
self.req.m_symbol = symbol
self.req.m_currency = currency
self.req.m_exchange = exchange
self.primaryExch = primaryExchange
self.req.endtime = endtime
self.req.duration = duration
self.req.bar_size = bar_size
self.req.what_to_show = what_to_show
self.req.use_rth = use_rth
self._conn.reqHistoricalData(self._next_valid_id, self.req, endtime, duration,
bar_size, what_to_show, use_rth, 1)
def get_dataframe(self,
sec_type=None,
symbol=None,
currency='USD',
exchange=None,
PrimaryExchange='NASDAQ',
endtime=None,
duration='250 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth='1',
timeoutsecs=300):
date_format = '%Y%m%d %H:%M:%S'
self.timeout = 0
# build filename
self.filename = symbol + '_' + exchange + '_' + \
bar_size.replace(' ', '') + '_' + \
what_to_show + '_' + str(use_rth) + '.csv'
self.filename = self.filename.replace('/', '.')
self.filename = self._data_path + '/' + self.filename
print ("filename: ", self.filename)
# check if we have this cached
data_file = Path(self.filename)
if data_file.is_file():
self._df = pd.read_csv(self.filename,
parse_dates=True,
index_col=0)
else:
self._df = pd.DataFrame(columns=['Date', 'Open', 'High', 'Low', 'Close', 'Volume', 'OpenInterest'])
self._df.set_index('Date', inplace=True)
self._s = pd.Series()
# get filedt (the last datetime in the file)
if not self._df.empty:
filedt = pd.to_datetime(self._df.index[-1])
else:
filedt = datetime(1970, 1, 1)
if filedt < datetime.strptime(endtime, date_format): # DAYS ONLY
#Set duration of ENDTIME-FILEDT = new duration to downloadD
timediff = datetime.strptime(endtime, date_format) - filedt
timediff = timedelta(timediff.days)
else:
timediff = timedelta(0) #download nothing as the file is up to date
if self._df.empty or timediff > timedelta(0):
durationint = [int(s) for s in duration.split() if s.isdigit()]
durationint = durationint[0]
#test if the date difference to the end of the file exists, if so then scrape the non-cached data into the dataframe
#if the difference to the last bar is larger than the specified duration, just do the user specified duration -- the goal is to save time, not cost time
if timediff.days > 0 and filedt != datetime(1970, 1, 1) and timediff.days < durationint:
duration = str(timediff.days) + ' D'
# Not cached. Download it.
# Establish a Contract object and the params for the request
# set up connection:
self._reset_data(self._host, self._port, self._client_id)
self.inner(sec_type=sec_type, symbol=symbol, currency=currency, exchange=exchange,
primaryExchange=PrimaryExchange, endtime=endtime, duration=duration, bar_size=bar_size,
what_to_show=what_to_show, use_rth=use_rth)
# Make sure the connection doesn't get disconnected prior the response data return
self.timeout = time() + timeoutsecs
while self._conn.isConnected() and time() < self.timeout:
#print(".", end="", flush=True)
pass
if time() >= self.timeout:
self.timeout = 0
print ('Timed Out, disconnecting.')
self._conn.disconnect()
if not self._df.empty:
self._df.sort_index(inplace=True)
self._df.to_csv(self.filename)
return self._df
if __name__ == '__main__':
date_format = '%Y%m%d %H:%M:%S'
downloader_kwargs = dict(
data_path='../data',
date_format=date_format,
host='127.0.0.1',
port=4001,
client_id=randrange(99999999)
)
downloader = IBDataCache_FlatFile(**downloader_kwargs)
'''
stock_kwargs = dict(
sec_type='STK',
symbol='AAPL',
currency='USD',
exchange='ISLAND',
PrimaryExchange='NASDAQ',
endtime=datetime(2019, 8, 26, 15, 59).strftime(date_format),
duration='2 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth=1
)
df = downloader.get_dataframe(**stock_kwargs)
print(df)
'''
stock_kwargs = dict(
sec_type='STK',
symbol='AAPL',
currency='USD',
exchange='ISLAND',
PrimaryExchange='NASDAQ',
endtime=datetime(2019, 12, 10, 16, 1).strftime(date_format),
duration='2 D',
bar_size='1 min',
what_to_show='TRADES',
use_rth=1
)
df = downloader.get_dataframe(**stock_kwargs)
Fixed a bug that unnecessarily initialized the ipby connection before a download. There are a few tab errors due to c&ping:
from ib.opt import ibConnection, message
from ib.ext.Contract import Contract
from time import time, strftime
from datetime import datetime, timedelta
import pandas as pd
from pathlib import Path
from random import randrange #for random connection ID only
class IBDataCache_FlatFile(object):
def _reset_data(self, host='127.0.0.1', port=4001, client_id=None):
self._df = pd.DataFrame(columns=['Date', 'Open', 'High', 'Low', 'Close', 'Volume', 'OpenInterest'])
self._df.set_index('Date', inplace=True)
self._s = pd.Series()
#Initialize connection as long as it's not already connected:
if (not hasattr(self, '_conn')) or (not self._conn.isConnected()):
self._conn = ibConnection(host, port, client_id)
self._conn.enableLogging()
# Register the response callback function and type of data to be returned
self._conn.register(self._error_handler, message.Error)
self._conn.register(self._historical_data_handler, message.historicalData)
self._conn.register(self._save_order_id, 'NextValidId')
self._conn.register(self._nextValidId_handler, message.nextValidId)
self._conn.connect()
def _save_order_id(self, msg):
self._next_valid_id = msg.orderId
print('save order id', msg.orderId)
def _nextValidId_handler(self, msg):
print("nextValidId_handler: ", msg)
def _error_handler(self, msg):
if not msg:
print('disconnecting', self._conn.disconnect())
elif 'No security definition has been found' in str(msg):
print ('ibcachedata: triggering timout')
self.timeout = time() # trigger timeout
def __init__(self, data_path='/docker_stocks/data', date_format='%Y%m%d %H:%M:%S', host='127.0.0.1', port=4001, client_id=None):
self._data_path = data_path
self._date_format = date_format
self._next_valid_id = 1
self._host = host
self._port = port
self._client_id = client_id
def _historical_data_handler(self, msg):
"""
Define historical data handler for IB - this will populate our pandas data frame
"""
if not 'finished' in str(msg.date):
try:
dateidx = datetime.strptime(msg.date, self._date_format)
self._s = ([msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
except:
#for dates only with no time in str:
dateidx = datetime.strptime(msg.date, "%Y%m%d")
self._s = ([msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
self._df.loc[dateidx] = self._s
else:
print("ibcachedata: historical_data_handler: Received finish")
if not self._df.empty:
self._df.sort_index(inplace=True)
self._df.to_csv(self.filename)
self._conn.disconnect()
def inner(self,
con_Id=0,
sec_type='',
symbol='',
currency='USD',
exchange='',
primaryExchange='',
endtime='',
duration='250 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth='1'):
print ("called inner... setting up req.")
self.req = Contract()
self.req.m_conId = con_Id
self.req.m_secType = sec_type
self.req.m_symbol = symbol
self.req.m_currency = currency
self.req.m_exchange = exchange
self.primaryExch = primaryExchange
self.req.endtime = endtime
self.req.duration = duration
self.req.bar_size = bar_size
self.req.what_to_show = what_to_show
self.req.use_rth = use_rth
self._conn.reqHistoricalData(self._next_valid_id, self.req, endtime, duration,
bar_size, what_to_show, use_rth, 1)
def get_dataframe(self,
sec_type=None,
symbol=None,
currency='USD',
exchange=None,
PrimaryExchange='NASDAQ',
endtime=None,
duration='250 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth='1',
timeoutsecs=300):
date_format = '%Y%m%d %H:%M:%S'
self.timeout = 0
# build filename
self.filename = symbol + '_' + exchange + '_' + \
bar_size.replace(' ', '') + '_' + \
what_to_show + '_' + str(use_rth) + '.csv'
self.filename = self.filename.replace('/', '.')
self.filename = self._data_path + '/' + self.filename
print ("filename: ", self.filename)
# check if we have this cached
data_file = Path(self.filename)
if data_file.is_file():
self._df = pd.read_csv(self.filename,
parse_dates=True,
index_col=0)
else:
self._df = pd.DataFrame() #just create empty dataframe
#print ("ibcachedata: self._df.Date: ", self._df.Date)
#self._df.index = pd.to_datetime(self._df[Date], format='%Y-%m-%d %H:%M:%S')
#self._df.dropna(inplace=True)
# get filedt (the last datetime in the file)
if not self._df.empty:
filedt = pd.to_datetime(self._df.index[-1])
else:
filedt = datetime(1970, 1, 1)
if filedt < datetime.strptime(endtime, date_format): # DAYS ONLY
#SET DURATION SHORTER ENDTIME-FILEDT = NEW DURATION TO DOWNLOAD
timediff = datetime.strptime(endtime, date_format) - filedt
timediff = timedelta(timediff.days)
else:
timediff = timedelta(0) #download nothing as the file is up to date
if self._df.empty or timediff > timedelta(0):
durationint = [int(s) for s in duration.split() if s.isdigit()]
durationint = durationint[0]
#test if the date difference to the end of the file exists, if so then scrape the non-cached data into the dataframe
#if the difference to the last bar is larger than the specified duration, just do the user specified duration -- the goal is to save time, not cost time
if timediff.days > 0 and filedt != datetime(1970, 1, 1) and timediff.days < durationint:
duration = str(timediff.days) + ' D'
# Not cached. Download it.
# Establish a Contract object and the params for the request
# set up connection:
self._reset_data(self._host, self._port, self._client_id)
self.inner(sec_type=sec_type, symbol=symbol, currency=currency, exchange=exchange,
primaryExchange=PrimaryExchange, endtime=endtime, duration=duration, bar_size=bar_size,
what_to_show=what_to_show, use_rth=use_rth)
# Make sure the connection doesn't get disconnected prior the response data return
self.timeout = time() + timeoutsecs
while self._conn.isConnected() and time() < self.timeout:
pass
if time() >= self.timeout:
self.timeout = 0
print ('Timed Out, disconnecting.')
self._conn.disconnect()
if not self._df.empty:
self._df.sort_index(inplace=True)
self._df.to_csv(self.filename)
return self._df
if __name__ == '__main__':
date_format = '%Y%m%d %H:%M:%S'
downloader_kwargs = dict(
data_path='../data',
date_format=date_format,
host='127.0.0.1',
port=4001,
client_id=randrange(99999999)
)
downloader = IBDataCache_FlatFile(**downloader_kwargs)
stock_kwargs = dict(
sec_type='STK',
symbol='AAPL',
currency='USD',
exchange='ISLAND',
PrimaryExchange='NASDAQ',
endtime=datetime(2019, 12, 10, 16, 1).strftime(date_format),
duration='2 D',
bar_size='1 min',
what_to_show='TRADES',
use_rth=1
)
df = downloader.get_dataframe(**stock_kwargs)
print(df)
I've built a handy class, attached, allowing caching of IB data into single flat files. This allows a speedup when pulling long sequences of historical data from IB, as it doesn't download the full data sequence where possible. It also backfills into the cache which allows later backtesting when IB is offline.
This rewrite has the advantage that it uses one cache file per symbol and the period (ie. 1 minute data) specified.
This is a complete rewrite of other's work in a previous thread: ( https://community.backtrader.com/topic/1007/how-to-cached-data-from-ib-and-use-it-later/4 ) That routine created many flat files that if the timeframe were change slightly, the cache would not be used. This also fixes a few bugs, such as reducing timeout delays.
The disadvantages here are that for specifying duration of the backfill (not period), this routine only assumes the duration is specified in days. You can request 2 days of 1 minute data, but you cannot request six hours of one minute data. It will download what it needs of the given days specified. As well, this routine is unnecessarily disk intensive when it does download, rewriting the entire pandas dataframe back to disk.
Please note that this routine does not autofill in gaps in data. It will backfill the duration specified with the exception of what is already cached. If the cached data is missing a week mid series, you could not manually run this routine with that week's end date and week duration specified to fill it in. It solely fills at the end of the cache.
If this helps you, let me know.
B.
from ib.opt import ibConnection, message
from ib.ext.Contract import Contract
from time import time, strftime
from datetime import datetime, timedelta
import pandas as pd
from pathlib import Path
from random import randrange #for random connection ID only
class IBDataCache_FlatFile(object):
def _reset_data(self, host='127.0.0.1', port=4001, client_id=None):
self._df = pd.DataFrame(columns=['Date', 'Open', 'High', 'Low', 'Close', 'Volume', 'OpenInterest'])
self._df.set_index('Date', inplace=True)
self._s = pd.Series()
#Initialize connection as long as it's not already connected:
if (not hasattr(self, '_conn')) or (not self._conn.isConnected()):
self._conn = ibConnection(host, port, client_id)
self._conn.enableLogging()
# Register the response callback function and type of data to be returned
self._conn.register(self._error_handler, message.Error)
self._conn.register(self._historical_data_handler, message.historicalData)
self._conn.register(self._save_order_id, 'NextValidId')
self._conn.register(self._nextValidId_handler, message.nextValidId)
self._conn.connect()
def _save_order_id(self, msg):
self._next_valid_id = msg.orderId
print('save order id', msg.orderId)
def _nextValidId_handler(self, msg):
print("nextValidId_handler: ", msg)
def _error_handler(self, msg):
if not msg:
print('disconnecting', self._conn.disconnect())
elif 'No security definition has been found' in str(msg):
print ('ibcachedata: triggering timout')
self.timeout = time() # trigger timeout
def __init__(self, data_path='/docker_stocks/data', date_format='%Y%m%d %H:%M:%S', host='127.0.0.1', port=4001, client_id=None):
self._data_path = data_path
self._date_format = date_format
self._next_valid_id = 1
self._host = host
self._port = port
self._client_id = client_id
def _historical_data_handler(self, msg):
"""
Define historical data handler for IB - this will populate our pandas data frame
"""
if not 'finished' in str(msg.date):
try:
dateidx = datetime.strptime(msg.date, self._date_format)
self._s = ([msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
except:
#for dates only with no time in str:
dateidx = datetime.strptime(msg.date, "%Y%m%d")
self._s = ([msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
self._df.loc[dateidx] = self._s
else:
print("ibcachedata: historical_data_handler: Received finish")
if not self._df.empty:
self._df.sort_index(inplace=True)
self._df.to_csv(self.filename)
self._conn.disconnect()
def inner(self,
con_Id=0,
sec_type='',
symbol='',
currency='USD',
exchange='',
primaryExchange='',
endtime='',
duration='250 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth='1'):
print ("called inner... setting up req.")
self.req = Contract()
self.req.m_conId = con_Id
self.req.m_secType = sec_type
self.req.m_symbol = symbol
self.req.m_currency = currency
self.req.m_exchange = exchange
self.primaryExch = primaryExchange
self.req.endtime = endtime
self.req.duration = duration
self.req.bar_size = bar_size
self.req.what_to_show = what_to_show
self.req.use_rth = use_rth
self._conn.reqHistoricalData(self._next_valid_id, self.req, endtime, duration,
bar_size, what_to_show, use_rth, 1)
def get_dataframe(self,
sec_type=None,
symbol=None,
currency='USD',
exchange=None,
PrimaryExchange='NASDAQ',
endtime=None,
duration='250 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth='1',
timeoutsecs=300):
date_format = '%Y%m%d %H:%M:%S'
self.timeout = 0
# build filename
self.filename = symbol + '_' + exchange + '_' + \
bar_size.replace(' ', '') + '_' + \
what_to_show + '_' + str(use_rth) + '.csv'
self.filename = self.filename.replace('/', '.')
self.filename = self._data_path + '/' + self.filename
print ("filename: ", self.filename)
# set up connection:
self._reset_data(self._host, self._port, self._client_id)
# check if we have this cached
data_file = Path(self.filename)
if data_file.is_file():
self._df = pd.read_csv(self.filename,
parse_dates=True,
index_col=0)
# get filedt (the last datetime in the file)
if not self._df.empty:
filedt = pd.to_datetime(self._df.index[-1])
else:
filedt = datetime(1970, 1, 1)
if filedt < datetime.strptime(endtime, date_format): # DAYS ONLY
#SET DURATION SHORTER ENDTIME-FILEDT = NEW DURATION TO DOWNLOAD
timediff = datetime.strptime(endtime, date_format) - filedt
timediff = timedelta(timediff.days)
else:
timediff = timedelta(0) #download nothing as the file is up to date
if self._df.empty or timediff > timedelta(0):
durationint = [int(s) for s in duration.split() if s.isdigit()]
durationint = durationint[0]
#test if the date difference to the end of the file exists, if so then scrape the non-cached data into the dataframe
#if the difference to the last bar is larger than the specified duration, just do the user specified duration -- the goal is to save time, not cost time
if timediff.days > 0 and filedt != datetime(1970, 1, 1) and timediff.days < durationint:
duration = str(timediff.days) + ' D'
# Not cached. Download it.
# Establish a Contract object and the params for the request
self.inner(sec_type=sec_type, symbol=symbol, currency=currency, exchange=exchange,
primaryExchange=PrimaryExchange, endtime=endtime, duration=duration, bar_size=bar_size,
what_to_show=what_to_show, use_rth=use_rth)
# Make sure the connection doesn't get disconnected prior the response data return
self.timeout = time() + timeoutsecs
while self._conn.isConnected() and time() < self.timeout:
pass
if time() >= self.timeout:
self.timeout = 0
print ('Timed Out, disconnecting.')
self._conn.disconnect()
if not self._df.empty:
self._df.sort_index(inplace=True)
self._df.to_csv(self.filename)
return self._df
if __name__ == '__main__':
date_format = '%Y%m%d %H:%M:%S'
downloader_kwargs = dict(
data_path='../data',
date_format=date_format,
host='127.0.0.1',
port=4001,
client_id=randrange(99999999)
)
downloader = IBDataCache_FlatFile(**downloader_kwargs)
stock_kwargs = dict(
sec_type='STK',
symbol='AAPL',
currency='USD',
exchange='ISLAND',
PrimaryExchange='NASDAQ',
endtime=datetime(2019, 12, 10, 16, 1).strftime(date_format),
duration='2 D',
bar_size='1 min',
what_to_show='TRADES',
use_rth=1
)
df = downloader.get_dataframe(**stock_kwargs)
print(df)
And a simple test script to run it:
import backtrader as bt
import sys
sys.path.append('/docker_stocks/bbwavelet')
from datetime import datetime
import backtrader.indicators as btind
from bbwaveletlifting import bbhaarlift, bbdaubechieslift
class firstStrategy(bt.Strategy):
def __init__(self):
self.startlevel = 1
self.endlevel = 5
self.dbsig = [None] * 10
self.barcount = bbbarcounter(largedist=500, majordist=100, minordist=10, plotdistance=0.07)
for i in range(self.startlevel, self.endlevel+1):
threshamount = (self.endlevel+1-i)/1000
threshlist = [threshamount] * (self.endlevel)
self.dbsig[i] = bbdaubechieslift(self.data.close, threshall=threshamount, threshtype = "S", donoholev = 3, donflag=True, subplot=True)
#Another example usage: self.dbsig[i] = bbdaubechieslift(self.data.close, threshlist = [0.5, 0.5, 0.5, 0.5, 0.5], threshtype = "S", donoholev = 3, donflag=True, subplot=True)
# Variable for our starting cash
startcash = 10000
# Create an instance of cerebro
cerebro = bt.Cerebro()
# Add our strategy
cerebro.addstrategy(firstStrategy)
# Get Apple data from Yahoo Finance.
data = bt.feeds.YahooFinanceData(
dataname='AAPL',
fromdate=datetime(2017, 4, 1),
todate=datetime(2019, 4, 10),
buffered=True
)
# Add the data to Cerebro
cerebro.adddata(data)
# Set our desired cash start
cerebro.broker.setcash(startcash)
# Run over everything
cerebro.run()
# Get final portfolio Value
portvalue = cerebro.broker.getvalue()
pnl = portvalue - startcash
# Print out the final result
print('Final Portfolio Value: ${}'.format(portvalue))
print('P/L: ${}'.format(pnl))
# Finally plot the end results
cerebro.plot(style='candlestick')
'''
Author: B. Bradford adapted from instructional information at
bearcave.com/Ian Kaplan.
MIT License
Copyright (c) B. Bradford
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
'''
class bbdaubechieslift(bt.indicators.PeriodN):
sqrt3 = math.sqrt(3)
sqrt2 = math.sqrt(2)
plotinfo = dict(subplot=True)
lines = ('dblift', )
plotlines = dict(haarlift=dict(alpha=0.75, ls='-', linewidth=1.0))
params = (('coeffpick', None),
('donoholev', 1),
('donflag', True),
('threshlist', []),
('threshall', None),
('threshtype', "H"), ) #Hard or soft thresholding rules 2**coefflev data points required or 2**coefflev
@staticmethod
def next_power_of_2(x):
return 1 if x == 0 else 2 ** (x.bit_length() -1)
def init(self):
self.addminperiod(2 ** self.params.coefflev + 1)
def once(self, start, end):
self.blnforward = 1
self.blninverse = 2
#create dummy line to manipulate and desired 2** data length (max or selected)
lstinputdata = self.data.array[:] #make a copy of the list/array
#db is padded out in the case of None coefflev rather than truncated
pow2 = self.next_power_of_2(len(lstinputdata)) * 2 # input needs to be power of two data points pad up to next pow2
#Count backwards through line to leave most recent value in place if incremental: -- not provided here
inlist = lstinputdata[:]
#mirror inlist to avoid edge discontinuities -- make sure it's even length for daubechies:
mirlistlen = len(self.data.array)
mirpostlen = int((pow2 - len(lstinputdata))/2)
mirprelen = pow2 - mirlistlen - mirpostlen
mirlist = inlist[:mirprelen][::-1] + inlist[:mirlistlen] + inlist[:-mirpostlen-1:-1]
ftresult = bbdaubechieslift.forwardtrans(self, vec=mirlist) #forward transform
#threshedresult = ftresult
threshedresult = bbdaubechieslift.wavshrink(self, coeffary=ftresult, threshall=self.p.threshall, threshtype=self.p.threshtype, threshlist=self.params.threshlist, donoholev = self.p.donoholev, donflag=self.p.donflag)
#threshedresult = ftresult[:] #if you don't want to threshold/wavshrink/donoho
itresult = bbdaubechieslift.inversetrans(self, vec=threshedresult) #inverse transform
#demirror array:
dummyaryout = itresult[mirprelen:mirlistlen + mirprelen]
lstinputdata = dummyaryout[:] # self.lines[0].array[pow2padding:] + itresult
#Only run once if not doing incremental -- incremental not provided here.
self.lines.dblift.array = lstinputdata #arr.array('d', [lstinputdata[lstinphalf]] * (len(lstinputdata) - lstinphalf)) + lstinputdata[lstinphalf:]
def predict(self, vec, N, direction):
half = N >> 1
if (direction == self.blnforward):
#unnecessarily aggressive floats to avoid casting as int or concatenating strings:
vec[half] = float(vec[half]) - (self.sqrt3/4.0)*float(vec[0]) - (((self.sqrt3-2.0)/4.0)*float(vec[half-1]))
elif (direction == self.blninverse):
vec[half] = float(vec[half]) + (self.sqrt3/4.0)*float(vec[0]) + (((self.sqrt3-2.0)/4.0)*float(vec[half-1]))
else:
debug.print("daubechies:predict: bad direction value")
# predict, forward
for n in range (0, half, 1):
if (direction == self.blnforward):
vec[half+n] = float(vec[half+n]) - (self.sqrt3/4.0)*float(vec[n]) - (((self.sqrt3-2.0)/4.0)*float(vec[n-1]))
elif (direction == self.blninverse):
vec[half+n] = float(vec[half+n]) + (self.sqrt3/4.0)*float(vec[n]) + (((self.sqrt3-2.0)/4.0)*float(vec[n-1]))
else:
debug.print("daubechies: predict: bad direction value")
break
return vec
def updateone(self, vec, N, direction):
half = N >> 1
for n in range(0, half):
updateVal = bbdaubechieslift.sqrt3 * float(vec[half+n])
if (direction == self.blnforward):
vec[n] = float(vec[n]) + updateVal
elif (direction == self.blninverse):
vec[n] = float(vec[n]) - updateVal
else:
debug.print("daubechies:updateone: bad direction value")
break
return vec
def update(self, vec, N, direction):
half = N >> 1
for i in range (0, half-1, 1):
if (direction == self.blnforward):
vec[i] = float(vec[i]) - float(vec[half + i + 1])
elif (direction == self.blninverse):
vec[i] = float(vec[i]) + float(vec[half + i + 1])
else:
debug.print("haar update: bad direction value")
break
if (direction == self.blnforward):
vec[half-1] = float(vec[half-1]) - float(vec[half])
elif (direction == self.blninverse):
vec[half-1] = float(vec[half-1]) + float(vec[half])
return vec
def forwardtrans(self, vec):
#The result of forwardTrans is a set of wavelet coefficients
#ordered by increasing frequency and an approximate average
#of the input data set in vec.
N = len(vec)
# set for counter:
n = N
while n > 1:
# beware, vec is altered in-place
bbdaubechieslift.split(vec, n)
bbdaubechieslift.updateone(self, vec, n, self.blnforward)
bbdaubechieslift.predict(self, vec, n, self.blnforward)
bbdaubechieslift.update(self, vec, n, self.blnforward)
bbdaubechieslift.normalize(self, vec, n, self.blnforward)
n = n >> 1
return vec
def inversetrans(self, vec):
N = len(vec)
n = 2
while n <= N:
# beware, vec is altered in-place
bbdaubechieslift.normalize(self, vec, n, self.blninverse)
bbdaubechieslift.update(self, vec, n, self.blninverse)
bbdaubechieslift.predict(self, vec, n, self.blninverse)
bbdaubechieslift.updateone(self, vec, n, self.blninverse)
bbdaubechieslift.merge(self, vec, n)
n = n << 1
return vec
def calcdonoho(self, coeffary, donoholev, accpos):
n = len(coeffary)
#donoholev -- Get donoho coeffs to threshold from this wavelet level
dcoeffstart = -int(sum(accpos[-donoholev:]))
if -donoholev + 1 == 0:
coefflist = coeffary[dcoeffstart:]
else:
dcoeffend = -int(sum(accpos[-donoholev + 1:]))
coefflist = coeffary[dcoeffstart:dcoeffend]
# noise variance sigma from dh is a numeric scalar representing (an estimate of) the additive Gaussian white noise variance.
# Estimate the noise variance based on the median absolute deviation (MAD) of the scale one wavelet coefficients.
n = len(coefflist)
dh = median(coefflist)
mad = median([abs(x - dh) for x in
coefflist]) # list(map(lambda x: abs(x-dh), coefflist))) #difference from median for 1st scale coeffs
# sigma = list(map(abs, diff))
sigma = mad / 0.6745 # noise standard deviation
donoho = math.sqrt(2 * math.log(n)) * sigma
print ('donoho is: ', donoho)
return donoho
def normalize(self, vec, N, direction):
half = N >> 1
for n in range(0, half):
if direction == self.blnforward:
vec[half + n] = float(vec[half + n]) - (bbdaubechieslift.sqrt3 / 4.0) * float(vec[n]) - (
((bbdaubechieslift.sqrt3 - 2) / 4.0) * float(vec[n - 1]))
elif (direction == self.blninverse):
vec[half + n] = float(vec[half + n]) + (bbdaubechieslift.sqrt3 / 4.0) * float(vec[n]) + (
((bbdaubechieslift.sqrt3 - 2.0) / 4.0) * float(vec[n - 1]))
else:
print("daubechies:normalize: bad direction value")
break
return vec
def wavshrink(self, coeffary, threshlist, threshall, threshtype, donoholev, donflag=True):
accpos = list()
accpos += [math.floor((len(coeffary) - 1) / 2) + 2]
while sum(accpos) <= len(coeffary):
accpos.insert(0, math.floor((accpos[0] - 1) / 2) + 2)
accpos = accpos[1:]
donoho = self.calcdonoho(coeffary, donoholev = self.p.donoholev, accpos=accpos)
if threshlist:
for i in range(len(threshlist)-1,-1,-1):
sumaccpos = -int(sum(accpos[i:]))
sumaccposend = -int(sum(accpos[i + 1:]))
if donflag:
donohomult = donoho * threshlist[i]
if donohomult:
for x in range(sumaccpos, sumaccposend):
print ("x is : ", x)
if threshtype.upper() == "S":
if abs(coeffary[x]) < donohomult:
coeffary[x] = 0
elif coeffary[x] < -donohomult:
coeffary[x] += donohomult
elif coeffary[x] > donohomult:
coeffary[x] -= donohomult
elif threshtype.upper() == "H":
if abs(coeffary[x]) < donohomult:
coeffary[x] = 0
else:
for x in range(sumaccpos, sumaccposend):
coeffary[x] = float(coeffary[x]*threshlist[i])
threshedresult = coeffary[:]
#versus thresholding all coefficients:
if threshall is not None:
if donflag:
threshval = donoho * threshall
else:
threshval = threshall
if threshval:
print("threshval is: ", threshval)
for x in range(len(coeffary)):
if threshtype.upper() == "S":
if abs(coeffary[x]) < threshval:
coeffary[x] = 0
if coeffary[x] < -threshval:
coeffary[x] += threshval
elif coeffary[x] > threshval:
coeffary[x] -= threshval
elif threshtype.upper() == "H":
if abs(coeffary[x]) < threshval:
coeffary[x] = 0
threshedresult = coeffary[:]
return threshedresult
def split(vec, N):
start = 1
end = N - 1
while (start < end):
#Could do this with list comprehension, but want to stick to the original structure:
for i in range(start, end, 2):
tmp = vec[i]
vec[i] = vec[i + 1]
vec[i+1] = tmp
start += 1
end -= 1
return vec
def merge(self, vec, N):
half = N >> 1
start = half-1
end = half
while (start > 0):
for i in range(start, end, 2):
tmp = vec[i]
vec[i] = vec[i+1]
vec[i+1] = tmp
start -= 1
end += 1
return vec
Here's an indicator (bbdaubechieslift) that provides a Daubechies D4 wavelet transform via lifting for deconstructing a line into frequency sub-bands. It provides calculation of the Donoho threshold amount to minimize theoretical noise. Again, as with the other wavelet indicators, my general goal has been to comply with Daniel's R. vision of minimization of libraries. There is a math.floor library reference but this is easily coded around if desired.
Because this routine pads outwards to nearest 2**n bars, rather than getting a slice of the line, an incremental option isn't included. However, this padding out with symmetrical data does reduce edge effects/discontinuities in the Daubechies D4.
Example usage:
x = bbdaubechieslift(self.data.close, threshall=threshamount, threshtype = "S", donoholev = 3, donflag=True, subplot=True)
or, another way to threshold individual subbands:
x = bbdaubechieslift(self.data.close, threshlist = [0.5, 0.5, 0.5, 0.5, 0.5], threshtype = "S", donoholev = 3, donflag=True, subplot=True)
Threshlist and threshall are mutually exclusive.
donoholev specifies the subband to use for calculating the "optimum" theoretical noise filter -- usually the highest detail subband. Normally donoholev=1 however that doesn't always provide for the optimum noise reduction depending on the signal and the time frames of the forecast intended.
Threshall is the threshold multiplier for the calculated Donoho value which blanket thresholds all wavelet coefficients. The threshlist parameter can be supplied to threshold out individual coefficients for sub-bands, ie. threshlist = [0.5, 0.5, 0.5, 0.5, 0.5]. Please understand that in Daubechies D4 there is some overlap in signal between the bands and thresholding some bands and not others can result in erratic results. The Haar wavelet indicator I previously posted is fully orthogonal as an alternative.
It is extremely easy with this indicator to give it nonsensical parameters, for example a threshlist that is longer than the n in the data len of 2**n. Or a Donoho multiplier that is too low or too high. You may wish to experiment by monitoring the print output showing the threshval calculated by calcdonoho routine. A donoholev of 1 (the first level) will result in small noise reductions, whereas a donoho level calculated from higher levels results in much larger thresholding. If this is your first time, do not use hard thresholding (threshtype = "H") as it will be more difficult to form a coherent result without discontinuities.
I will have open hours available for contract work starting at the end of August. My inundated spam mailbox is bigdavediode2 at the email provider yahoo.com. Apologies for any bugs, but should there be some, and there probably are, at least this gives a good starting point for others to improve upon.
Typical output at multiple subband levels:
Hi Amir, sorry for the delay in responding. There is no period as this pulls out highs and lows (peaks and valleys) where the high is highest compared to the previous low, and the subsequent low is lowest compared to the previous high so it starts from the line's beginning. A period would not work especially for situations such as [5,5,5,5,5,5,5] -- with five being ambiguous as to whether it's a high or a low or somewhere in between.
I haven't looked at this one, but perhaps this one will help? I tried to tighten an existing zigzag indicator up:
https://community.backtrader.com/topic/1390/zigzag-indicator
Although this routine is great for its simple flat-file data caching, it does have an implementation problem that can shut down the connection before the downloading of data is completed with its sleep statements. Another issue is that, even for small downloads, the sleep statements hold up processing and when one is trying to do analysis of 100-200 securities/hour these sleep statements have to be removed as below.
Ideally, the callback will just terminate the connection correctly when the "finished" response is received from the api. So I've altered the code below. It's a bit hard to follow due to the callbacks but appears to work effectively. Please note the timeoutsecs argument should be specified as it is the maximum that the routine will wait for the download.
from ib.opt import ibConnection, message
from ib.ext.Contract import Contract
from time import time, strftime
from datetime import datetime
import pandas as pd
class IBDataCache(object):
def _reset_data(self, host='127.0.0.1', port=4001, client_id=None):
self._df = pd.DataFrame(columns=['Date', 'Open', 'High', 'Low', 'Close', 'Volume', 'OpenInterest'])
self._s = pd.Series()
#Initialize connection as long as it's not already connected:
if (not hasattr(self, '_conn')) or (not self._conn.isConnected()):
self._conn = ibConnection(host, port, client_id)
self._conn.enableLogging()
# Register the response callback function and type of data to be returned
self._conn.register(self._error_handler, message.Error)
self._conn.register(self._historical_data_handler, message.historicalData)
self._conn.register(self._save_order_id, 'NextValidId')
self._conn.register(self._nextValidId_handler, message.nextValidId)
self._conn.connect()
def _save_order_id(self, msg):
self._next_valid_id = msg.orderId
print('save order id',msg.orderId)
def _nextValidId_handler(self, msg):
print("nextValidId_handler: ", msg)
self.inner(sec_type=self.req.m_secType, symbol=self.req.m_symbol, currency=self.req.m_currency, exchange=self.req.m_exchange, \
primaryExchange=self.req.m_primaryExch, endtime=self.req.endtime, duration=self.req.duration, \
bar_size=self.req.bar_size, what_to_show=self.req.what_to_show, use_rth=self.req.use_rth)
def _error_handler(self, msg):
print("error: ", msg)
if not msg:
print('disconnecting', self._conn.disconnect())
def __init__(self, data_path='/docker_stocks/data', date_format='%Y%m%d %H:%M:%S', host='127.0.0.1', port=4001, client_id=None):
self._data_path = data_path
self._date_format = date_format
self._next_valid_id = 1
self._host = host
self._port = port
self._client_id = client_id
def _historical_data_handler(self, msg):
"""
Define historical data handler for IB - this will populate our pandas data frame
"""
# print (msg.reqId, msg.date, msg.open, msg.close, msg.high, msg.low)
if not 'finished' in str(msg.date):
#print ("historical_data_handler: ", msg)
try:
self._s = ([datetime.strptime(msg.date, self._date_format),
msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
except:
#for dates only with no time to str:
self._s = ([datetime.strptime(msg.date, "%Y%m%d"),
msg.open, msg.high, msg.low, msg.close, msg.volume, 0])
self._df.loc[len(self._df)] = self._s
else:
self._df.set_index('Date', inplace=True)
def setAllWithKwArgs(self, **kwargs):
#set all attributes to the kwargs to pass along
for key, value in kwargs.items():
setattr(self, key, value)
def inner(self, sec_type, symbol, currency, exchange, primaryExchange, endtime, duration, bar_size, what_to_show, use_rth):
print ("calling inner... setting up req.")
self.req = Contract()
self.req.m_secType = sec_type
self.req.m_symbol = symbol
self.req.m_currency = currency
self.req.m_exchange = exchange
self.primaryExch = primaryExchange
self.req.endtime = endtime
self.req.duration = duration
self.req.bar_size = bar_size
self.req.what_to_show = what_to_show
self.req.use_rth = use_rth
self._conn.reqHistoricalData(self._next_valid_id, self.req, endtime, duration,
bar_size, what_to_show, use_rth, 1)
def get_dataframe(self, sec_type, symbol, currency, exchange, primaryExchange, endtime, duration, bar_size, what_to_show, use_rth, timeoutsecs):
# build filename
self.filename = symbol + '_' + sec_type + '_' + exchange + '_' + currency + '_' + \
endtime.replace(' ', '') + '_' + duration.replace(' ', '') + '_' + bar_size.replace(' ', '') + '_' + \
what_to_show + '_' + str(use_rth) + '.csv'
self.filename = self.filename.replace('/', '.')
self.filename = self._data_path + '/' + self.filename
print ("filename: ", self.filename)
try:
# check if we have this cached
self._df = pd.read_csv(self.filename,
parse_dates=True,
index_col=0)
self._df.index = pd.to_datetime(self._df.index, format='%Y-%m-%d %H:%M:%S')
except IOError:
#set up connection:
self._reset_data(self._host, self._port, self._client_id)
# Not cached. Download it.
# Establish a Contract object and the params for the request
self.inner(sec_type, symbol, currency, exchange,
primaryExchange, endtime, duration, bar_size,
what_to_show, use_rth)
# Make sure the connection doesn't get disconnected prior the response data return
timeout = time() + timeoutsecs
while self._conn.isConnected() and time() < timeout:
#print(".", end="", flush=True)
pass
self._conn.disconnect()
self._df.to_csv(self.filename)
return self._df
if __name__ == '__main__':
date_format = '%Y%m%d %H:%M:%S'
downloader_kwargs = dict(
data_path='../data',
date_format=date_format,
host='127.0.0.1',
port=4001,
client_id=992
)
downloader = IBDataCache(**downloader_kwargs)
stock_kwargs = dict(
sec_type='STK',
symbol='AAPL',
currency='USD',
exchange='SMART',
primaryExchange='NASDAQ',
endtime=datetime(2018, 10, 26, 15, 59).strftime(date_format),
duration='2 D',
bar_size='30 mins',
what_to_show='TRADES',
use_rth=1
)
df = downloader.get_dataframe(**stock_kwargs)
print(df)
stock_kwargs = dict(
sec_type='STK',
symbol='MSFT',
currency='USD',
exchange='SMART',
primaryExchange='NASDAQ',
endtime=datetime(2018, 10, 26, 15, 59).strftime(date_format),
duration='1 D',
bar_size='1 min',
what_to_show='TRADES',
use_rth=1
)
df = downloader.get_dataframe(**stock_kwargs)
print ("IBCacheData")
print(df)
(As a side note, the int64 error mentioned above is due to the sleep statement from the original code terminating before the downloading is completed.)