diff --git a/catalyst/data/bundles/base.py b/catalyst/data/bundles/base.py index 613f5051..c8a6370c 100644 --- a/catalyst/data/bundles/base.py +++ b/catalyst/data/bundles/base.py @@ -18,7 +18,7 @@ from itertools import count import tarfile from time import time, sleep -from abc import abstractmethod +from abc import abstractmethod, abstractproperty import logbook import pandas as pd @@ -78,6 +78,14 @@ class BaseBundle(object): def wait_time(self): raise NotImplementedError() + @abstractproperty + def splits(self): + raise NotImplementedError() + + @abstractproperty + def dividends(self): + raise NotImplementedError() + @abstractmethod def fetch_raw_metadata_frame(self, api_key, page_number): raise NotImplementedError() @@ -185,7 +193,21 @@ class BaseBundle(object): # contains an appropriately initialized file structure. We don't # forsee a usecase for adjustments at this time, but may later # choose to expose this functionality in the future. - adjustment_writer.write() + if len(self.splits) > 0 or len(self.dividends) > 0: + adjustment_writer.write( + splits=( + pd.concat(self.splits, ignore_index=True) + #if self.splits is not None \ + #and len(self.splits) > 0 else + #None + ), + dividends=( + pd.concat(self.dividends, ignore_index=True) + #if self.dividends is not None \ + #and len(dividends) > 0 else + #None + ), + ) else: # Otherwise, user has instructed to download and untar bundle # directly from the bundles `tar_url`. @@ -246,9 +268,16 @@ class BaseBundle(object): page_number, ) break - except ValueError: + except ValueError as e: raw = pd.DataFrame([]) break + except Exception as e: + log.exception( + 'Failed to load metadata from {}. ' + 'Retrying.'.format( + name=self.name, + ) + ) else: raise ValueError( 'Failed to download metadata page %d after %d ' @@ -297,6 +326,7 @@ class BaseBundle(object): # Perform and require post-processing of metadata. final_symbol_metadata = self.post_process_symbol_metadata( + asset_id, metadata.iloc[asset_id], raw_data, ) @@ -335,9 +365,11 @@ class BaseBundle(object): api_key, cache, symbol, + calendar, start_session, end_session, data_frequency, + retries, ) # TODO(cfromknecht) further data validation? @@ -359,9 +391,11 @@ class BaseBundle(object): api_key, cache, symbol, + calendar, start_session, end_session, - data_frequency): + data_frequency, + retries): # Attempt to load pre-existing symbol data from cache. try: @@ -371,54 +405,68 @@ class BaseBundle(object): # Select the most recent date in cached dataset if it exists, # otherwise use the provided `start_session`. - last = ( - raw_data.index[-1].tz_localize('UTC') - if raw_data is not None and not raw_data.empty else - start_session - ) + last = start_session + if raw_data is not None and len(raw_data) > 0: + last = raw_data.index[-1].tz_localize('UTC') + + should_sleep = False # Determine time at which cached data will be considered stale. - cache_expiration = last + pd.Timedelta(minutes=5) - if start_time <= cache_expiration: + cache_expiration = last + pd.Timedelta(days=2) + if start_time <= cache_expiration and raw_data is not None: # Data is fresh enough to reuse, no need to update. Iterator can # proceed to next symbol directly since no API call was required. - should_sleep = False + return raw_data, should_sleep + + # Data for symbol is old enough to attempt an update or is not + # present in the cache. Fetch raw data for a single symbol + # with requested intervals and frequency. Retry as necessary. + for _ in range(retries): + try: + raw_data = self.fetch_raw_symbol_frame( + api_key, + symbol, + calendar, + start_session, + end_session, + data_frequency, + ) + raw_data.index = pd.to_datetime(raw_data.index, utc=True) + + # Filter incoming data to fit start and end sessions. + raw_data = raw_data[ + (raw_data.index >= start_session) & + (raw_data.index <= end_session) + ] + + # Filter out any duplicates entries, keep last one, since + # previous frame is probably an incomplete. + raw_data = raw_data[~raw_data.index.duplicated(keep='last')] + + # Cache latest symbol data. + cache[symbol] = raw_data + + # If we arrive here, we must have attempted an API call. + # This flag tells the iterator to pause before starting the next + # asset, that we don't exceed the data source's rate limit. + should_sleep = True + + return raw_data, should_sleep + + except Exception as e: + log.exception( + 'Exception raised fetching {name} data. Retrying.' + .format(name=self.name) + ) else: - # Data for symbol is old enough to attempt an update or is not - # present in the cache. Fetch raw data for a single symbol - # with requested intervals and frequency. - raw_diff = self.fetch_raw_symbol_frame( - api_key, - symbol, - last, - end_session, - data_frequency, + raise ValueError( + 'Failed to download data for symbol {sym} ' + 'after {n} attempts.'.format( + sym=symbol, + n=retries, + ) ) - # Filter incoming data to minimize overlap. - raw_diff = raw_diff[ - (raw_diff.index >= last) & - (raw_diff.index <= end_session) - ] - - # Append incoming data to cached data if it exists, - # otherwise treat incoming data as the entire raw dataset. - raw_data = cache[symbol] = ( - raw_data.append(raw_diff) - if raw_data is not None else - raw_diff - ) - - # Filter out any duplicates entries, keep last one as previous - # one was probably an incomplete frame. - raw_data = raw_data[~raw_data.index.duplicated(keep='last')] - - # If we arrive here, we must have attempted an API call. - # This flag tells the iterator to pause before starting the next - # asset, that we don't exceed the data source's rate limit. - should_sleep = True - - return raw_data, should_sleep def _write_symbol_for_freq(self, pricing_iter, diff --git a/catalyst/data/bundles/base_pricing.py b/catalyst/data/bundles/base_pricing.py index 8ddc95da..86d1f042 100644 --- a/catalyst/data/bundles/base_pricing.py +++ b/catalyst/data/bundles/base_pricing.py @@ -38,6 +38,9 @@ class BasePricingBundle(BaseBundle): ] class BaseCryptoPricingBundle(BasePricingBundle): + def __init__(self): + super(BasePricingBundle, self).__init__() + @lazyval def calendar_name(self): return 'OPEN' @@ -46,7 +49,20 @@ class BaseCryptoPricingBundle(BasePricingBundle): def minutes_per_day(self): return 1440 + @property + def splits(self): + return [] + + @property + def dividends(self): + return [] + class BaseEquityPricingBundle(BasePricingBundle): + def __init__(self): + super(BasePricingBundle, self).__init__() + self._splits = [] + self._dividends = [] + @lazyval def calendar_name(self): return 'NYSE' @@ -54,3 +70,12 @@ class BaseEquityPricingBundle(BasePricingBundle): @lazyval def minutes_per_day(self): return 390 + + + @property + def splits(self): + return self._splits + + @property + def dividends(self): + return self._dividends diff --git a/catalyst/data/bundles/poloniex.py b/catalyst/data/bundles/poloniex.py index 2d6b9ebd..298ba03a 100644 --- a/catalyst/data/bundles/poloniex.py +++ b/catalyst/data/bundles/poloniex.py @@ -67,15 +67,17 @@ class PoloniexBundle(BaseCryptoPricingBundle): inplace=True, ) + raw = raw[raw['isFrozen'] == 0] + return raw - def post_process_symbol_metadata(self, metadata, data): - start_date = data.index[0].tz_localize(None) - end_date = data.index[-1].tz_localize(None) + def post_process_symbol_metadata(self, asset_id, sym_md, sym_data): + start_date = sym_data.index[0].tz_localize(None) + end_date = sym_data.index[-1].tz_localize(None) ac_date = end_date + pd.Timedelta(days=1) return ( - metadata.symbol, + sym_md.symbol, start_date, end_date, ac_date, @@ -84,6 +86,7 @@ class PoloniexBundle(BaseCryptoPricingBundle): def fetch_raw_symbol_frame(self, api_key, symbol, + calendar, start_date, end_date, frequency): @@ -130,7 +133,6 @@ class PoloniexBundle(BaseCryptoPricingBundle): period_map = { 'daily': 86400, '5-minute': 300, - 'minute': 60, } try: diff --git a/catalyst/data/bundles/quandl.py b/catalyst/data/bundles/quandl.py index 5a3a9dae..fc7a40c7 100644 --- a/catalyst/data/bundles/quandl.py +++ b/catalyst/data/bundles/quandl.py @@ -1,3 +1,28 @@ +# +# Copyright 2017 Enigma MPC, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from datetime import datetime + +import pandas as pd + +from six.moves.urllib.parse import urlencode + +from catalyst.data.bundles.core import register_bundle +from catalyst.data.bundles.base_pricing import BaseEquityPricingBundle +from catalyst.utils.memoize import lazyval + """ Module for building a complete daily dataset from Quandl's WIKI dataset. """ @@ -17,350 +42,190 @@ from . import core as bundles log = Logger(__name__) seconds_per_call = (pd.Timedelta('10 minutes') / 2000).total_seconds() -# Invalid symbols that quandl has had in its metadata: -excluded_symbols = frozenset({'TEST123456789'}) +class QuandlBundle(BaseEquityPricingBundle): + @lazyval + def name(self): + return 'quandl' -def _fetch_raw_metadata(api_key, cache, retries, environ): - """Generator that yields each page of data from the metadata endpoint - as a dataframe. - """ - for page_number in count(1): - key = 'metadata-page-%d' % page_number - try: - raw = cache[key] - except KeyError: - for _ in range(retries): - try: - raw = pd.read_csv( - format_metadata_url(api_key, page_number), - date_parser=pd.tseries.tools.to_datetime, - parse_dates=[ - 'oldest_available_date', - 'newest_available_date', - ], - dtypes={ - 'dataset_code': 'int', - 'name': 'str', - 'oldest_available_date': 'str', - 'newest_available_date': 'str', - }, - usecols=[ - 'dataset_code', - 'name', - 'oldest_available_date', - 'newest_available_date', - ], - ) - break - except ValueError: - # when we are past the last page we will get a value - # error because there will be no columns - raw = pd.DataFrame([]) - break - except Exception: - pass - else: - raise ValueError( - 'Failed to download metadata page %d after %d' - ' attempts.' % (page_number, retries), - ) + @lazyval + def exchange(self): + return 'QUANDL' - cache[key] = raw + @lazyval + def frequencies(self): + return set(('daily',)) - if raw.empty: - # use the empty dataframe to signal completion - break - yield raw + @lazyval + def tar_url(self): + return 'https://s3.amazonaws.com/quantopian-public-zipline-data/quandl' + @lazyval + def wait_time(self): + return pd.Timedelta(milliseconds=300) -def fetch_symbol_metadata_frame(api_key, - cache, - retries=5, - environ=None, - show_progress=False): - """ - Download Quandl symbol metadata. + @lazyval + def _excluded_symbols(self): + """ + Invalid symbols that quandl has had in its metadata: + """ + return frozenset({'TEST123456789'}) - Parameters - ---------- - api_key : str - The quandl api key to use. If this is None then no api key will be - sent. - cache : DataFrameCache - The cache to use for persisting the intermediate data. - retries : int, optional - The number of times to retry each request before failing. - environ : mapping[str -> str], optional - The environment to use to find the catalyst home. By default this - is ``os.environ``. - show_progress : bool, optional - Show a progress bar for the download of this data. - - Returns - ------- - metadata_frame : pd.DataFrame - A dataframe with the following columns: - symbol: the asset's symbol - name: the full name of the asset - start_date: the first date of data for this asset - end_date: the last date of data for this asset - auto_close_date: end_date + one day - exchange: the exchange for the asset; this is always 'quandl' - The index of the dataframe will be used for symbol->sid mappings but - otherwise does not have specific meaning. - """ - raw_iter = _fetch_raw_metadata(api_key, cache, retries, environ) - - def item_show_func(_, _it=iter(count())): - 'Downloading page: %d' % next(_it) - - with maybe_show_progress(raw_iter, - show_progress, - item_show_func=item_show_func, - label='Downloading WIKI metadata: ') as blocks: - data = pd.concat(blocks, ignore_index=True).rename(columns={ - 'dataset_code': 'symbol', - 'name': 'asset_name', - 'oldest_available_date': 'start_date', - 'newest_available_date': 'end_date', - }).sort_values('symbol') - - data = data[~data.symbol.isin(excluded_symbols)] - # cut out all the other stuff in the name column - # we need to escape the paren because it is actually splitting on a regex - data.asset_name = data.asset_name.str.split(r' \(', 1).str.get(0) - data['exchange'] = 'QUANDL' - - data['start_date'] = data['start_date'].astype(datetime) - data['end_date'] = data['end_date'].astype(datetime) - - data['auto_close_date'] = data['end_date'] + pd.Timedelta(days=1) - return data - - -def format_metadata_url(api_key, page_number): - """Build the query RL for the quandl WIKI metadata. - """ - query_params = [ - ('per_page', '100'), - ('sort_by', 'id'), - ('page', str(page_number)), - ('database_code', 'WIKI'), - ] - if api_key is not None: - query_params = [('api_key', api_key)] + query_params - return ( - 'https://www.quandl.com/api/v3/datasets.csv?' + urlencode(query_params) - ) - - -def format_wiki_url(api_key, symbol, start_date, end_date): - """ - Build a query URL for a quandl WIKI dataset. - """ - query_params = [ - ('start_date', start_date.strftime('%Y-%m-%d')), - ('end_date', end_date.strftime('%Y-%m-%d')), - ('order', 'asc'), - ] - if api_key is not None: - query_params = [('api_key', api_key)] + query_params - - return ( - "https://www.quandl.com/api/v3/datasets/WIKI/" - "{symbol}.csv?{query}".format( - symbol=symbol, - query=urlencode(query_params), - ) - ) - - -def fetch_single_equity(api_key, - symbol, - start_date, - end_date, - retries=5): - """ - Download data for a single equity. - """ - for _ in range(retries): - try: - return pd.read_csv( - format_wiki_url(api_key, symbol, start_date, end_date), - parse_dates=['Date'], - index_col='Date', - usecols=[ - 'Open', - 'High', - 'Low', - 'Close', - 'Volume', - 'Date', - 'Ex-Dividend', - 'Split Ratio', - ], - na_values=['NA'], - ).rename(columns={ - 'Open': 'open', - 'High': 'high', - 'Low': 'low', - 'Close': 'close', - 'Volume': 'volume', - 'Date': 'date', - 'Ex-Dividend': 'ex_dividend', - 'Split Ratio': 'split_ratio', - }) - except Exception: - log.exception("Exception raised reading Quandl data. Retrying.") - else: - raise ValueError( - "Failed to download data for %r after %d attempts." % ( - symbol, retries - ) + def fetch_raw_metadata_frame(self, api_key, page_number): + raw = pd.read_csv( + self._format_metadata_url(api_key, page_number), + date_parser=pd.tseries.tools.to_datetime, + parse_dates=[ + 'oldest_available_date', + 'newest_available_date', + ], + dtype={ + 'dataset_code': 'str', + 'name': 'str', + 'oldest_available_date': 'str', + 'newest_available_date': 'str', + }, + usecols=[ + 'dataset_code', + 'name', + 'oldest_available_date', + 'newest_available_date', + ], + ).rename( + columns={ + 'dataset_code': 'symbol', + 'name': 'asset_name', + 'oldest_available_date': 'start_date', + 'newest_available_date': 'end_date', + }, ) + raw['start_date'] = raw['start_date'].astype(datetime) + raw['end_date'] = raw['end_date'].astype(datetime) + raw['ac_date'] = raw['end_date'] + pd.Timedelta(days=1) -def _update_splits(splits, asset_id, raw_data): - split_ratios = raw_data.split_ratio - df = pd.DataFrame({'ratio': 1 / split_ratios[split_ratios != 1]}) - df.index.name = 'effective_date' - df.reset_index(inplace=True) - df['sid'] = asset_id - splits.append(df) + # Filter out invalid symbols + raw = raw[~raw.symbol.isin(self._excluded_symbols)] + # cut out all the other stuff in the name column + # we need to escape the paren because it is actually splitting on a regex + raw.asset_name = raw.asset_name.str.split(r' \(', 1).str.get(0) -def _update_dividends(dividends, asset_id, raw_data): - divs = raw_data.ex_dividend - df = pd.DataFrame({'amount': divs[divs != 0]}) - df.index.name = 'ex_date' - df.reset_index(inplace=True) - df['sid'] = asset_id - # we do not have this data in the WIKI dataset - df['record_date'] = df['declared_date'] = df['pay_date'] = pd.NaT - dividends.append(df) + return raw - -def gen_symbol_data(api_key, - cache, - symbol_map, - calendar, - start_session, - end_session, - splits, - dividends, - retries): - for asset_id, symbol in symbol_map.iteritems(): - start_time = time() - try: - # see if we have this data cached. - raw_data = cache[symbol] - should_sleep = False - except KeyError: - # we need to fetch the data and then write it to our cache - raw_data = cache[symbol] = fetch_single_equity( + def fetch_raw_symbol_frame(self, + api_key, + symbol, + calendar, + start_session, + end_session, + data_frequency): + raw_data = pd.read_csv( + self._format_wiki_url( api_key, symbol, - start_date=start_session, - end_date=end_session, - ) - should_sleep = True - - _update_splits(splits, asset_id, raw_data) - _update_dividends(dividends, asset_id, raw_data) + start_session, + end_session, + data_frequency, + ), + parse_dates=['Date'], + index_col='Date', + usecols=[ + 'Open', + 'High', + 'Low', + 'Close', + 'Volume', + 'Date', + 'Ex-Dividend', + 'Split Ratio', + ], + na_values=['NA'], + ).rename(columns={ + 'Open': 'open', + 'High': 'high', + 'Low': 'low', + 'Close': 'close', + 'Volume': 'volume', + 'Date': 'date', + 'Ex-Dividend': 'ex_dividend', + 'Split Ratio': 'split_ratio', + }) sessions = calendar.sessions_in_range(start_session, end_session) - raw_data = raw_data.reindex( + return raw_data.reindex( sessions.tz_localize(None), copy=False, ).fillna(0.0) - yield asset_id, raw_data - if should_sleep: - remaining = seconds_per_call - time() - start_time - if remaining > 0: - sleep(remaining) + def post_process_symbol_metadata(self, asset_id, sym_md, sym_data): + self._update_splits(asset_id, sym_data) + self._update_dividends(asset_id, sym_data) + + return sym_md + + def _update_splits(self, asset_id, raw_data): + split_ratios = raw_data.split_ratio + df = pd.DataFrame({'ratio': 1 / split_ratios[split_ratios != 1]}) + df.index.name = 'effective_date' + df.reset_index(inplace=True) + df['sid'] = asset_id + self.splits.append(df) -@bundles.register('quandl') -def quandl_bundle(environ, - asset_db_writer, - minute_bar_writer, - daily_bar_writer, - adjustment_writer, - calendar, - start_session, - end_session, - cache, - show_progress, - output_dir): - """Build a catalyst data bundle from the Quandl WIKI dataset. - """ - api_key = environ.get('QUANDL_API_KEY') - metadata = fetch_symbol_metadata_frame( - api_key, - cache=cache, - show_progress=show_progress, - ) - symbol_map = metadata.symbol - - # data we will collect in `gen_symbol_data` - splits = [] - dividends = [] - - asset_db_writer.write(metadata) - daily_bar_writer.write( - gen_symbol_data( - api_key, - cache, - symbol_map, - calendar, - start_session, - end_session, - splits, - dividends, - environ.get('QUANDL_DOWNLOAD_ATTEMPTS', 5), - ), - assets=metadata.index, - show_progress=show_progress, - ) - adjustment_writer.write( - splits=pd.concat(splits, ignore_index=True), - dividends=pd.concat(dividends, ignore_index=True), - ) + def _update_dividends(self, asset_id, raw_data): + divs = raw_data.ex_dividend + df = pd.DataFrame({'amount': divs[divs != 0]}) + df.index.name = 'ex_date' + df.reset_index(inplace=True) + df['sid'] = asset_id + # we do not have this data in the WIKI dataset + df['record_date'] = df['declared_date'] = df['pay_date'] = pd.NaT + self.dividends.append(df) -QUANTOPIAN_QUANDL_URL = ( - 'https://s3.amazonaws.com/quantopian-public-zipline-data/quandl' -) + def _format_metadata_url(self, api_key, page_number): + """Build the query RL for the quandl WIKI metadata. + """ + query_params = [ + ('per_page', '100'), + ('sort_by', 'id'), + ('page', str(page_number)), + ('database_code', 'WIKI'), + ] + if api_key is not None: + query_params = [('api_key', api_key)] + query_params - -@bundles.register('quantopian-quandl', create_writers=False) -def quantopian_quandl_bundle(environ, - asset_db_writer, - minute_bar_writer, - daily_bar_writer, - adjustment_writer, - calendar, - start_session, - end_session, - cache, - show_progress, - output_dir): - if show_progress: - data = bundles.download_with_progress( - QUANTOPIAN_QUANDL_URL, - chunk_size=bundles.ONE_MEGABYTE, - label="Downloading Bundle: quantopian-quandl", + return ( + 'https://www.quandl.com/api/v3/datasets.csv?' + urlencode(query_params) ) - else: - data = bundles.download_without_progress(QUANTOPIAN_QUANDL_URL) - - with tarfile.open('r', fileobj=data) as tar: - if show_progress: - print("Writing data to %s." % output_dir) - tar.extractall(output_dir) -register_calendar_alias("QUANDL", "NYSE") + def _format_wiki_url(self, + api_key, + symbol, + start_date, + end_date, + data_frequency): + """ + Build a query URL for a quandl WIKI dataset. + """ + query_params = [ + ('start_date', start_date.strftime('%Y-%m-%d')), + ('end_date', end_date.strftime('%Y-%m-%d')), + ('order', 'asc'), + ] + if api_key is not None: + query_params = [('api_key', api_key)] + query_params + + return ( + "https://www.quandl.com/api/v3/datasets/WIKI/" + "{symbol}.csv?{query}".format( + symbol=symbol, + query=urlencode(query_params), + ) + ) + +register_calendar_alias('QUANDL', 'NYSE') +register_bundle(QuandlBundle)