For others who may have a similar need to use history databases combined with the standard usstock-minute zipline bundle, I've put together the following utility method which automatically uses the correct extended hours history database depending on whether or not adjustments exist for the sids within the requested time frame.
In short, I use custom history database from IBKR to add pre-market and post-market prices to the intraday usstock-minute bundle. Together, I'm able to get full prices for the us stock market. While Zipline bundles apply adjustments on the fly at query time, history databases do not, and so this helps add that functionality to my extended hours history databases.
Brian, if there's a better way to deal with this, def let me know as this isn't an ideal solution, but it works.
# Example Use
from codeload.research.utils import get_extended_prices
extended_prices = get_extended_prices(
'usstock-minute',
sids='FIBBG000BBJQV0',
start_date='2021-07-18',
end_date='2021-07-20',
times=['09:25:00', '09:30:00'],
fields=['Open', 'High', 'Low', 'Close', 'Volume']
)
extended_prices
def get_extended_prices(code:str, universes=None, sids=None,
fields=None, start_date=None, end_date=None, **kwargs):
"""
Get regular and extended hours price history from
intraday zipline bundle and extended hours IBKR
history databases; adjustments are applied on
the fly.
"""
# Setup params
if start_date == None and end_date == None:
raise Exception('Start and end dates are required.')
# Figure out which backup database to use via splits lookup
backup = 'usstock-minute-extended-unadjusted' # default to unadjusted
securities = get_securities(universes=universes, sids=sids,
fields=['Sid', 'Symbol']).reset_index()
if has_adjustments(securities['Sid'].to_list(), start_date, end_date):
backup = 'usstock-minute-extended'
# Get prices
return get_prices(
codes=[code, backup],
universes=universes,
sids=sids,
fields=fields,
start_date=start_date,
end_date=end_date,
**kwargs
)
def has_adjustments(sids:List[str], start_date, end_date,
code:str='usstock-daily'):
"""
Check if given securities have adjustments in date range.
"""
adjustments_applied = False
# Setup dates
sid_list = ','.join(f"'{sid}'" for sid in sids)
start_date = pd.to_datetime(start_date).date()
end_date = pd.to_datetime(end_date).date()
# Get path to year's daily price history db
db_root = f'{settings.DATA_PATH}/quantrocket.v2.history.{code}.sqlite'
db_path = f'{db_root}/quantrocket.v2.history.{code}.sqlite'
# Open sqlite database connection
with closing(sqlite3.connect(db_path)) as connection:
with closing(connection.cursor()) as db:
try:
# Query sid via symbol from price history
rows = db.execute(
f"SELECT Sid, ExDate FROM Split "
f"WHERE Sid IN ({sid_list}) AND "
f"ExDate BETWEEN '{start_date}' AND '{end_date}' "
f"LIMIT 5"
).fetchall()
if len(rows) > 0:
# If splits exist in date range, for provided sids,
# then adjustments will be applied
adjustments_applied = True
except Exception as e:
print(f'An error occurred while querying adjustments. '
f'Error: {e}')
# Return results
return adjustments_applied
