From 4649b31d89a221168850c6d5974a93045743e948 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Tue, 17 Oct 2017 18:29:48 -0400 Subject: [PATCH] Fixed issues with daily bundles --- catalyst/examples/simple_loop.py | 11 +- catalyst/exchange/data_portal_exchange.py | 29 +++-- catalyst/exchange/exchange.py | 105 ++--------------- catalyst/exchange/exchange_bundle.py | 136 +++++++++++----------- tests/exchange/test_bundle.py | 6 +- 5 files changed, 108 insertions(+), 179 deletions(-) diff --git a/catalyst/examples/simple_loop.py b/catalyst/examples/simple_loop.py index 84ecc488..51c68426 100644 --- a/catalyst/examples/simple_loop.py +++ b/catalyst/examples/simple_loop.py @@ -7,7 +7,7 @@ from catalyst.api import symbol def initialize(context): print('initializing') - context.asset = symbol('gno_btc') + context.asset = symbol('etc_btc') def handle_data(context, data): @@ -19,18 +19,19 @@ def handle_data(context, data): prices = data.history( context.asset, fields='price', - bar_count=50, - frequency='1m' + bar_count=15, + frequency='1d' ) rsi = talib.RSI(prices.values, timeperiod=14)[-1] print('got rsi: {}'.format(rsi)) + pass run_algorithm( capital_base=250, - start=pd.to_datetime('2017-9-5', utc=True), + start=pd.to_datetime('2017-9-01', utc=True), end=pd.to_datetime('2017-9-30', utc=True), - data_frequency='minute', + data_frequency='daily', initialize=initialize, handle_data=handle_data, analyze=None, diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index fe9acd09..ed2f8ed6 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -299,26 +299,36 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): raise InvalidHistoryFrequencyError(frequency=data_frequency) try: - values = reader.load_raw_arrays( + # values = reader.load_raw_arrays( + # fields=[field], + # start_dt=dts[0], + # end_dt=dts[-1], + # sids=[asset.sid for asset in assets], + # )[0] + values = bundle.get_raw_arrays( + assets=assets, fields=[field], start_dt=dts[0], end_dt=dts[-1], - sids=[asset.sid for asset in assets], + data_frequency=data_frequency )[0] except Exception: + first_trading_day = self._get_first_trading_day(assets) + symbols = [asset.symbol.encode('utf-8') for asset in assets] + + symbol_list = ','.join(symbols) raise PricingDataNotLoadedError( field=field, - first_trading_day=self._get_first_trading_day(assets), + first_trading_day=first_trading_day, exchange=exchange.name, - symbols=[asset.symbol.encode('utf-8') for asset in assets], + symbols=symbols, + symbol_list=symbol_list ) series = dict() for index, asset in enumerate(assets): - asset_values = [] - for value in values: - asset_values.append(value[index]) + asset_values = values[:, index] value_series = pd.Series(asset_values, index=dts) series[asset] = value_series @@ -333,7 +343,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): first_trading_day=first_trading_day, exchange=assets[0].exchange, symbols=[asset.symbol.encode('utf-8') for asset in assets], - dt = dt, + dt=dt, ) def get_exchange_spot_value(self, exchange, assets, field, dt, @@ -358,7 +368,8 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): first_trading_day=self._get_first_trading_day(assets), exchange=exchange.name, symbols=[asset.symbol.encode('utf-8') for asset in assets], - symbol_list = ''.join([asset.symbol.encode('utf-8') for asset in assets]) + symbol_list=''.join( + [asset.symbol.encode('utf-8') for asset in assets]) ) return values diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 84e71aba..8785402f 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -419,98 +419,6 @@ class Exchange: return value - @deprecated - def get_history(self, assets, end_dt, bar_count, data_frequency, - fallback_exchange=True): - """ - 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, - fallback_exchange=fallback_exchange - ) - return candles - - @deprecated - def get_asset_history(self, asset, end, bar_count, data_frequency, - fallback_exchange=True): - """ - 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 = get_start_dt(end, bar_count, data_frequency) - - exchange_start = None - catalyst_end = None - - if asset.end_minute is not None and 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_end = end - exchange_start = start - - 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.symbol, # TODO: use Catalyst symbol - start=catalyst_start, - end=catalyst_end - ) - data += candles - - if exchange_start is not None and fallback_exchange: - candles = self.get_candles( - data_frequency=data_frequency, - assets=[asset], - bar_count=bar_count - ) - data += candles[asset] - - return data - def get_history_window(self, assets, end_dt, @@ -565,10 +473,12 @@ class Exchange: raise InvalidHistoryFrequencyError(frequency) if unit.lower() == 'd': - data_frequency = 'daily' + if data_frequency != 'daily': + raise InvalidHistoryFrequencyError(frequency=frequency) elif unit.lower() == 'm': - data_frequency = 'minute' + if data_frequency != 'minute': + raise InvalidHistoryFrequencyError(frequency=frequency) else: raise InvalidHistoryFrequencyError(frequency) @@ -635,12 +545,12 @@ class Exchange: writer=writer ) - reader = bundle.get_reader(data_frequency) - values = reader.load_raw_arrays( + values = bundle.get_raw_arrays( + assets=assets, fields=[field], start_dt=start_dt, end_dt=end_dt, - sids=[asset.sid for asset in assets], + data_frequency=data_frequency )[0] series = dict() @@ -648,6 +558,7 @@ class Exchange: all_dates = [] asset_values = [] + # TODO: use numpy to avoid the loop date = start_dt for value in values: all_dates.append(date) diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 6da300f7..090775aa 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -78,40 +78,40 @@ class ExchangeBundle: return start, end - def get_reader(self, data_frequency): + def get_reader(self, data_frequency, path=None): """ Get a data writer object, either a new object or from cache :return: BcolzMinuteBarReader or BcolzDailyBarReader """ - if data_frequency in self._readers \ - and self._readers[data_frequency] is not None: - return self._readers[data_frequency] + if path is None: + root = get_exchange_folder(self.exchange.name) + path = BUNDLE_NAME_TEMPLATE.format( + root=root, + frequency=data_frequency + ) - root = get_exchange_folder(self.exchange.name) - input_dir = BUNDLE_NAME_TEMPLATE.format( - root=root, - frequency=data_frequency - ) + if path in self._readers and self._readers[path] is not None: + return self._readers[path] - self._readers[data_frequency] = None + self._readers[path] = None if data_frequency == 'minute': try: - self._readers[data_frequency] = BcolzMinuteBarReader(input_dir) + self._readers[path] = BcolzMinuteBarReader(path) except IOError: - log.debug('no reader data found in {}'.format(input_dir)) + log.debug('no reader data found in {}'.format(path)) elif data_frequency == 'daily': try: - self._readers[data_frequency] = BcolzDailyBarReader(input_dir) + self._readers[path] = BcolzDailyBarReader(path) except IOError: - log.debug('no reader data found in {}'.format(input_dir)) + log.debug('no reader data found in {}'.format(path)) else: raise InvalidHistoryFrequencyError( frequency=data_frequency ) - return self._readers[data_frequency] + return self._readers[path] def update_metadata(self, writer, start_dt, end_dt): pass @@ -122,21 +122,21 @@ class ExchangeBundle: :return: BcolzMinuteBarWriter or BcolzDailyBarWriter """ - key = data_frequency - if key in self._writers: - return self._writers[key] - root = get_exchange_folder(self.exchange.name) - output_dir = BUNDLE_NAME_TEMPLATE.format( + path = BUNDLE_NAME_TEMPLATE.format( root=root, frequency=data_frequency ) - ensure_directory(output_dir) + + if path in self._writers: + return self._writers[path] + + ensure_directory(path) if data_frequency == 'minute': - if len(os.listdir(output_dir)) > 0: + if len(os.listdir(path)) > 0: - metadata = BcolzMinuteBarMetadata.read(output_dir) + metadata = BcolzMinuteBarMetadata.read(path) write_metadata = False if start_dt < metadata.start_session: @@ -152,9 +152,9 @@ class ExchangeBundle: else: end_session = metadata.end_session - self._writers[key] = \ + self._writers[path] = \ BcolzMinuteBarWriter( - output_dir, + path, metadata.calendar, start_session, end_session, @@ -164,8 +164,8 @@ class ExchangeBundle: write_metadata=write_metadata ) else: - self._writers[key] = BcolzMinuteBarWriter( - rootdir=output_dir, + self._writers[path] = BcolzMinuteBarWriter( + rootdir=path, calendar=self.calendar, minutes_per_day=self.minutes_per_day, start_session=start_dt, @@ -176,8 +176,8 @@ class ExchangeBundle: elif data_frequency == 'daily': end_session = end_dt.floor('1d') - self._writers[key] = BcolzDailyBarWriter( - filename=output_dir, + self._writers[path] = BcolzDailyBarWriter( + filename=path, calendar=self.calendar, start_session=start_dt, end_session=end_session @@ -187,7 +187,7 @@ class ExchangeBundle: frequency=data_frequency ) - return self._writers[key] + return self._writers[path] def filter_existing_assets(self, assets, start_dt, end_dt, data_frequency): """ @@ -315,6 +315,30 @@ class ExchangeBundle: return data + def get_raw_arrays(self, assets, start_dt, end_dt, fields, data_frequency, + path=None): + reader = self.get_reader(data_frequency, path) + + if data_frequency == 'minute': + values = reader.load_raw_arrays( + fields=fields, + start_dt=start_dt, + end_dt=end_dt, + sids=[asset.sid for asset in assets], + ) + else: + + # Note that the parameters convention is totally different + # from the minute reader. + values = reader.load_raw_arrays( + columns=fields, + start_date=start_dt, + end_date=end_dt, + assets=assets + ) + + return values + def download_bundle(self, name): """ @@ -347,41 +371,23 @@ class ExchangeBundle: period=period ) - sid = asset.sid - if data_frequency == 'minute': - reader = BcolzMinuteBarReader(path) + # TODO: is this the optimal approach? + # Ensures that we read exact range which we want to write + start_dt = writer._start_session + end_dt = writer._end_session - start = reader.first_trading_day - end = reader.last_available_dt + periods = self.calendar.minutes_in_range(start_dt, end_dt) \ + if data_frequency == 'minute' \ + else self.calendar.sessions_in_range(start_dt, end_dt) - periods = self.calendar.minutes_in_range(start, end) - - arrays = reader.load_raw_arrays( - fields=['open', 'high', 'low', 'close', 'volume'], - start_dt=start, - end_dt=end, - sids=[sid] - ) - - elif data_frequency == 'daily': - reader = BcolzDailyBarReader(path) - - start = writer._start_session - end = writer._end_session - - periods = self.calendar.sessions_in_range(start, end) - - # Note that the parameters convention is totally different - # from the minute reader. - arrays = reader.load_raw_arrays( - columns=['open', 'high', 'low', 'close', 'volume'], - start_date=start, - end_date=end, - assets=[asset] - ) - - else: - raise InvalidHistoryFrequencyError(frequency=data_frequency) + arrays = self.get_raw_arrays( + assets=[asset], + fields=['open', 'high', 'low', 'close', 'volume'], + start_dt=start_dt, + end_dt=end_dt, + data_frequency=data_frequency, + path=path + ) ohlcv = dict( open=arrays[0].flatten(), @@ -442,7 +448,7 @@ class ExchangeBundle: data = [] if not df.empty: df.sort_index(inplace=True) - data.append((sid, df)) + data.append((asset.sid, df)) self._write(data, writer, data_frequency) @@ -573,5 +579,5 @@ class ExchangeBundle: data_frequency=data_frequency, period=chunk['period'], writer=writer, - empty_rows_behavior='ignore' + empty_rows_behavior='strip' ) diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index ea3da2f8..c0110476 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -15,15 +15,15 @@ class ExchangeBundleTestCase: exchange_name = 'poloniex' # start = pd.to_datetime('2017-09-01', utc=True) - start = pd.to_datetime('2017-1-1', utc=True) + start = pd.to_datetime('2017-9-1', utc=True) end = pd.to_datetime('2017-9-30', utc=True) exchange_bundle = ExchangeBundle(get_exchange(exchange_name)) log.info('ingesting exchange bundle {}'.format(exchange_name)) exchange_bundle.ingest( - data_frequency='minute', - include_symbols='gno_btc', + data_frequency='daily', + include_symbols='etc_btc', # include_symbols=None, exclude_symbols=None, start=start,