From 24d26f9e63e0581e3b34b37479fb04018d955b4d Mon Sep 17 00:00:00 2001 From: Scott Sanderson Date: Thu, 22 Oct 2015 08:59:48 -0400 Subject: [PATCH] MAINT: Rewrite the benchmark loader. --- zipline/data/benchmarks.py | 143 +++++-------------- zipline/data/loader.py | 275 ++++++++++++++++++------------------- 2 files changed, 174 insertions(+), 244 deletions(-) diff --git a/zipline/data/benchmarks.py b/zipline/data/benchmarks.py index 9f89f47f..d3620bce 100644 --- a/zipline/data/benchmarks.py +++ b/zipline/data/benchmarks.py @@ -12,125 +12,56 @@ # 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. -import collections - -from datetime import datetime - -import csv - -from functools import partial - -import requests import pandas as pd -from six import iteritems - -from . loader_utils import ( - date_conversion, - source_to_records, - Mapping -) - -DailyReturn = collections.namedtuple('DailyReturn', ['date', 'returns']) +from six.moves.urllib_parse import urlencode -class BenchmarkDataNotFoundError(Exception): - pass - -_BENCHMARK_MAPPING = { - # Need to add 'symbol' - 'volume': (int, 'Volume'), - 'open': (float, 'Open'), - 'close': (float, 'Close'), - 'high': (float, 'High'), - 'low': (float, 'Low'), - 'adj_close': (float, 'Adj Close'), - 'date': (partial(date_conversion, date_pattern='%Y-%m-%d'), 'Date') -} - - -def benchmark_mappings(): - return {key: Mapping(*value) - for key, value - in iteritems(_BENCHMARK_MAPPING)} - - -def get_raw_benchmark_data(start_date, end_date, symbol): - - # create benchmark files - # ^GSPC 19500103 - params = collections.OrderedDict(( - ('s', symbol), - # start_date month, zero indexed - ('a', start_date.month - 1), - # start_date day - ('b', start_date.day), - # start_date year - ('c', start_date.year), - # end_date month, zero indexed - ('d', end_date.month - 1), - # end_date day str(int(todate[6:8])) #day - ('e', end_date.day), - # end_date year str(int(todate[0:4])) - ('f', end_date.year), - # daily frequency - ('g', 'd'), - )) - - res = requests.get('http://ichart.finance.yahoo.com/table.csv', - params=params, stream=True) - - if not res.ok: - raise BenchmarkDataNotFoundError(""" -No benchmark data found for date range. -start_date={start_date}, end_date={end_date}, url={url}""".strip(). - format(start_date=start_date, - end_date=end_date, - url=res.url)) - - return csv.DictReader(res.text.splitlines()) - - -def get_benchmark_data(symbol, start_date=None, end_date=None): +def format_yahoo_index_url(symbol, start_date, end_date): """ - Benchmarks from Yahoo. + Format a URL for querying Yahoo Finance for Index data. """ - if start_date is None: - start_date = datetime(year=1950, month=1, day=3) - if end_date is None: - end_date = datetime.utcnow() - - raw_benchmark_data = get_raw_benchmark_data(start_date, end_date, symbol) - - mappings = benchmark_mappings() - - return source_to_records(mappings, raw_benchmark_data) + return ( + 'http://ichart.finance.yahoo.com/table.csv?' + urlencode({ + 's': symbol, + # start_date month, zero indexed + 'a': start_date.month - 1, + # start_date day + 'b': start_date.day, + # start_date year + 'c': start_date.year, + # end_date month, zero indexed + 'd': end_date.month - 1, + # end_date day + 'e': end_date.day, + # end_date year + 'f': end_date.year, + # daily frequency + 'g': 'd', + }) + ) def get_benchmark_returns(symbol, start_date=None, end_date=None): """ - Returns a list of return percentages in chronological order. + Get a Series of benchmark returns from Yahoo. """ if start_date is None: - start_date = datetime(year=1950, month=1, day=3) + start_date = pd.Timestamp(0, tz='UTC') if end_date is None: - end_date = datetime.utcnow() + end_date = pd.Timestamp('now', tz='UTC') - # Get the benchmark data and convert it to a list in chronological order. - data_points = list(get_benchmark_data(symbol, start_date, end_date)) - data_points.reverse() + data = pd.read_csv( + format_yahoo_index_url(symbol, start_date, end_date), + parse_dates=['Date'], + index_col='Date', + usecols=["Open", "Close", "Date"], + ).sort_index().tz_localize('UTC') - # Calculate the return percentages. - benchmark_returns = [] - for i, data_point in enumerate(data_points): - if i == 0: - curr_open = data_points[i]['open'] - returns = (data_points[i]['close'] - curr_open) / curr_open - else: - prev_close = data_points[i - 1]['close'] - returns = (data_point['close'] - prev_close) / prev_close - date = pd.tseries.tools.normalize_date(data_point['date']) - daily_return = DailyReturn(date=date, returns=returns) - benchmark_returns.append(daily_return) + returns = data["Close"].pct_change() + # Calculate the returns for the first day using the open of that day since + # we don't have the close of the previous day. + first_open, first_close = data.ix[0, ["Open", "Close"]] + returns.iloc[0] = (first_close - first_open) / first_open - return benchmark_returns + return returns diff --git a/zipline/data/loader.py b/zipline/data/loader.py index 56ad7a84..5f6c2599 100644 --- a/zipline/data/loader.py +++ b/zipline/data/loader.py @@ -17,7 +17,6 @@ import importlib import os from collections import OrderedDict -from datetime import timedelta import logbook @@ -27,15 +26,16 @@ import pytz from six import iteritems -from . import benchmarks from . benchmarks import get_benchmark_returns from .paths import ( cache_root, data_root, ) -from zipline.utils.tradingcalendar import trading_day as trading_day_nyse -from zipline.utils.tradingcalendar import trading_days as trading_days_nyse +from zipline.utils.tradingcalendar import ( + trading_day as trading_day_nyse, + trading_days as trading_days_nyse, +) logger = logbook.Logger('Loader') @@ -72,148 +72,147 @@ def get_cache_filepath(name): return os.path.join(cr, name) -def dump_treasury_curves(module_name, filename): - """ - Dumps data to be used with zipline. +def get_benchmark_filename(symbol): + return "%s_benchmark.csv" % symbol - Puts source treasury and data into zipline. + +def has_data_for_dates(series_or_df, first_date, last_date): """ + Does `series_or_df` have data on or before first_date and on or after + last_date? + """ + dts = series_or_df.index + if not isinstance(dts, pd.DatetimeIndex): + raise TypeError("Expected a DatetimeIndex, but got %s." % type(dts)) + first, last = dts[[0, -1]] + return (first <= first_date) and (last >= last_date) + + +def load_market_data(trading_day=trading_day_nyse, + trading_days=trading_days_nyse, bm_symbol='^GSPC'): + first_date = trading_days[0] + + # We expect to have benchmark and treasury data that's current up until + # two full trading days prior to the most recently completed trading day. + # Example: + # On Thu Oct 22 2015, the previous completed trading day is Wed Oct 21. + # However, data for Oct 21 doesn't become available until the early morning + # hours of Oct 22. This means that there are times on the 22nd at which we + # cannot reasonably expect to have data for the 21st available. To be + # conservative, we instead expect that at any time on the 22nd, we can + # download data for Tuesday the 20th, which is two full trading days prior + # to the date on which we're running a test. + + # We'll attempt to download new data if the latest entry in our cache is + # before this date. + last_date = ( + pd.Timestamp('now', tz='UTC').normalize() - (2 * trading_day) + ) + benchmark_returns = ensure_benchmark_data( + bm_symbol, + first_date, + last_date, + ) + treasury_curves = ensure_treasury_data( + bm_symbol, + first_date, + last_date, + ) + return benchmark_returns, treasury_curves + + +def ensure_benchmark_data(symbol, first_date, last_date): + """ + Ensure we have benchmark data for `symbol` from `first_date` to `last_date` + + Parameters + ---------- + symbol : str + The symbol for the benchmark to load. + first_date : pd.Timestamp + First required date for the cache. + last_date : pd.Timestamp + Last required date for the cache. + + We attempt to download data unless we already have data stored at the data + cache for `symbol` whose first entry is before or on `first_date` and whose + last entry is on or after `last_date`. + """ + path = get_data_filepath(get_benchmark_filename(symbol)) + try: + data = pd.Series.from_csv(path).tz_localize('UTC') + if has_data_for_dates(data, first_date, last_date): + return data + except (OSError, IOError, ValueError) as e: + # These can all be raised by various versions of pandas on various + # classes of malformed input. Treat them all as cache misses. + logger.info( + "Loading data for {path} failed with error [{error}].".format( + path=path, error=e, + ) + ) + logger.info( + "Cache at {path} does not have data from {start} to {end}.\n" + "Downloading benchmark data for '{symbol}'.", + start=first_date, + end=last_date, + symbol=symbol, + path=path, + ) + + data = get_benchmark_returns(symbol, first_date, last_date) + data.to_csv(path) + if not has_data_for_dates(data, first_date, last_date): + logger.warn("Still don't have expected data after redownload!") + return data + + +def ensure_treasury_data(bm_symbol, first_date, last_date): + """ + Ensure we have treasury data from treasury module associated with + `bm_symbol`. + + Parameters + ---------- + bm_symbol : str + Benchmark symbol for which we're loading associated treasury curves. + first_date : pd.Timestamp + First date required to be in the cache. + last_date : pd.Timestamp + Last date required to be in the cache. + + We attempt to download data unless we already have data stored in the cache + for `module_name` whose first entry is before or on `first_date` and whose + last entry is on or after `last_date`. + """ + module_name, filename, source = INDEX_MAPPING.get( + bm_symbol, INDEX_MAPPING['^GSPC'] + ) + path = get_data_filepath(filename) + try: + data = pd.DataFrame.from_csv(path).tz_localize('UTC') + if has_data_for_dates(data, first_date, last_date): + return data + except (OSError, IOError, ValueError) as e: + # These can all be raised by various versions of pandas on various + # classes of malformed input. Treat them all as cache misses. + logger.info( + "Loading data for {path} failed with error [{error}].".format( + path=path, error=e, + ) + ) + try: m = importlib.import_module("." + module_name, package='zipline.data') except ImportError: raise NotImplementedError( 'Treasury curve {0} module not implemented'.format(module_name)) - curves = m.get_treasury_data() - - data_filepath = get_data_filepath(filename) - curves.to_csv(data_filepath) - return curves - - -def dump_benchmarks(symbol): - """ - Dumps data to be used with zipline. - - Puts source treasury and data into zipline. - """ - benchmark_data = [] - for daily_return in get_benchmark_returns(symbol): - # Not ideal but massaging data into expected format - benchmark = (daily_return.date, daily_return.returns) - benchmark_data.append(benchmark) - - data_filepath = get_data_filepath(get_benchmark_filename(symbol)) - benchmark_returns = pd.Series(dict(benchmark_data)) - benchmark_returns.to_csv(data_filepath) - - -def update_benchmarks(symbol, last_date): - """ - Updates data in the zipline message pack - - last_date should be a datetime object of the most recent data - - Puts source benchmark into zipline. - """ - datafile = get_data_filepath(get_benchmark_filename(symbol)) - saved_benchmarks = pd.Series.from_csv(datafile) - - try: - start = last_date + timedelta(days=1) - for daily_return in get_benchmark_returns(symbol, start_date=start): - # Not ideal but massaging data into expected format - benchmark = pd.Series({daily_return.date: daily_return.returns}) - saved_benchmarks = saved_benchmarks.append(benchmark) - - datafile = get_data_filepath(get_benchmark_filename(symbol)) - saved_benchmarks.to_csv(datafile) - except benchmarks.BenchmarkDataNotFoundError as exc: - logger.warn(exc) - return saved_benchmarks - - -def get_benchmark_filename(symbol): - return "%s_benchmark.csv" % symbol - - -def load_market_data(trading_day=trading_day_nyse, - trading_days=trading_days_nyse, bm_symbol='^GSPC'): - bm_filepath = get_data_filepath(get_benchmark_filename(bm_symbol)) - try: - saved_benchmarks = pd.Series.from_csv(bm_filepath) - except (OSError, IOError, ValueError): - logger.info( - "No cache found at {path}. " - "Downloading benchmark data for '{symbol}'.", - symbol=bm_symbol, - path=bm_filepath, - ) - - dump_benchmarks(bm_symbol) - saved_benchmarks = pd.Series.from_csv(bm_filepath) - - saved_benchmarks = saved_benchmarks.tz_localize('UTC') - - most_recent = pd.Timestamp('today', tz='UTC') - trading_day - most_recent_index = trading_days.searchsorted(most_recent) - days_up_to_now = trading_days[:most_recent_index + 1] - - # Find the offset of the last date for which we have trading data in our - # list of valid trading days - last_bm_date = saved_benchmarks.index[-1] - last_bm_date_offset = days_up_to_now.searchsorted(last_bm_date) - - # If more than 1 trading days has elapsed since the last day where - # we have data,then we need to update - # We're doing "> 2" rather than "> 1" because we're subtracting an array - # _length_ from an array _index_, and therefore even if we had data up to - # and including the current day, the difference would still be 1. - if len(days_up_to_now) - last_bm_date_offset > 2: - benchmark_returns = update_benchmarks(bm_symbol, last_bm_date) - if benchmark_returns.index.tz is None or \ - benchmark_returns.index.tz.zone != 'UTC': - benchmark_returns = benchmark_returns.tz_localize('UTC') - else: - benchmark_returns = saved_benchmarks - if benchmark_returns.index.tz is None or\ - benchmark_returns.index.tz.zone != 'UTC': - benchmark_returns = benchmark_returns.tz_localize('UTC') - - # Get treasury curve module, filename & source from mapping. - # Default to USA. - module, filename, source = INDEX_MAPPING.get( - bm_symbol, INDEX_MAPPING['^GSPC']) - - tr_filepath = get_data_filepath(filename) - try: - saved_curves = pd.DataFrame.from_csv(tr_filepath) - except (OSError, IOError, ValueError): - logger.info( - "No cache found at {path}. " - "Downloading treasury data from {source}.", - path=tr_filepath, - source=source, - ) - - dump_treasury_curves(module, filename) - saved_curves = pd.DataFrame.from_csv(tr_filepath) - - # Find the offset of the last date for which we have trading data in our - # list of valid trading days - last_tr_date = saved_curves.index[-1] - last_tr_date_offset = days_up_to_now.searchsorted( - last_tr_date.strftime('%Y/%m/%d')) - - # If more than 1 trading days has elapsed since the last day where - # we have data,then we need to update - # Comment above explains why this is "> 2". - if len(days_up_to_now) - last_tr_date_offset > 2: - treasury_curves = dump_treasury_curves(module, filename) - else: - treasury_curves = saved_curves.tz_localize('UTC') - - return benchmark_returns, treasury_curves + data = m.get_treasury_data() + data.to_csv(path) + if not has_data_for_dates(data, first_date, last_date): + logger.warn("Still don't have expected data after redownload!") + return data def _load_raw_yahoo_data(indexes=None, stocks=None, start=None, end=None):