Fixed date range issues and issues retrieving the benchmark data

catalyst/data/loader.py

@@ -37,11 +37,11 @@ logger = logbook.Logger('Loader')
 # Mapping from index symbol to appropriate bond data
 INDEX_MAPPING = {
     'SPY':
-    (treasuries, 'treasury_curves.csv', 'www.federalreserve.gov'),
+        (treasuries, 'treasury_curves.csv', 'www.federalreserve.gov'),
     '^GSPTSE':
-    (treasuries_can, 'treasury_curves_can.csv', 'bankofcanada.ca'),
+        (treasuries_can, 'treasury_curves_can.csv', 'bankofcanada.ca'),
     '^FTSE':  # use US treasuries until UK bonds implemented
-    (treasuries, 'treasury_curves.csv', 'www.federalreserve.gov'),
+        (treasuries, 'treasury_curves.csv', 'www.federalreserve.gov'),
 }
 
 ONE_HOUR = pd.Timedelta(hours=1)
@@ -89,14 +89,19 @@ def has_data_for_dates(series_or_df, first_date, last_date):
     if not isinstance(dts, pd.DatetimeIndex):
         raise TypeError("Expected a DatetimeIndex, but got %s." % type(dts))
     first, last = dts[[0, -1]].tz_localize(None)
-    return (first <= first_date.tz_localize(None)) and (last >= last_date.tz_localize(None))
+    return (first <= first_date.tz_localize(None)) and (
+        last >= last_date.tz_localize(None))
 
-def load_crypto_market_data(trading_day=None, trading_days=None, bm_symbol='USDT_BTC',
-                            bundle=None, bundle_data=None, environ=None):
 
+def load_crypto_market_data(trading_day=None, trading_days=None,
+                            bm_symbol=None, bundle=None, bundle_data=None,
+                            environ=None, exchange=None):
     if trading_day is None:
         trading_day = get_calendar('OPEN').trading_day
-    #if trading_days is None:
+
+    # TODO: consider making configurable
+    bm_symbol = 'btc_usdt'
+    # if trading_days is None:
     #    trading_days = get_calendar('OPEN').schedule
 
     first_date = get_calendar('OPEN').first_trading_session
