From 83af12c52c55da1e56eb1195f34b56e6a2d6f22f Mon Sep 17 00:00:00 2001 From: fredfortier Date: Mon, 9 Oct 2017 16:23:02 -0400 Subject: [PATCH] Added exchange get_history method which merge historical bars from the Catalyst and exchange APIs --- catalyst/exchange/bundle_utils.py | 53 +++++++++++---- catalyst/exchange/exchange.py | 97 +++++++++++++++++++++++++++- catalyst/exchange/exchange_bundle.py | 77 ++++++++-------------- 3 files changed, 162 insertions(+), 65 deletions(-) diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 9dd1e436..5078f52d 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/catalyst/exchange/bundle_utils.py @@ -6,6 +6,7 @@ import pandas as pd import pytz from catalyst.data.bundles import from_bundle_ingest_dirname +from catalyst.utils.deprecate import deprecated from catalyst.utils.paths import data_path log = Logger('test_exchange_bundle') @@ -106,6 +107,46 @@ def get_history(exchange_name, data_frequency, symbol, start=None, end=None): return data +def get_ffill_candles(candles, start_dt, end_dt, data_frequency, + previous_candle=None): + """ + Create candles for each period of the specified range, forward-filling + missing candles with the previous value. + + :param candles: + :param start_dt: + :param end_dt: + :param data_frequency: + :param previous_candle: + + :return: + """ + all_dates = [] + all_candles = [] + date = start_dt + + while date <= end_dt: + candle = next(( + candle for candle in candles if candle['last_traded'] == date + ), previous_candle) + + if candle is not None: + all_dates.append(date) + all_candles.append(candle) + + previous_candle = candle + + if data_frequency == 'minute': + date += datetime.timedelta(minutes=1) + elif data_frequency == 'daily': + date += datetime.timedelta(days=1) + else: + raise ValueError('invalid data frequency') + + return all_dates, all_candles + + +@deprecated def get_history_mock(exchange_name, data_frequency, symbol, start_ms, end_ms, exchanges): """ @@ -174,18 +215,6 @@ def get_history_mock(exchange_name, data_frequency, symbol, start_ms, end_ms, return ohlcv -def fetch_candles_chunk(exchange, assets, data_frequency, end_dt, bar_count): - calc_start_dt = end_dt - datetime.timedelta(minutes=bar_count) - candles = exchange.get_candles( - data_frequency=data_frequency, - assets=assets, - bar_count=bar_count, - start_dt=calc_start_dt, - end_dt=end_dt - ) - return candles - - def find_most_recent_time(bundle_name): """ Find most recent "time folder" for a given bundle. diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 471215d1..6bf6ee05 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -10,6 +10,7 @@ from catalyst.assets._assets import TradingPair from logbook import Logger from catalyst.data.data_portal import BASE_FIELDS +from catalyst.exchange import bundle_utils from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \ InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange from catalyst.exchange.exchange_execution import ExchangeStopLimitOrder, \ @@ -412,6 +413,95 @@ class Exchange: return value + def get_history(self, assets, end_dt, bar_count, data_frequency): + """ + Retrieve OHLCV bars from the Catalyst and/or exchange API. + + If Catalyst does not have the full data set, retrieve the missing + portion from the exchange API if the exchanges supports + historical data. + + :param assets: list[TradingPair] + The TradingPair asset. + :param data_frequency: str + The bar frequency: daily or minute + :param bar_count: int + The number of bars desired. + :param end: datetime + The last trading date of the last bar. + + :return: + """ + candles = dict() + for asset in assets: + candles[asset] = self.get_asset_history( + asset=asset, + end=end_dt, + bar_count=bar_count, + data_frequency=data_frequency + ) + return candles + + def get_asset_history(self, asset, end, bar_count, data_frequency): + """ + Retrieve the OHLVC bars of a single asset. + + :param asset: TradingPair + The TradingPair asset. + :param data_frequency: str + The bar frequency: daily or minute + :param bar_count: int + The number of bars desired. + :param end: datetime + The last trading date of the last bar. + :return: + """ + start = end - timedelta(minutes=bar_count) + + exchange_start = None + catalyst_end = None + + if start < asset.end_minute: + catalyst_start = start + if end <= asset.end_minute: + catalyst_end = end + else: + catalyst_end = asset.end_minute + + delta = timedelta(minutes=1) \ + if data_frequency == 'minute' else timedelta(days=1) + exchange_start = catalyst_end + delta + + exchange_end = end + + else: + exchange_start = start + exchange_end = end + + data = [] + if catalyst_end is not None: + # TODO: support multiple assets in the Catalyst API. + candles = bundle_utils.get_history( + exchange_name=self.name, + data_frequency=data_frequency, + symbol=asset.exchange_symbol, # TODO: use Catalyst symbol + start=catalyst_start, + end=catalyst_end + ) + data += candles + + if exchange_start is not None: + candles = self.get_candles( + data_frequency=data_frequency, + assets=[asset], + bar_count=bar_count, + start_dt=exchange_start, + end_dt=exchange_end + ) + data += candles[asset] + + return data + def get_history_window(self, assets, end_dt, @@ -455,11 +545,12 @@ class Exchange: A dataframe containing the requested data. """ - candles = self.get_candles( - data_frequency=frequency, + # TODO: try to read from bundle first + candles = self.get_history( assets=assets, + end_dt=end_dt, bar_count=bar_count, - end_dt=end_dt + data_frequency=data_frequency ) series = dict() diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 0b4d8fdd..0e6e30af 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -10,8 +10,7 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \ BcolzMinuteBarWriter, BcolzMinuteBarReader from catalyst.data.us_equity_pricing import BcolzDailyBarWriter, \ BcolzDailyBarReader -from catalyst.exchange.bundle_utils import fetch_candles_chunk, get_history, \ - get_seconds_from_date +from catalyst.exchange.bundle_utils import get_ffill_candles from catalyst.exchange.exchange_utils import get_exchange_folder from catalyst.exchange.init_utils import get_exchange from catalyst.utils.cli import maybe_show_progress @@ -196,6 +195,16 @@ class ExchangeBundle: def ingest_chunk(self, chunk, previous_candle, data_frequency, assets, writer): + """ + Retrieve the specified OHLCV chunk and write it to the bundle + + :param chunk: + :param previous_candle: + :param data_frequency: + :param assets: + :param writer: + :return: + """ chunk_end = chunk['end'] chunk_start = chunk_end - timedelta(minutes=chunk['bar_count']) @@ -215,36 +224,12 @@ class ExchangeBundle: log.debug('the data chunk already exists') return - candles = dict() - for asset in missing_assets: - if chunk_start < asset.end_minute: - # TODO: fetch delta candles from exchanges - history_end = chunk_end \ - if chunk_end <= asset.end_minute else asset.end_minute - - # TODO: switch to Catalyst symbol convention - candles[asset] = get_history( - exchange_name=self.exchange.name, - data_frequency=data_frequency, - symbol=asset.exchange_symbol, - start=chunk_start, - end=history_end - ) - else: - log.debug( - 'no data in Catalyst api for chunk ' - '{} to {}'.format(chunk_start, chunk_end) - ) - # if data_frequency == 'minute': - # # TODO: ensure correct behavior for assets starting in the chunk - # candles = fetch_candles_chunk( - # exchange=self.exchange, - # assets=missing_assets, - # data_frequency=data_frequency, - # end_dt=chunk_end, - # bar_count=chunk['bar_count'] - # ) - # else: + candles = self.exchange.get_history( + assets=missing_assets, + end_dt=chunk_end, + bar_count=chunk['bar_count'], + data_frequency=data_frequency + ) num_candles = 0 data = [] @@ -260,25 +245,17 @@ class ExchangeBundle: ) continue - all_dates = [] - all_candles = [] - date = chunk_start - while date <= chunk_end: + previous = previous_candle[asset] \ + if asset in previous_candle else None - previous = previous_candle[asset] \ - if asset in previous_candle else None - - candle = next((candle for candle in asset_candles \ - if candle['last_traded'] == date), - previous) - - if candle is not None: - all_dates.append(date) - all_candles.append(candle) - - previous_candle[asset] = candle - - date += timedelta(minutes=1) + all_dates, all_candles = get_ffill_candles( + candles=asset_candles, + start_dt=chunk_start, + end_dt=chunk_end, + data_frequency=data_frequency, + previous_candle=previous + ) + previous_candle[asset] = all_candles[-1] df = pd.DataFrame(all_candles, index=all_dates) if not df.empty: