Reading data from bundles first and other fixes

catalyst/__main__.py

@@ -9,6 +9,7 @@ from six import text_type
 
 from catalyst.data import bundles as bundles_module
 from catalyst.exchange.exchange_bundle import ExchangeBundle
+from catalyst.exchange.init_utils import get_exchange
 from catalyst.utils.cli import Date, Timestamp
 from catalyst.utils.run_algo import _run, load_extensions
 
@@ -492,7 +493,8 @@ def ingest_exchange(exchange_name, data_frequency, start, end,
     """
     Ingest data for the given exchange.
     """
-    exchange_bundle = ExchangeBundle(exchange_name)
+    exchange=get_exchange(exchange_name)
+    exchange_bundle = ExchangeBundle(exchange)
 
     click.echo('ingesting exchange bundle {}'.format(exchange_name))
     exchange_bundle.ingest(

catalyst/examples/buy_low_sell_high_neo_with_interface.py

@@ -36,8 +36,8 @@ def _handle_data(context, data):
     prices = data.history(
         context.asset,
         fields='price',
-        bar_count=30,
-        frequency='30m'
+        bar_count=50,
+        frequency='1m'
     )
     rsi = talib.RSI(prices.values, timeperiod=14)[-1]
     log.info('got rsi: {}'.format(rsi))
@@ -148,27 +148,27 @@ def analyze(context, stats):
     pass
 
 
-# run_algorithm(
-#     initialize=initialize,
-#     handle_data=handle_data,
-#     analyze=analyze,
-#     exchange_name='bitfinex',
-#     live=True,
-#     algo_namespace=algo_namespace,
-#     base_currency='btc',
-#     live_graph=False
-# )
-
-# Backtest
 run_algorithm(
-    capital_base=250,
-    start=pd.to_datetime('2017-09-08', utc=True),
-    end=pd.to_datetime('2017-09-15', utc=True),
-    data_frequency='minute',
     initialize=initialize,
     handle_data=handle_data,
     analyze=analyze,
     exchange_name='bitfinex',
+    live=True,
     algo_namespace=algo_namespace,
-    base_currency='btc'
+    base_currency='btc',
+    live_graph=False
 )
+
+# Backtest
+# run_algorithm(
+#     capital_base=250,
+#     start=pd.to_datetime('2017-09-08', utc=True),
+#     end=pd.to_datetime('2017-09-15', utc=True),
+#     data_frequency='minute',
+#     initialize=initialize,
+#     handle_data=handle_data,
+#     analyze=analyze,
+#     exchange_name='bitfinex',
+#     algo_namespace=algo_namespace,
+#     base_currency='btc'
+# )

catalyst/exchange/bitfinex/bitfinex.py

@@ -14,7 +14,6 @@ import six
 from catalyst.assets._assets import TradingPair
 from logbook import Logger
 
-# from websocket import create_connection
 from catalyst.exchange.exchange import Exchange
 from catalyst.exchange.exchange_errors import (
     ExchangeRequestError,

catalyst/exchange/bundle_utils.py

@@ -1,4 +1,5 @@
-import datetime, requests
+import requests
+from datetime import timedelta, datetime
 import os
 from logging import Logger
 import pandas as pd
@@ -16,11 +17,11 @@ API_URL = 'http://data.enigma.co/api/v1'
 
 
 def get_date_from_ms(ms):
-    return datetime.datetime.fromtimestamp(ms / 1000.0)
+    return datetime.fromtimestamp(ms / 1000.0)
 
 
 def get_seconds_from_date(date):
-    epoch = datetime.datetime.utcfromtimestamp(0)
+    epoch = datetime.utcfromtimestamp(0)
     epoch = epoch.replace(tzinfo=pytz.UTC)
 
     return int((date - epoch).total_seconds())
@@ -107,14 +108,30 @@ 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,
+def get_delta(periods, data_frequency):
+    return timedelta(minutes=periods) \
+        if data_frequency == 'minute' else timedelta(days=periods)
+
+
+def get_start_dt(end_dt, bar_count, data_frequency):
+    periods = bar_count - 1
+    if periods > 1:
+        delta = get_delta(periods, data_frequency)
+        start_dt = end_dt - delta
+    else:
+        start_dt = end_dt
+
+    return start_dt
+
+
+def get_ffill_candles(candles, bar_count, 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 bar_count:
     :param end_dt:
     :param data_frequency:
     :param previous_candle:
@@ -123,6 +140,8 @@ def get_ffill_candles(candles, start_dt, end_dt, data_frequency,
     """
     all_dates = []
     all_candles = []
+
+    start_dt = get_start_dt(end_dt, bar_count, data_frequency)
     date = start_dt
 
     while date <= end_dt:
@@ -130,18 +149,15 @@ def get_ffill_candles(candles, start_dt, end_dt, data_frequency,
             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)
+        if candle is None:
+            candle = candles[0]
 
-            previous_candle = candle
+        all_dates.append(date)
+        all_candles.append(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')
+        previous_candle = candle
+
+        date += get_delta(1, data_frequency)
 
     return all_dates, all_candles
 

catalyst/exchange/data_portal_exchange.py

@@ -260,8 +260,8 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
         self.minute_history_loaders = dict()
 
         for exchange_name in self.exchanges:
-            self.exchange_bundles[exchange_name] = \
-                ExchangeBundle(exchange_name)
+            exchange = self.exchanges[exchange_name]
+            self.exchange_bundles[exchange_name] = ExchangeBundle(exchange)
 
     def _get_first_trading_day(self, assets):
         first_date = None

catalyst/exchange/exchange.py

@@ -1,5 +1,6 @@
 import abc
 import random
+import re
 from abc import ABCMeta, abstractmethod, abstractproperty
 from datetime import timedelta
 from time import sleep
@@ -11,8 +12,12 @@ from logbook import Logger
 
 from catalyst.data.data_portal import BASE_FIELDS
 from catalyst.exchange import bundle_utils
+from catalyst.exchange.bundle_utils import get_ffill_candles, get_start_dt, \
+    get_delta
+from catalyst.exchange.exchange_bundle import ExchangeBundle
 from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \
-    InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange
+    InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \
+    InvalidHistoryFrequencyError
 from catalyst.exchange.exchange_execution import ExchangeStopLimitOrder, \
     ExchangeLimitOrder, ExchangeStopOrder
 from catalyst.exchange.exchange_portfolio import ExchangePortfolio
@@ -456,7 +461,7 @@ class Exchange:
             The last trading date of the last bar.
         :return:
         """
-        start = end - timedelta(minutes=bar_count)
+        start = get_start_dt(end, bar_count, data_frequency)
 
         exchange_start = None
         catalyst_end = None
@@ -475,8 +480,8 @@ class Exchange:
                 exchange_end = end
 
         else:
-            exchange_start = start
             exchange_end = end
+            exchange_start = start
 
         data = []
         if catalyst_end is not None:
@@ -495,7 +500,7 @@ class Exchange:
                 data_frequency=data_frequency,
                 assets=[asset],
                 bar_count=bar_count,
-                start_dt=exchange_start,
+                start_dt=exchange_start if bar_count > 1 else None,
                 end_dt=exchange_end
             )
             data += candles[asset]
@@ -545,25 +550,86 @@ class Exchange:
         A dataframe containing the requested data.
         """
 
-        # TODO: try to read from bundle first
-        candles = self.get_history(
+        bundle = ExchangeBundle(self)
+
+        freq_match = re.match(r'([0-9].*)(m|M|d|D)', frequency, re.M | re.I)
+        if freq_match:
+            candle_size = int(freq_match.group(1))
+            unit = freq_match.group(2)
+        else:
+            raise InvalidHistoryFrequencyError(frequency)
+
+        if unit.lower() == 'd':
+            data_frequency = 'daily'
+
+        elif unit.lower() == 'm':
+            data_frequency = 'minute'
+
+        else:
+            raise InvalidHistoryFrequencyError(frequency)
+
+        adj_bar_count = candle_size * bar_count
+        start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency)
+
+        missing_assets = bundle.filter_existing_assets(
             assets=assets,
+            start_dt=start_dt,
             end_dt=end_dt,
-            bar_count=bar_count,
             data_frequency=data_frequency
         )
 
+        if len(missing_assets) > 0:
+            writer = bundle.get_writer(start_dt, end_dt, data_frequency)
+
+            bundle.ingest_chunk(
+                bar_count=adj_bar_count,
+                end_dt=end_dt,
+                data_frequency=data_frequency,
+                assets=missing_assets,
+                writer=writer
+            )
+
+        reader = bundle.get_reader(data_frequency)
+        values = reader.load_raw_arrays(
+            fields=[field],
+            start_dt=start_dt,
+            end_dt=end_dt,
+            sids=[asset.sid for asset in assets],
+        )[0]
+
         series = dict()
-        for asset in assets:
-            asset_candles = candles[asset]
+        for asset_index, asset in enumerate(assets):
+            all_dates = []
+            asset_values = []
 
-            values = map(lambda candle: candle[field], asset_candles)
-            dates = map(lambda candle: candle['last_traded'], asset_candles)
+            date = start_dt
+            for value in values:
+                all_dates.append(date)
+                asset_values.append(value[asset_index])
 
-            value_series = pd.Series(values, index=dates)
+                date += get_delta(1, data_frequency)
+
+            value_series = pd.Series(asset_values, index=all_dates)
             series[asset] = value_series
 
         df = pd.DataFrame(series)
+
+        if candle_size > 1:
+            if field == 'open':
+                agg = 'first'
+            elif field == 'high':
+                agg = 'max'
+            elif field == 'low':
+                agg = 'min'
+            elif field == 'close':
+                agg = 'last'
+            elif field == 'volume':
+                agg = 'sum'
+            else:
+                raise ValueError('invalid field')
+
+            df = df.resample('{}T'.format(candle_size)).agg(agg)
+
         return df
 
     def synchronize_portfolio(self):

catalyst/exchange/exchange_bundle.py

@@ -4,15 +4,15 @@ from datetime import timedelta
 import numpy as np
 import pandas as pd
 from logbook import Logger
+from pandas import DatetimeIndex
 
 from catalyst import get_calendar
 from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
-    BcolzMinuteBarWriter, BcolzMinuteBarReader
+    BcolzMinuteBarWriter, BcolzMinuteBarReader, BcolzMinuteBarMetadata
 from catalyst.data.us_equity_pricing import BcolzDailyBarWriter, \
     BcolzDailyBarReader
-from catalyst.exchange.bundle_utils import get_ffill_candles
+from catalyst.exchange.bundle_utils import get_ffill_candles, get_start_dt
 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
 from catalyst.utils.paths import ensure_directory
 
@@ -26,8 +26,8 @@ log = Logger('exchange_bundle')
 
 
 class ExchangeBundle:
-    def __init__(self, exchange_name, ):
-        self.exchange = get_exchange(exchange_name)
+    def __init__(self, exchange):
+        self.exchange = exchange
         self.minutes_per_day = 1440
         self.default_ohlc_ratio = 1000000
         self._writers = dict()
@@ -72,7 +72,8 @@ class ExchangeBundle:
 
         :return: BcolzMinuteBarReader or BcolzDailyBarReader
         """
-        if data_frequency in self._readers:
+        if data_frequency in self._readers \
+                and self._readers[data_frequency] is not None:
             return self._readers[data_frequency]
 
         root = get_exchange_folder(self.exchange.name)
@@ -81,6 +82,7 @@ class ExchangeBundle:
             frequency=data_frequency
         )
 
+        self._readers[data_frequency] = None
         if data_frequency == 'minute':
             try:
                 self._readers[data_frequency] = BcolzMinuteBarReader(input_dir)
@@ -99,13 +101,16 @@ class ExchangeBundle:
 
         return self._readers[data_frequency]
 
-    def get_writer(self, data_frequency, start, end):
+    def update_metadata(self, writer, start_dt, end_dt):
+        pass
+
+    def get_writer(self, start_dt, end_dt, data_frequency):
         """
         Get a data writer object, either a new object or from cache
 
         :return: BcolzMinuteBarWriter or BcolzDailyBarWriter
         """
-        key = (data_frequency, start, end)
+        key = data_frequency
         if key in self._writers:
             return self._writers[key]
 
@@ -120,27 +125,59 @@ class ExchangeBundle:
 
         if data_frequency == 'minute':
             if len(os.listdir(output_dir)) > 0:
+
+                metadata = BcolzMinuteBarMetadata.read(output_dir)
+
+                write_metadata = False
+                if start_dt < metadata.start_session:
+                    write_metadata = True
+                    start_session = start_dt.floor('1d')
+                else:
+                    start_session = metadata.start_session
+
+                if end_dt > metadata.end_session:
+                    write_metadata = True
+
+                    # TODO: workaround, improve the calendar logic?
+                    if end_dt == start_dt:
+                        end_dt += timedelta(days=1)
+
+                    end_session = end_dt.floor('1d')
+                else:
+                    end_session = metadata.end_session
+
                 self._writers[key] = \
-                    BcolzMinuteBarWriter.open(output_dir, end)
+                    BcolzMinuteBarWriter(
+                        output_dir,
+                        metadata.calendar,
+                        start_session,
+                        end_session,
+                        metadata.minutes_per_day,
+                        metadata.default_ohlc_ratio,
+                        metadata.ohlc_ratios_per_sid,
+                        write_metadata=write_metadata
+                    )
             else:
                 self._writers[key] = BcolzMinuteBarWriter(
                     rootdir=output_dir,
                     calendar=open_calendar,
                     minutes_per_day=self.minutes_per_day,
-                    start_session=start,
-                    end_session=end,
+                    start_session=start_dt,
+                    end_session=end_dt,
                     write_metadata=True,
                     default_ohlc_ratio=self.default_ohlc_ratio
                 )
+
         elif data_frequency == 'daily':
             if len(os.listdir(output_dir)) > 0:
-                self._writers[key] = BcolzDailyBarWriter.open(output_dir, end)
+                self._writers[key] = \
+                    BcolzDailyBarWriter.open(output_dir, end_dt)
             else:
-                end_session = end.floor('1d')
+                end_session = end_dt.floor('1d')
                 self._writers[key] = BcolzDailyBarWriter(
                     filename=output_dir,
                     calendar=open_calendar,
-                    start_session=start,
+                    start_session=start_dt,
                     end_session=end_session
                 )
         else:
@@ -150,7 +187,7 @@ class ExchangeBundle:
 
         return self._writers[key]
 
-    def filter_existing_assets(self, assets, start, end, data_frequency):
+    def filter_existing_assets(self, assets, start_dt, end_dt, data_frequency):
         """
         For each asset, get the close on the start and end dates of the chunk.
             If the data exists, the chunk ingestion is complete.
@@ -158,9 +195,9 @@ class ExchangeBundle:
 
         :param assets: list[TradingPair]
             The assets is scope.
-        :param start:
+        :param start_dt:
             The chunk start date.
-        :param end:
+        :param end_dt:
             The chunk end date.
         :return: list[TradingPair]
             The assets missing from the bundle
@@ -171,13 +208,15 @@ class ExchangeBundle:
             has_data = True
             if has_data and reader is not None:
                 try:
-                    start_close = reader.get_value(asset.sid, start, 'close')
+                    start_close = \
+                        reader.get_value(asset.sid, start_dt, 'close')
 
                     if np.isnan(start_close):
                         has_data = False
 
                     else:
-                        end_close = reader.get_value(asset.sid, end, 'close')
+                        end_close = reader.get_value(asset.sid, end_dt,
+                                                     'close')
 
                         if np.isnan(end_close):
                             has_data = False
@@ -193,8 +232,8 @@ class ExchangeBundle:
 
         return missing_assets
 
-    def ingest_chunk(self, chunk, previous_candle, data_frequency, assets,
-                     writer):
+    def ingest_chunk(self, bar_count, end_dt, data_frequency, assets,
+                     writer, previous_candle=dict()):
         """
         Retrieve the specified OHLCV chunk and write it to the bundle
 
@@ -205,18 +244,17 @@ class ExchangeBundle:
         :param writer:
         :return:
         """
-        chunk_end = chunk['end']
-        chunk_start = chunk_end - timedelta(minutes=chunk['bar_count'])
 
         chunk_assets = []
         for asset in assets:
-            if asset.start_date <= chunk_end:
+            if asset.start_date <= end_dt:
                 chunk_assets.append(asset)
 
+        start_dt = get_start_dt(end_dt, bar_count, data_frequency)
         missing_assets = self.filter_existing_assets(
             assets=chunk_assets,
-            start=chunk_start,
-            end=chunk_end,
+            start_dt=start_dt,
+            end_dt=end_dt,
             data_frequency=data_frequency
         )
 
@@ -226,8 +264,8 @@ class ExchangeBundle:
 
         candles = self.exchange.get_history(
             assets=missing_assets,
-            end_dt=chunk_end,
-            bar_count=chunk['bar_count'],
+            end_dt=end_dt,
+            bar_count=bar_count,
             data_frequency=data_frequency
         )
 
@@ -240,7 +278,7 @@ class ExchangeBundle:
                     'no data: {symbols} on {exchange}, date {end}'.format(
                         symbols=missing_assets,
                         exchange=self.exchange.name,
-                        end=chunk_end
+                        end=end_dt
                     )
                 )
                 continue
@@ -250,14 +288,18 @@ class ExchangeBundle:
 
             all_dates, all_candles = get_ffill_candles(
                 candles=asset_candles,
-                start_dt=chunk_start,
-                end_dt=chunk_end,
+                bar_count=bar_count,
+                end_dt=end_dt,
                 data_frequency=data_frequency,
                 previous_candle=previous
             )
             previous_candle[asset] = all_candles[-1]
 
-            df = pd.DataFrame(all_candles, index=all_dates)
+            df = pd.DataFrame(
+                data=all_candles,
+                index=all_dates,
+                columns=['open', 'high', 'low', 'close', 'volume']
+            )
             if not df.empty:
                 df.sort_index(inplace=True)
 
@@ -268,24 +310,37 @@ class ExchangeBundle:
 
         try:
             log.debug(
-                'writing {num_candles} candles from {start} to {end}'.format(
+                'writing {num_candles} candles for {bar_count} bars'
+                'ending {end}'.format(
                     num_candles=num_candles,
-                    start=chunk_start,
-                    end=chunk_end
+                    bar_count=bar_count,
+                    end=end_dt
                 )
             )
 
-            for pair in data:
-                log.debug('data for sid {}\n{}\n{}'.format(
-                    pair[0], pair[1].head(2), pair[1].tail(2)))
-
             writer.write(
                 data=data,
                 show_progress=False,
                 invalid_data_behavior='raise'
             )
         except BcolzMinuteOverlappingData as e:
-            log.warn('chunk already exists {}: {}'.format(chunk, e))
+            log.warn('chunk already exists: {}'.format(e))
+        except Exception as e:
+            log.warn('error when writing data: {}, trying again'.format(e))
+
+            # This is workaround, there is an issue with empty
+            # session_label when using a newly created writer
+            del self._writers[data_frequency]
+
+            # TODO: these are the dates of the chunk, not the job
+            writer = self.get_writer(start_dt, end_dt, data_frequency)
+            writer.write(
+                data=data,
+                show_progress=False,
+                invalid_data_behavior='raise'
+            )
+
+        return data
 
     def ingest(self, data_frequency, include_symbols=None,
                exclude_symbols=None, start=None, end=None,
@@ -327,7 +382,7 @@ class ExchangeBundle:
         else:
             raise ValueError('frequency not supported')
 
-        writer = self.get_writer(data_frequency, start, end)
+        writer = self.get_writer(start, end, data_frequency)
 
         if delta_periods > self.exchange.num_candles_limit:
             bar_count = self.exchange.num_candles_limit
@@ -359,9 +414,10 @@ class ExchangeBundle:
             previous_candle = dict()
             for chunk in it:
                 self.ingest_chunk(
-                    chunk=chunk,
-                    previous_candle=previous_candle,
+                    bar_count=chunk['bar_count'],
+                    end_dt=chunk['end'],
                     data_frequency=data_frequency,
                     assets=assets,
-                    writer=writer
+                    writer=writer,
+                    previous_candle=previous_candle,
                 )

tests/exchange/test_bundle.py

@@ -3,6 +3,7 @@ from logging import Logger
 import pandas as pd
 
 from catalyst.exchange.exchange_bundle import ExchangeBundle
+from catalyst.exchange.init_utils import get_exchange
 
 log = Logger('test_exchange_bundle')
 
@@ -14,7 +15,7 @@ class ExchangeBundleTestCase:
         start = pd.to_datetime('2017-09-01', utc=True)
         end = pd.Timestamp.utcnow()
 
-        exchange_bundle = ExchangeBundle(exchange_name)
+        exchange_bundle = ExchangeBundle(get_exchange(exchange_name))
 
         log.info('ingesting exchange bundle {}'.format(exchange_name))
         exchange_bundle.ingest(
@@ -33,7 +34,7 @@ class ExchangeBundleTestCase:
         start = pd.to_datetime('2017-09-01', utc=True)
         end = pd.Timestamp.utcnow()
 
-        exchange_bundle = ExchangeBundle(exchange_name)
+        exchange_bundle = ExchangeBundle(get_exchange(exchange_name))
 
         log.info('ingesting exchange bundle {}'.format(exchange_name))
         exchange_bundle.ingest(