Merge branch 'develop' - Release 0.3.3

catalyst/__main__.py

@@ -38,7 +38,7 @@ except NameError:
     '--default-extension/--no-default-extension',
     is_flag=True,
     default=True,
-    help="Don't load the default catalyst extension.py file in $ZIPLINE_HOME.",
+    help="Don't load the default catalyst extension.py file in $CATALYST_HOME.",
 )
 @click.version_option()
 def main(extension, strict_extensions, default_extension):
@@ -495,6 +495,10 @@ def ingest_exchange(exchange_name, data_frequency, start, end,
     """
     Ingest data for the given exchange.
     """
+
+    if exchange_name is None:
+        ctx.fail("must specify an exchange name '-x'")
+
     exchange = get_exchange(exchange_name)
     exchange_bundle = ExchangeBundle(exchange)
 

catalyst/examples/simple_loop.py

@@ -7,7 +7,7 @@ from catalyst.api import symbol
 
 def initialize(context):
     print('initializing')
-    context.asset = symbol('btc_usd')
+    context.asset = symbol('burst_btc')
 
 
 def handle_data(context, data):
@@ -20,7 +20,7 @@ def handle_data(context, data):
         context.asset,
         fields='price',
         bar_count=15,
-        frequency='1m'
+        frequency='1d'
     )
     rsi = talib.RSI(prices.values, timeperiod=14)[-1]
     print('got rsi: {}'.format(rsi))
@@ -35,7 +35,7 @@ run_algorithm(
     initialize=initialize,
     handle_data=handle_data,
     analyze=None,
-    exchange_name='bitfinex',
+    exchange_name='poloniex',
     algo_namespace='simple_loop',
     base_currency='btc'
 )

catalyst/exchange/bitfinex/bitfinex.py

