diff --git a/catalyst/data/loader.py b/catalyst/data/loader.py index 4167bd84..fa80317a 100644 --- a/catalyst/data/loader.py +++ b/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 diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 2dd7da07..e5f970d8 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/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): """ diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index ed2f8ed6..d7b163ac 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/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) diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 7c6f5d41..93cc128b 100644 --- a/catalyst/exchange/exchange.py +++ b/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 ) diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 090775aa..dbc42c0b 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/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' ) diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py index 7ec2f59e..b6c51155 100644 --- a/catalyst/exchange/exchange_errors.py +++ b/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() diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index c0110476..5772f6ea 100644 --- a/tests/exchange/test_bundle.py +++ b/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' )