@@ -128,28 +133,29 @@ def load_crypto_market_data(trading_day=None, trading_days=None, bm_symbol='USDT
     '''
     last_date = trading_days[trading_days.get_loc(now, method='ffill') - 1]
 
-    # This is exceptional, since placing the import at the module scope breaks things
-    # and it's only needed here
-    from catalyst.exchange.poloniex.poloniex import Poloniex
+    if exchange is None:
+        # This is exceptional, since placing the import at the module scope
+        #  breaks things and it's only needed here
+        from catalyst.exchange.poloniex.poloniex import Poloniex
+        exchange = Poloniex('', '', '')
 
-    exchange = Poloniex('','','')
-    btc_usdt = exchange.get_asset('btc_usdt')
+    benchmark_asset = exchange.get_asset(bm_symbol)
 
     # exchange.get_history_window() already ensures that we have the right data
     # for the right dates
     br = exchange.get_history_window(
-                assets = [btc_usdt,], 
-                end_dt = last_date, 
-                bar_count = pd.Timedelta(last_date - first_date).days,
-                frequency =  '1d',
-                field  = 'close',
-                data_frequency = 'daily')
+        assets=[benchmark_asset],
+        end_dt=last_date,
+        bar_count=pd.Timedelta(last_date - first_date).days,
+        frequency='1d',
+        field='close',
+        data_frequency='daily')
     br.columns = ['close']
     br = br.pct_change(1).iloc[1:]
 
     # Override first_date for treasury data since we have it for many more years
     # and is independent of crypto data
-    first_date_treasury = pd.Timestamp('1990-01-02', tz='UTC') 
+    first_date_treasury = pd.Timestamp('1990-01-02', tz='UTC')
     tc = ensure_treasury_data(
         bm_symbol,
         first_date_treasury,
@@ -158,7 +164,8 @@ def load_crypto_market_data(trading_day=None, trading_days=None, bm_symbol='USDT
         environ,
     )
     benchmark_returns = br[br.index.slice_indexer(first_date, last_date)]
-    treasury_curves = tc[tc.index.slice_indexer(first_date_treasury, last_date)]
+    treasury_curves = tc[
+        tc.index.slice_indexer(first_date_treasury, last_date)]
     return benchmark_returns, treasury_curves
 
 
@@ -256,12 +263,11 @@ def ensure_crypto_benchmark_data(symbol,
                                  bundle,
                                  bundle_data,
                                  environ=None):
-
     filename = get_benchmark_filename(symbol)
 
     logger.info(
         ('Loading benchmark data for {symbol!r} '
-            'from {first_date} to {last_date}'),
+         'from {first_date} to {last_date}'),
         symbol=symbol,
         first_date=first_date,
         last_date=last_date
@@ -282,7 +288,7 @@ def ensure_crypto_benchmark_data(symbol,
     # If no cached data was found or it was missing any dates then download the
     # necessary data.
 
-    if(bundle == 'poloniex'):
+    if (bundle == 'poloniex'):
         '''
         If we're using the Poloniex bundle, we'll get the benchmark from the bundle
         instead of downloading it from Poloniex every time we need it.
@@ -290,27 +296,34 @@ def ensure_crypto_benchmark_data(symbol,
         prevents users abroad from getting Catalyst to work
         '''
         logger.info(
-            ('Retrieving benchmark data from bundle for {symbol!r} from {first_date} to {last_date}'),
+            (
+                'Retrieving benchmark data from bundle for {symbol!r} from {first_date} to {last_date}'),
             symbol=symbol, first_date=first_date, last_date=last_date)
 
-        asset = bundle_data.asset_finder.lookup_symbol(symbol=symbol,as_of_date=None)
+        asset = bundle_data.asset_finder.lookup_symbol(symbol=symbol,
+                                                       as_of_date=None)
         fields = ['day', 'close']
         raw = bundle_data.daily_bar_reader.load_raw_arrays(
             columns=fields,
             start_date=first_date - trading_day,
             end_date=last_date,
-            assets=[asset,])
-        bench_raw = pd.concat([pd.DataFrame(raw[0], columns=['date']),pd.DataFrame(raw[1], columns=['close'])], axis=1)
-        bench_raw['date'] = pd.to_datetime(bench_raw['date'],unit='s')
+            assets=[asset, ])
+        bench_raw = pd.concat([pd.DataFrame(raw[0], columns=['date']),
+                               pd.DataFrame(raw[1], columns=['close'])],
+                              axis=1)
+        bench_raw['date'] = pd.to_datetime(bench_raw['date'], unit='s')
         bench_raw.set_index('date', inplace=True)
         bench_raw.sort_index(inplace=True)
-        bench_raw = bench_raw[pd.to_datetime(first_date - trading_day):pd.to_datetime(last_date)]
+        bench_raw = bench_raw[
+                    pd.to_datetime(first_date - trading_day):pd.to_datetime(
+                        last_date)]
 
     else:
         # This is how it used to be: downloading the benchmark everytime. 
         # Leaving this code here to be repurposed in the future for other bundles.
         logger.info(
-            ('Downloading benchmark data for {symbol!r} from {first_date} to {last_date}'),
+            (
+                'Downloading benchmark data for {symbol!r} from {first_date} to {last_date}'),
             symbol=symbol, first_date=first_date, last_date=last_date)
 
         raise DeprecationWarning('poloniex bundle deprecated')
@@ -386,7 +399,7 @@ def ensure_benchmark_data(symbol, first_date, last_date, now, trading_day,
     # necessary data.
     logger.info(
         ('Downloading benchmark data for {symbol!r} '
-            'from {first_date} to {last_date}'),
+         'from {first_date} to {last_date}'),
         symbol=symbol,
         first_date=first_date - trading_day,
         last_date=last_date
@@ -447,7 +460,7 @@ def ensure_benchmark_data(symbol, first_date, last_date, now, trading_day,
     # necessary data.
     logger.info(
         ('Downloading benchmark data for {symbol!r} '
-            'from {first_date} to {last_date}'),
+         'from {first_date} to {last_date}'),
         symbol=symbol,
         first_date=first_date - trading_day,
         last_date=last_date
@@ -531,7 +544,8 @@ def _load_cached_data(filename, first_date, last_date, now, resource_name,
             data = pd.DataFrame.from_csv(path)
             if data.empty:
                 raise ValueError("File is empty.")
-            data.index = pd.to_datetime(data.index, infer_datetime_format=True, errors='coerce' ).tz_localize('UTC')
+            data.index = pd.to_datetime(data.index, infer_datetime_format=True,
+                                        errors='coerce').tz_localize('UTC')
             if has_data_for_dates(data, first_date, last_date):
                 return data
 

catalyst/exchange/bundle_utils.py

@@ -1,8 +1,9 @@
+import calendar
 import tarfile
 import shutil
 
 import requests
-from datetime import timedelta, datetime
+from datetime import timedelta, datetime, date
 import os
 from logging import Logger
 import pandas as pd
@@ -12,7 +13,8 @@ import pytz
 
 from catalyst.data.bundles import from_bundle_ingest_dirname
 from catalyst.data.bundles.core import download_without_progress
-from catalyst.exchange.exchange_errors import ApiCandlesError
+from catalyst.exchange.exchange_errors import ApiCandlesError, \
+    PricingDataBeforeTradingError
 from catalyst.exchange.exchange_utils import get_exchange_bundles_folder
 from catalyst.utils.deprecate import deprecated
 from catalyst.utils.paths import data_path
@@ -102,6 +104,82 @@ def get_start_dt(end_dt, bar_count, data_frequency):
     return start_dt
 
 
+def get_adj_dates(start, end, assets, data_frequency):
+    """
+    Contains a date range to the trading availability of the specified pairs.
+
+    :param start:
+    :param end:
+    :param assets:
+    :param data_frequency:
+    :return:
+    """
+    earliest_trade = None
+    last_entry = None
+    for asset in assets:
+        if earliest_trade is None or earliest_trade > asset.start_date:
+            earliest_trade = asset.start_date
+
+        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 start is None or earliest_trade > start:
+        log.debug(
+            'adjusting start date to earliest trade date found {}'.format(
+                earliest_trade
+            ))
+        start = earliest_trade
+
+    if end is None or (last_entry is not None and end > last_entry):
+        log.debug('adjusting the end date to now {}'.format(last_entry))
+        end = last_entry
+
+    if start >= end:
+        raise PricingDataBeforeTradingError(
+            symbols=[asset.symbol],
+            exchange=asset.exchange,
+            first_trading_day=earliest_trade,
+            dt=end
+        )
+
+    return start, end
+
+
+def get_month_start_end(dt):
+    """
+    Returns the first and last day of the month for the specified date.
+
+    :param dt:
+    :return:
+    """
+    month_range = calendar.monthrange(dt.year, dt.month)
+    month_start = pd.to_datetime(datetime(
+        dt.year, dt.month, 1, 0, 0, 0, 0
+    ), utc=True)
+
+    month_end = pd.to_datetime(datetime(
+        dt.year, dt.month, month_range[1], 23, 59, 0, 0
+    ), utc=True)
+
+    return month_start, month_end
+
+
+def get_year_start_end(dt):
+    """
+    Returns the first and last day of the year for the specified date.
+
+    :param dt:
+    :return:
+    """
+    year_start = pd.to_datetime(date(dt.year, 1, 1), utc=True)
+    year_end = pd.to_datetime(date(dt.year, 12, 31), utc=True)
+
+    return year_start, year_end
+
+
 def get_ffill_candles(candles, bar_count, end_dt, data_frequency,
                       previous_candle=None):
     """

catalyst/exchange/data_portal_exchange.py

@@ -281,7 +281,6 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
                                     ffill=True):
         bundle = self.exchange_bundles[exchange.name]
 
-        reader = bundle.get_reader(data_frequency)
         if data_frequency == 'minute':
             dts = self.trading_calendar.minutes_window(
                 end_dt, -bar_count
@@ -299,12 +298,6 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
             raise InvalidHistoryFrequencyError(frequency=data_frequency)
 
         try:
-            # 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],
@@ -333,8 +326,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
             value_series = pd.Series(asset_values, index=dts)
             series[asset] = value_series
 
-        df = pd.DataFrame(series)
-        return df
+        return pd.DataFrame(series)
 
     def ensure_after_first_day(self, dt, assets):
         first_trading_day = self._get_first_trading_day(assets)

catalyst/exchange/exchange.py

@@ -13,7 +13,7 @@ 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_start_dt, \
-    get_delta, get_trailing_candles_dt, get_periods
+    get_delta, get_trailing_candles_dt, get_periods, get_adj_dates
 from catalyst.exchange.exchange_bundle import ExchangeBundle
 from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \
     InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \
@@ -486,6 +486,9 @@ class Exchange:
         adj_bar_count = candle_size * bar_count
         start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency)
 
+        start_dt, end_dt = get_adj_dates(start_dt, end_dt, assets,
+                                         data_frequency)
+
         missing_assets = bundle.filter_existing_assets(
             assets=assets,
             start_dt=start_dt,
@@ -511,6 +514,8 @@ class Exchange:
                     asset=chunk['asset'],
                     data_frequency=data_frequency,
                     period=chunk['period'],
+                    start_dt=chunk['period_start'],
+                    end_dt=chunk['period_end'],
                     writer=writer
                 )
 

catalyst/exchange/exchange_bundle.py

@@ -5,6 +5,7 @@ from datetime import timedelta, datetime
 
 import pandas as pd
 from logbook import Logger, INFO
+from pandas.tseries.offsets import MonthBegin, YearBegin, YearEnd
 
 from catalyst import get_calendar
 from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
@@ -12,9 +13,10 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
 from catalyst.data.us_equity_pricing import BcolzDailyBarWriter, \
     BcolzDailyBarReader
 from catalyst.exchange.bundle_utils import get_ffill_candles, range_in_bundle, \
-    get_bcolz_chunk, get_delta
+    get_bcolz_chunk, get_delta, get_adj_dates, get_month_start_end, \
+    get_year_start_end
 from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
-    InvalidHistoryFrequencyError
+    InvalidHistoryFrequencyError, PricingDataBeforeTradingError
 from catalyst.exchange.exchange_utils import get_exchange_folder
 from catalyst.utils.cli import maybe_show_progress
 from catalyst.utils.paths import ensure_directory
@@ -48,36 +50,6 @@ class ExchangeBundle:
         else:
             return self.exchange.get_assets()
 
-    def get_adj_dates(self, start, end, assets, data_frequency):
-
-        earliest_trade = None
-        last_entry = None
-        for asset in assets:
-            if earliest_trade is None or earliest_trade > asset.start_date:
-                earliest_trade = asset.start_date
-
-            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 start is None or earliest_trade > start:
-            log.debug(
-                'adjusting start date to earliest trade date found {}'.format(
-                    earliest_trade
-                ))
-            start = earliest_trade
-
-        if end is None or (last_entry is not None and end > last_entry):
-            log.debug('adjusting the end date to now {}'.format(last_entry))
-            end = last_entry
-
-        if start >= end:
-            raise ValueError('start date cannot be after end date')
-
-        return start, end
-
     def get_reader(self, data_frequency, path=None):
         """
         Get a data writer object, either a new object or from cache
@@ -346,8 +318,8 @@ class ExchangeBundle:
         :return:
         """
 
-    def ingest_ctable(self, asset, data_frequency, period, writer,
-                      empty_rows_behavior='strip', cleanup=False):
+    def ingest_ctable(self, asset, data_frequency, period, start_dt, end_dt,
+                      writer, empty_rows_behavior='strip', cleanup=False):
         """
         Merge a ctable bundle chunk into the main bundle for the exchange.
 
@@ -371,11 +343,6 @@ class ExchangeBundle:
             period=period
         )
 
-        # 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
-
         periods = self.calendar.minutes_in_range(start_dt, end_dt) \
             if data_frequency == 'minute' \
             else self.calendar.sessions_in_range(start_dt, end_dt)
@@ -474,10 +441,9 @@ class ExchangeBundle:
         for asset in assets:
             try:
                 asset_start, asset_end = \
-                    self.get_adj_dates(start_dt, end_dt, [asset],
-                                       data_frequency)
+                    get_adj_dates(start_dt, end_dt, [asset], data_frequency)
 
-            except ValueError:
+            except PricingDataBeforeTradingError:
                 continue
 
             sessions = self.calendar.sessions_in_range(asset_start, asset_end)
@@ -491,46 +457,48 @@ class ExchangeBundle:
                 if period not in periods:
                     periods.append(period)
 
+                    # Adjusting the period dates to match the availability
+                    # of the trading pair
                     if data_frequency == 'minute':
-                        month_range = calendar.monthrange(dt.year, dt.month)
-                        period_start = pd.to_datetime(
-                            datetime(dt.year, dt.month, 1, 0, 0, 0, 0),
-                            utc=True)
+                        period_start, period_end = get_month_start_end(dt)
+                        asset_start_month, _ = get_month_start_end(asset_start)
 
-                        period_end = pd.to_datetime(
-                            datetime(
-                                dt.year, dt.month, month_range[1], 23, 59, 0,
-                                0),
-                            utc=True
-                        )
+                        if asset_start_month == period_start \
+                                and period_start < asset_start:
+                            period_start = asset_start
+
+                        _, asset_end_month = get_month_start_end(asset_end)
+                        if asset_end_month == period_end \
+                                and period_end > asset_end:
+                            period_end = asset_end
 
                     elif data_frequency == 'daily':
-                        period_start = pd.to_datetime(
-                            datetime(dt.year, 1, 1, 0, 0, 0, 0),
-                            utc=True)
+                        period_start, period_end = get_year_start_end(dt)
+                        asset_start_year, _ = get_year_start_end(asset_start)
 
-                        period_end = pd.to_datetime(
-                            datetime(
-                                dt.year, 12, 31, 23, 59, 0, 0),
-                            utc=True
-                        )
+                        if asset_start_year == period_start \
+                                and period_start < asset_start:
+                            period_start = asset_start
+
+                        _, asset_end_year = get_year_start_end(asset_end)
+                        if asset_end_year == period_end \
+                                and period_end > asset_end:
+                            period_end = asset_end
                     else:
                         raise InvalidHistoryFrequencyError(
                             frequency=data_frequency
                         )
 
-                    if period_end > asset_end:
-                        period_end = asset_end
-
-                    has_data = \
-                        range_in_bundle(asset, period_start, period_end,
-                                        reader)
+                    has_data = range_in_bundle(
+                        asset, period_start, period_end, reader
+                    )
 
                     if not has_data:
                         log.debug('adding period: {}'.format(period))
                         chunks.append(
                             dict(
                                 asset=asset,
+                                period_start=period_start,
                                 period_end=period_end,
                                 period=period
                             )
@@ -557,7 +525,7 @@ class ExchangeBundle:
         :return:
         """
         assets = self.get_assets(include_symbols, exclude_symbols)
-        start, end = self.get_adj_dates(start, end, assets, data_frequency)
+        start, end = get_adj_dates(start, end, assets, data_frequency)
 
         writer = self.get_writer(start, end, data_frequency)
         chunks = self.prepare_chunks(
@@ -578,6 +546,8 @@ class ExchangeBundle:
                     asset=chunk['asset'],
                     data_frequency=data_frequency,
                     period=chunk['period'],
+                    start_dt=chunk['period_start'],
+                    end_dt=chunk['period_end'],
                     writer=writer,
                     empty_rows_behavior='strip'
                 )

catalyst/exchange/exchange_errors.py

@@ -189,6 +189,5 @@ 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

@@ -92,6 +92,8 @@ class ExchangeBundleTestCase:
             asset=asset,
             data_frequency=data_frequency,
             period='2017',
+            start_dt=start,
+            end_dt=end,
             writer=writer,
             empty_rows_behavior='strip'
         )