@@ -23,7 +23,7 @@ from catalyst.exchange.exchange_errors import (
 from catalyst.exchange.exchange_execution import ExchangeLimitOrder, \
     ExchangeStopLimitOrder, ExchangeStopOrder
 from catalyst.exchange.exchange_utils import get_exchange_symbols_filename, \
-    download_exchange_symbols
+    download_exchange_symbols, get_symbols_string
 from catalyst.finance.order import Order, ORDER_STATUS
 from catalyst.protocol import Account
 
@@ -255,6 +255,16 @@ class Bitfinex(Exchange):
         '1m', '5m', '15m', '30m', '1h', '3h', '6h', '12h', '1D', '7D', '14D',
          '1M'
         """
+        log.debug(
+            'retrieving {bars} {freq} candles on {exchange} from '
+            '{end_dt} for markets {symbols}, '.format(
+                bars=bar_count,
+                freq=data_frequency,
+                exchange=self.name,
+                end_dt=end_dt,
+                symbols=get_symbols_string(assets)
+            )
+        )
 
         freq_match = re.match(r'([0-9].*)(m|h|d)', data_frequency, re.M | re.I)
         if freq_match:

catalyst/exchange/bittrex/bittrex.py

@@ -1,6 +1,7 @@
 import json
 
 import pandas as pd
+import time
 from catalyst.assets._assets import TradingPair
 from logbook import Logger
 from six.moves import urllib
@@ -13,10 +14,12 @@ from catalyst.exchange.exchange_errors import InvalidHistoryFrequencyError, \
     ExchangeRequestError, InvalidOrderStyle, OrderNotFound, OrderCancelError, \
     CreateOrderError
 from catalyst.exchange.exchange_utils import get_exchange_symbols_filename, \
-    download_exchange_symbols
+    download_exchange_symbols, get_symbols_string
 from catalyst.finance.execution import LimitOrder, StopLimitOrder
 from catalyst.finance.order import Order, ORDER_STATUS
 
+# TODO: consider using this: https://github.com/mondeja/bittrex_v2
+
 log = Logger('Bittrex', level=LOG_LEVEL)
 
 URL2 = 'https://bittrex.com/Api/v2.0'
@@ -217,10 +220,27 @@ class Bittrex(Exchange):
         :param data_frequency:
         :param assets:
         :param bar_count:
+        :param start_dt
+        :param end_dt
         :return:
         """
-        log.info('retrieving candles')
 
+        # TODO: this has no effect at the moment
+        if end_dt is None:
+            end_dt = pd.Timestamp.utcnow()
+
+        log.debug(
+            'retrieving {bars} {freq} candles on {exchange} from '
+            '{end_dt} for markets {symbols}, '.format(
+                bars=bar_count,
+                freq=data_frequency,
+                exchange=self.name,
+                end_dt=end_dt,
+                symbols=get_symbols_string(assets)
+            )
+        )
+
+        data_frequency = data_frequency.lower()
         if data_frequency == 'minute' or data_frequency == '1m':
             frequency = 'oneMin'
         elif data_frequency == '5m':
@@ -229,7 +249,7 @@ class Bittrex(Exchange):
             frequency = 'thirtyMin'
         elif data_frequency == '1h':
             frequency = 'hour'
-        elif data_frequency == 'daily' or data_frequency == '1D':
+        elif data_frequency == 'daily' or data_frequency == '1d':
             frequency = 'day'
         else:
             raise InvalidHistoryFrequencyError(
@@ -238,13 +258,14 @@ class Bittrex(Exchange):
 
         # Making sure that assets are iterable
         asset_list = [assets] if isinstance(assets, TradingPair) else assets
-        ohlc_map = dict()
         for asset in asset_list:
+            end = int(time.mktime(end_dt.timetuple()))
             url = '{url}/pub/market/GetTicks?marketName={symbol}' \
-                  '&tickInterval={frequency}&_=1499127220008'.format(
+                  '&tickInterval={frequency}&_={end}'.format(
                 url=URL2,
                 symbol=self.get_symbol(asset),
-                frequency=frequency
+                frequency=frequency,
+                end=end
             )
 
             try:
@@ -272,6 +293,7 @@ class Bittrex(Exchange):
                 return ohlc
 
             ordered_candles = list(reversed(candles))
+            ohlc_map = dict()
             if bar_count is None:
                 ohlc_map[asset] = ohlc_from_candle(ordered_candles[0])
             else:

catalyst/exchange/exchange.py

@@ -373,7 +373,7 @@ class Exchange:
         return value
 
     def get_series_from_candles(self, candles, start_dt, end_dt,
-                                field, previous_value=None):
+                                data_frequency, field, previous_value=None):
         """
         Get a series of field data for the specified candles.
 
@@ -388,9 +388,12 @@ class Exchange:
         dates = [candle['last_traded'] for candle in candles]
         values = [candle[field] for candle in candles]
 
-        periods = pd.date_range(start_dt, end_dt)
+        periods = self.bundle.get_calendar_periods_range(
+            start_dt, end_dt, data_frequency
+        )
         series = pd.Series(values, index=dates)
 
+        #TODO: ensure that this working as expected, if not use fillna
         series.reindex(periods, method='ffill', fill_value=previous_value)
 
         return series
@@ -487,6 +490,7 @@ class Exchange:
                     data_frequency=data_frequency,
                     assets=asset,
                     bar_count=trailing_bar_count,
+                    start_dt=start_dt,
                     end_dt=end_dt
                 )
 
@@ -497,6 +501,7 @@ class Exchange:
                     candles=candles,
                     start_dt=trailing_dt,
                     end_dt=end_dt,
+                    data_frequency=data_frequency,
                     field=field,
                     previous_value=last_value
                 )
@@ -784,13 +789,14 @@ class Exchange:
         pass
 
     @abc.abstractmethod
-    def get_orderbook(self, asset, order_type):
+    def get_orderbook(self, asset, order_type, limit):
         """
         Retrieve the the orderbook for the given trading pair.
 
         :param asset: TradingPair
         :param order_type: str
             The type of orders: bid, ask or all
+        :param limit
 
         :return:
         """

catalyst/exchange/exchange_bcolz.py

@@ -16,7 +16,7 @@ class BcolzExchangeBarWriter(BcolzMinuteBarWriter):
             end_session = end_session.floor('1d')
 
         minutes_per_day = 1440 if self._data_frequency == 'minute' else 1
-        default_ohlc_ratio = kwargs.pop('default_ohlc_ratio', 1000000)
+        default_ohlc_ratio = kwargs.pop('default_ohlc_ratio', 100000000)
         calendar = get_calendar('OPEN')
 
         super(BcolzExchangeBarWriter, self) \
@@ -79,8 +79,9 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader):
                 if mask is None:
                     mask = a != 0
 
+                inverse_ratio = self._ohlc_ratio_inverse_for_sid(sid)
                 out[:len(mask), i][mask] = (
-                    a[mask] * self._ohlc_ratio_inverse_for_sid(sid)
+                    a[mask] * inverse_ratio
                 )
 
             if field in fields:

catalyst/exchange/exchange_bundle.py

@@ -11,7 +11,7 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
     BcolzMinuteBarMetadata
 from catalyst.exchange.bundle_utils import range_in_bundle, \
     get_bcolz_chunk, get_delta, get_month_start_end, \
-    get_year_start_end, get_periods_range, get_df_from_arrays, get_start_dt
+    get_year_start_end, get_df_from_arrays, get_start_dt
 from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
     BcolzExchangeBarWriter
 from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
@@ -194,6 +194,71 @@ class ExchangeBundle:
             if data_frequency == 'minute' \
             else self.calendar.sessions_in_range(start_dt, end_dt)
 
+    def ingest_df(self, ohlcv_df, data_frequency, asset, writer,
+                  empty_rows_behavior='strip'):
+        """
+        Ingest a DataFrame of OHLCV data for a given market.
+
+        :param ohlcv_df:
+        :param data_frequency:
+        :param asset:
+        :param writer:
+        :param path:
+        :param empty_rows_behavior:
+        :return:
+        """
+        if empty_rows_behavior is not 'ignore':
+            nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index
+
+            if len(nan_rows) > 0:
+                dates = []
+                previous_date = None
+                for row_date in nan_rows.values:
+                    row_date = pd.to_datetime(row_date)
+
+                    if previous_date is None:
+                        dates.append(row_date)
+
+                    else:
+                        seq_date = previous_date + get_delta(1, data_frequency)
+
+                        if row_date > seq_date:
+                            dates.append(previous_date)
+                            dates.append(row_date)
+
+                    previous_date = row_date
+
+                dates.append(pd.to_datetime(nan_rows.values[-1]))
+
+                name = '{} from {} to {}'.format(
+                    asset.symbol, ohlcv_df.index[0], ohlcv_df.index[-1]
+                )
+                if empty_rows_behavior == 'warn':
+                    log.warn(
+                        '\n{name} with end minute {end_minute} has empty rows '
+                        'in ranges: {dates}'.format(
+                            name=name,
+                            end_minute=asset.end_minute,
+                            dates=dates
+                        )
+                    )
+
+                elif empty_rows_behavior == 'raise':
+                    raise EmptyValuesInBundleError(
+                        name=name,
+                        end_minute=asset.end_minute,
+                        dates=dates
+                    )
+                else:
+                    ohlcv_df.dropna(inplace=True)
+
+        data = []
+        if not ohlcv_df.empty:
+            ohlcv_df.sort_index(inplace=True)
+            data.append((asset.sid, ohlcv_df))
+
+        self._write(data, writer, data_frequency)
+
     def ingest_ctable(self, asset, data_frequency, period, start_dt, end_dt,
                       writer, empty_rows_behavior='strip', cleanup=False):
         """
@@ -242,62 +307,19 @@ class ExchangeBundle:
         periods = self.get_calendar_periods_range(
             start_dt, end_dt, data_frequency
         )
-
         df = get_df_from_arrays(arrays, periods)
-
-        if empty_rows_behavior is not 'ignore':
-            nan_rows = df[df.isnull().T.any().T].index
-
-            if len(nan_rows) > 0:
-                dates = []
-                previous_date = None
-                for row_date in nan_rows.values:
-                    row_date = pd.to_datetime(row_date)
-
-                    if previous_date is None:
-                        dates.append(row_date)
-
-                    else:
-                        seq_date = previous_date + get_delta(1, data_frequency)
-
-                        if row_date > seq_date:
-                            dates.append(previous_date)
-                            dates.append(row_date)
-
-                    previous_date = row_date
-
-                dates.append(pd.to_datetime(nan_rows.values[-1]))
-
-                name = path.split('/')[-1]
-                if empty_rows_behavior == 'warn':
-                    log.warn(
-                        '\n{name} with end minute {end_minute} has empty rows '
-                        'in ranges: {dates}'.format(
-                            name=name,
-                            end_minute=asset.end_minute,
-                            dates=dates
-                        )
-                    )
-
-                elif empty_rows_behavior == 'raise':
-                    raise EmptyValuesInBundleError(
-                        name=name,
-                        end_minute=asset.end_minute,
-                        dates=dates
-                    )
-                else:
-                    df.dropna(inplace=True)
-
-        data = []
-        if not df.empty:
-            df.sort_index(inplace=True)
-            data.append((asset.sid, df))
-
-        self._write(data, writer, data_frequency)
+        self.ingest_df(
+            ohlcv_df=df,
+            data_frequency=data_frequency,
+            asset=asset,
+            writer=writer,
+            empty_rows_behavior=empty_rows_behavior
+        )
 
         if cleanup:
-            log.debug('removing bundle folder following '
-                      'ingestion: {}'.format(path))
+            log.debug(
+                'removing bundle folder following ingestion: {}'.format(path)
+            )
             shutil.rmtree(path)
 
         return path
@@ -315,15 +337,22 @@ class ExchangeBundle:
         earliest_trade = None
         last_entry = None
         for asset in assets:
-            if (earliest_trade is None or earliest_trade > asset.start_date) \
-                    and asset.start_date >= self.calendar.first_session:
-                earliest_trade = asset.start_date
+            if earliest_trade is None or earliest_trade > asset.start_date:
+                if asset.start_date >= self.calendar.first_session:
+                    earliest_trade = asset.start_date
+
+                else:
+                    earliest_trade = self.calendar.first_session
 
             end_asset = asset.end_minute if data_frequency == 'minute' else \
                 asset.end_daily
-            if end_asset is not None and \
-                    (last_entry is None or end_asset > last_entry):
-                last_entry = end_asset
+            if end_asset is not None:
+                if last_entry is None or end_asset > last_entry:
+                    last_entry = end_asset
+
+            else:
+                end = None
+                last_entry = None
 
         if start is None or \
                 (earliest_trade is not None and earliest_trade > start):
@@ -363,8 +392,9 @@ class ExchangeBundle:
                     start_dt, end_dt, [asset], data_frequency
                 )
 
-            except NoDataAvailableOnExchange:
+            except NoDataAvailableOnExchange as e:
                 # If not, we continue to the next asset
+                log.debug('skipping {}: {}'.format(asset.symbol, e))
                 continue
 
             # This is either the first trading day of the asset or the
@@ -391,11 +421,6 @@ class ExchangeBundle:
                     if data_frequency == 'minute':
                         period_start, period_end = get_month_start_end(dt)
 
-                        # TODO: redundant gate, we are already filtering dates
-                        if first_trading_dt > period_start:
-                            dt += timedelta(days=1)
-                            continue
-
                         asset_start_month, _ = get_month_start_end(
                             first_trading_dt
                         )
@@ -414,11 +439,6 @@ class ExchangeBundle:
                     elif data_frequency == 'daily':
                         period_start, period_end = get_year_start_end(dt)
 
-                        # TODO: redundant gate, we are already filtering dates
-                        if first_trading_dt > period_start:
-                            dt += timedelta(days=1)
-                            continue
-
                         asset_start_year, _ = get_year_start_end(
                             first_trading_dt
                         )
@@ -427,11 +447,11 @@ class ExchangeBundle:
                             period_start = first_trading_dt
 
                         _, asset_end_year = get_year_start_end(
-                            asset.end_minute
+                            asset.end_daily
                         )
                         if asset_end_year == period_end \
-                                and period_end > asset.end_minute:
-                            period_end = asset.end_minute
+                                and period_end > asset.end_daily:
+                            period_end = asset.end_daily
 
                     else:
                         raise InvalidHistoryFrequencyError(

catalyst/exchange/exchange_utils.py

@@ -1,6 +1,8 @@
 import json
 import os
 import pickle
+
+from catalyst.assets._assets import TradingPair
 from six.moves.urllib import request
 from datetime import date, datetime
 
@@ -57,6 +59,11 @@ def get_exchange_symbols(exchange_name, environ=None):
         )
 
 
+def get_symbols_string(assets):
+    array = [assets] if isinstance(assets, TradingPair) else assets
+    return ', '.join([asset.symbol for asset in array])
+
+
 def get_exchange_auth(exchange_name, environ=None):
     exchange_folder = get_exchange_folder(exchange_name, environ)
     filename = os.path.join(exchange_folder, 'auth.json')

catalyst/exchange/poloniex/poloniex.py

@@ -22,7 +22,7 @@ from catalyst.exchange.exchange_errors import (
 from catalyst.exchange.exchange_execution import ExchangeLimitOrder, \
     ExchangeStopLimitOrder
 from catalyst.exchange.exchange_utils import get_exchange_symbols_filename, \
-    download_exchange_symbols
+    download_exchange_symbols, get_symbols_string
 from catalyst.exchange.poloniex.poloniex_api import Poloniex_api
 from catalyst.finance.order import Order, ORDER_STATUS
 from catalyst.finance.transaction import Transaction
@@ -186,22 +186,35 @@ class Poloniex(Exchange):
         '5m', '15m', '30m', '2h', '4h', '1D'
         """
 
-        # TODO: implement end_dt and start_dt filters
+        if end_dt is None:
+            end_dt = pd.Timestamp.utcnow()
 
-        if (
-                        data_frequency == '5m' or data_frequency == 'minute'):  # TODO: Polo does not have '1m'
+        log.debug(
+            'retrieving {bars} {freq} candles on {exchange} from '
+            '{end_dt} for markets {symbols}, '.format(
+                bars=bar_count,
+                freq=data_frequency,
+                exchange=self.name,
+                end_dt=end_dt,
+                symbols=get_symbols_string(assets)
+            )
+        )
+
+        if data_frequency == '5m':
             frequency = 300
-        elif (data_frequency == '15m'):
+        elif data_frequency == '15m':
             frequency = 900
-        elif (data_frequency == '30m'):
+        elif data_frequency == '30m':
             frequency = 1800
-        elif (data_frequency == '2h'):
+        elif data_frequency == '2h':
             frequency = 7200
-        elif (data_frequency == '4h'):
+        elif data_frequency == '4h':
             frequency = 14400
-        elif (data_frequency == '1D' or data_frequency == 'daily'):
+        elif data_frequency == '1D' or data_frequency == 'daily':
             frequency = 86400
         else:
+            # Poloniex does not offer 1m data candles
+            # It is likely to error out there frequently
             raise InvalidHistoryFrequencyError(
                 frequency=data_frequency
             )
@@ -212,8 +225,8 @@ class Poloniex(Exchange):
 
         for asset in asset_list:
 
-            end = int(time.time())
-            if (bar_count is None):
+            end = int(time.mktime(end_dt.timetuple()))
+            if bar_count is None:
                 start = end - 2 * frequency
             else:
                 start = end - bar_count * frequency

catalyst/exchange/stats_utils.py

@@ -49,3 +49,12 @@ def get_pretty_stats(stats_df, recorded_cols=None, num_rows=10):
         columns=columns,
         formatters=formatters
     )
+
+
+def df_to_string(df):
+    pd.set_option('display.expand_frame_repr', False)
+    pd.set_option('precision', 8)
+    pd.set_option('display.width', 1000)
+    pd.set_option('display.max_colwidth', 1000)
+
+    return df.to_string()

tests/exchange/test_bittrex.py

@@ -1,3 +1,4 @@
+import pandas as pd
 from catalyst.exchange.bittrex.bittrex import Bittrex
 from catalyst.finance.order import Order
 from base import BaseExchangeTestCase
@@ -7,15 +8,15 @@ from catalyst.exchange.exchange_utils import get_exchange_auth
 log = Logger('test_bittrex')
 
 
-class TestBittrexTestCase(BaseExchangeTestCase):
+class TestBittrex(BaseExchangeTestCase):
     @classmethod
     def setup(self):
-        print ('creating bittrex object')
         auth = get_exchange_auth('bittrex')
         self.exchange = Bittrex(
             key=auth['key'],
             secret=auth['secret'],
-            base_currency='btc'
+            base_currency=None,
+            portfolio=None
         )
 
     def test_order(self):
@@ -52,15 +53,18 @@ class TestBittrexTestCase(BaseExchangeTestCase):
         log.info('retrieving candles')
         ohlcv_neo = self.exchange.get_candles(
             data_frequency='5m',
-            assets=self.exchange.get_asset('neo_btc')
+            assets=self.exchange.get_asset('neo_btc'),
+            bar_count=20,
+            end_dt=pd.to_datetime('2017-10-20', utc=True)
         )
         ohlcv_neo_ubq = self.exchange.get_candles(
-            data_frequency='5m',
+            data_frequency='1d',
             assets=[
                 self.exchange.get_asset('neo_btc'),
                 self.exchange.get_asset('ubq_btc')
             ],
-            bar_count=14
+            bar_count=14,
+            end_dt=pd.to_datetime('2017-10-20', utc=True)
         )
         pass
 

tests/exchange/test_bundle.py

@@ -5,13 +5,14 @@ import pandas as pd
 
 from catalyst import get_calendar
 from catalyst.exchange.bundle_utils import get_bcolz_chunk, \
-    get_periods_range
+    get_periods_range, get_start_dt
 from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
     BcolzExchangeBarWriter
 from catalyst.exchange.exchange_bundle import ExchangeBundle, \
     BUNDLE_NAME_TEMPLATE
 from catalyst.exchange.exchange_utils import get_exchange_folder
 from catalyst.exchange.init_utils import get_exchange
+from catalyst.exchange.stats_utils import df_to_string
 from catalyst.utils.paths import ensure_directory
 
 log = getLogger('test_exchange_bundle')
@@ -44,11 +45,11 @@ class TestExchangeBundle:
         exchange = get_exchange(exchange_name)
         exchange_bundle = ExchangeBundle(exchange)
         assets = [
-            exchange.get_asset('neo_eth')
+            exchange.get_asset('iot_btc')
         ]
 
         # start = pd.to_datetime('2017-09-01', utc=True)
-        start = pd.to_datetime('2017-9-15', utc=True)
+        start = pd.to_datetime('2017-9-01', utc=True)
         end = pd.to_datetime('2017-9-30', utc=True)
 
         log.info('ingesting exchange bundle {}'.format(exchange_name))
@@ -315,3 +316,113 @@ class TestExchangeBundle:
             hashlib.sha256(symbol.encode('utf-8')).hexdigest(), 16
         ) % 10 ** 6
         pass
+
+    def test_validate_data(self):
+        exchange_name = 'bitfinex'
+        data_frequency = 'minute'
+
+        exchange = get_exchange(exchange_name)
+        exchange_bundle = ExchangeBundle(exchange)
+        assets = [exchange.get_asset('iot_btc')]
+
+        end_dt = pd.to_datetime('2017-9-2 1:00', utc=True)
+        bar_count = 60
+
+        bundle_series = exchange_bundle.get_history_window_series(
+            assets=assets,
+            end_dt=end_dt,
+            bar_count=bar_count * 5,
+            field='close',
+            data_frequency='minute',
+        )
+        candles = exchange.get_candles(
+            assets=assets,
+            end_dt=end_dt,
+            bar_count=bar_count,
+            data_frequency='minute'
+        )
+        start_dt = get_start_dt(end_dt, bar_count, data_frequency)
+
+        frames = []
+        for asset in assets:
+            bundle_df = pd.DataFrame(
+                data=dict(bundle_price=bundle_series[asset]),
+                index=bundle_series[asset].index
+            )
+            exchange_series = exchange.get_series_from_candles(
+                candles=candles[asset],
+                start_dt=start_dt,
+                end_dt=end_dt,
+                data_frequency=data_frequency,
+                field='close'
+            )
+            exchange_df = pd.DataFrame(
+                data=dict(exchange_price=exchange_series),
+                index=exchange_series.index
+            )
+
+            df = exchange_df.join(bundle_df, how='left')
+            df['last_traded'] = df.index
+            df['asset'] = asset.symbol
+            df.set_index(['asset', 'last_traded'], inplace=True)
+
+            frames.append(df)
+
+        df = pd.concat(frames)
+        print('\n' + df_to_string(df))
+        pass
+
+    def test_ingest_candles(self):
+        exchange_name = 'bitfinex'
+        data_frequency = 'minute'
+
+        exchange = get_exchange(exchange_name)
+        bundle = ExchangeBundle(exchange)
+        assets = [exchange.get_asset('iot_btc')]
+
+        end_dt = pd.to_datetime('2017-10-20', utc=True)
+        bar_count = 100
+
+        start_dt = get_start_dt(end_dt, bar_count, data_frequency)
+        candles = exchange.get_candles(
+            assets=assets,
+            start_dt=start_dt,
+            end_dt=end_dt,
+            bar_count=bar_count,
+            data_frequency=data_frequency
+        )
+
+        writer = bundle.get_writer(start_dt, end_dt, data_frequency)
+        for asset in assets:
+            dates = [candle['last_traded'] for candle in candles[asset]]
+
+            values = dict()
+            for field in ['open', 'high', 'low', 'close', 'volume']:
+                values[field] = [candle[field] for candle in candles[asset]]
+
+            periods = bundle.get_calendar_periods_range(
+                start_dt, end_dt, data_frequency
+            )
+            df = pd.DataFrame(values, index=dates)
+            df = df.loc[periods].fillna(method='ffill')
+
+            # TODO: why do I get an extra bar?
+            bundle.ingest_df(
+                ohlcv_df=df,
+                data_frequency=data_frequency,
+                asset=asset,
+                writer=writer,
+                empty_rows_behavior='raise'
+            )
+
+        bundle_series = bundle.get_history_window_series(
+            assets=assets,
+            end_dt=end_dt,
+            bar_count=bar_count,
+            field='close',
+            data_frequency=data_frequency,
+            reset_reader=True
+        )
+        df = pd.DataFrame(bundle_series)
+        print('\n' + df_to_string(df))
+        pass