diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 8e7bfd5d..0552bcc1 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -16,7 +16,8 @@ from catalyst.exchange.bundle_utils import get_start_dt, \ from catalyst.exchange.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \ InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \ - InvalidHistoryFrequencyError, MismatchingFrequencyError + InvalidHistoryFrequencyError, MismatchingFrequencyError, \ + BundleNotFoundError from catalyst.exchange.exchange_execution import ExchangeStopLimitOrder, \ ExchangeLimitOrder, ExchangeStopOrder from catalyst.exchange.exchange_portfolio import ExchangePortfolio @@ -526,6 +527,12 @@ class Exchange: ) reader = self.bundle.get_reader(data_frequency) + if reader is None: + raise BundleNotFoundError( + exchange=self.name, + data_frequency=data_frequency + ) + values = reader.load_raw_arrays( sids=[asset.sid for asset in assets], fields=[field], diff --git a/catalyst/exchange/exchange_bcolz.py b/catalyst/exchange/exchange_bcolz.py index a517b4fe..9fccd9a1 100644 --- a/catalyst/exchange/exchange_bcolz.py +++ b/catalyst/exchange/exchange_bcolz.py @@ -52,7 +52,10 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader): data = [] for field in fields: - out = np.full(shape, np.nan) + if field != 'volume': + out = np.full(shape, np.nan) + else: + out = np.zeros(shape, dtype=np.float64) for i, sid in enumerate(sids): carray = self._open_minute_file(field, sid) diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index f93dfd29..dc38cc7e 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -2,6 +2,7 @@ import os import shutil from datetime import timedelta +import numpy as np import pandas as pd from logbook import Logger, INFO @@ -14,7 +15,8 @@ from catalyst.exchange.bundle_utils import get_ffill_candles, range_in_bundle, \ from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ BcolzExchangeBarWriter from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \ - InvalidHistoryFrequencyError, PricingDataBeforeTradingError + InvalidHistoryFrequencyError, PricingDataBeforeTradingError, \ + TempBundleNotFoundError from catalyst.exchange.exchange_utils import get_exchange_folder from catalyst.utils.cli import maybe_show_progress from catalyst.utils.paths import ensure_directory @@ -70,7 +72,7 @@ class ExchangeBundle: data_frequency=data_frequency ) except IOError: - self.get_readers[path] = None + self._readers[path] = None return self._readers[path] @@ -300,6 +302,9 @@ class ExchangeBundle: else self.calendar.sessions_in_range(start_dt, end_dt) reader = self.get_reader(data_frequency, path=path) + if reader is None: + raise TempBundleNotFoundError(path=path) + arrays = reader.load_raw_arrays( sids=[asset.sid], fields=['open', 'high', 'low', 'close', 'volume'], @@ -310,13 +315,11 @@ class ExchangeBundle: if not arrays: return path - ohlcv = dict( - open=arrays[0].flatten(), - high=arrays[1].flatten(), - low=arrays[2].flatten(), - close=arrays[3].flatten(), - volume=arrays[4].flatten() - ) + ohlcv = dict() + for index, field in enumerate( + ['open', 'high', 'low', 'close', 'volume']): + ohlcv[field] = arrays[index].flatten() + ohlcv[field] = ohlcv[field][~np.isnan(ohlcv[field])] df = pd.DataFrame( data=ohlcv, @@ -400,7 +403,8 @@ class ExchangeBundle: except PricingDataBeforeTradingError: continue - sessions = get_periods_range(asset_start, asset_end, 'daily') + sessions = get_periods_range(asset_start, asset_end, + data_frequency) periods = [] dt = sessions[0] diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py index 3a3b6747..b121acc6 100644 --- a/catalyst/exchange/exchange_errors.py +++ b/catalyst/exchange/exchange_errors.py @@ -171,10 +171,14 @@ class SymbolNotFoundOnExchange(ZiplineError): class BundleNotFoundError(ZiplineError): - msg = ('Unable to find bundle data for exchange {exchange}. ' - 'Please ingest data using the command ' - '`catalyst ingest -b exchange_{exchange}`. ' - 'See catalyst documentation for details.').strip() + msg = ('Unable to find bundle data for exchange {exchange} and ' + 'data frequency {data_frequency}.' + 'Please ingest some price data.' + 'See `catalyst ingest-exchange --help` for details.').strip() + + +class TempBundleNotFoundError(ZiplineError): + msg = ('Temporary bundle not found in: {path}.').strip() class EmptyValuesInBundleError(ZiplineError): @@ -196,5 +200,6 @@ class PricingDataNotLoadedError(ZiplineError): '`catalyst ingest-exchange -x {exchange} -i {symbol_list}`. ' 'See catalyst documentation for details.').strip() + class ApiCandlesError(ZiplineError): msg = ('Unable to fetch candles from the remote API: {error}.').strip() diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index 25d1a6fd..1439b2d6 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -57,22 +57,46 @@ class ExchangeBundleTestCase: pass def test_ingest_daily(self): - exchange_name = 'bitfinex' + # exchange_name = 'bitfinex' + # data_frequency = 'daily' + # include_symbols = 'neo_btc,bch_btc,eth_btc' - start = pd.to_datetime('2017-01-01', utc=True) - end = pd.to_datetime('2017-09-30', utc=True) + exchange_name = 'poloniex' + data_frequency = 'daily' + include_symbols = 'btc_usdt' - exchange_bundle = ExchangeBundle(get_exchange(exchange_name)) + start = pd.to_datetime('2015-01-01', utc=True) + end = pd.to_datetime('2015-12-31', utc=True) + + exchange = get_exchange(exchange_name) + exchange_bundle = ExchangeBundle(exchange) log.info('ingesting exchange bundle {}'.format(exchange_name)) exchange_bundle.ingest( - data_frequency='daily', - include_symbols='neo_btc,bch_btc,eth_btc', + data_frequency=data_frequency, + include_symbols=include_symbols, exclude_symbols=None, start=start, end=end, show_progress=True ) + + symbols = include_symbols.split(',') + assets = [] + for pair_symbol in symbols: + assets.append(exchange.get_asset(pair_symbol)) + + reader = exchange_bundle.get_reader(data_frequency) + for asset in assets: + arrays = reader.load_raw_arrays( + sids=[asset.sid], + fields=['close'], + start_dt=start, + end_dt=end + ) + print('found {} rows for {} ingestion\n{}'.format( + len(arrays[0]), asset.symbol, arrays[0]) + ) pass def test_merge_ctables(self):