Fixed misc issues with the bundle refactoring

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],

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)

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]

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()

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):