diff --git a/zipline/data/benchmarks.py b/zipline/data/benchmarks.py index 9f89f47f..ee2da2ae 100644 --- a/zipline/data/benchmarks.py +++ b/zipline/data/benchmarks.py @@ -12,125 +12,50 @@ # 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): +def get_benchmark_returns(symbol, start_date, end_date): """ - Returns a list of return percentages in chronological order. + Get a Series of benchmark returns from Yahoo. + + Returns a Series with returns from (start_date, end_date]. + + start_date is **not** included because we need the close from day N - 1 to + compute the returns for day N. """ - if start_date is None: - start_date = datetime(year=1950, month=1, day=3) - if end_date is None: - end_date = datetime.utcnow() - - # 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() - - # 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) - - return benchmark_returns + return pd.read_csv( + format_yahoo_index_url(symbol, start_date, end_date), + parse_dates=['Date'], + index_col='Date', + usecols=["Adj Close", "Date"], + squeeze=True, # squeeze tells pandas to make this a Series + # instead of a 1-column DataFrame + ).sort_index().tz_localize('UTC').pct_change(1).iloc[1:] diff --git a/zipline/data/loader.py b/zipline/data/loader.py index 8a001f4e..2c42759a 100644 --- a/zipline/data/loader.py +++ b/zipline/data/loader.py @@ -12,12 +12,8 @@ # 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 importlib import os from collections import OrderedDict -from datetime import timedelta import logbook @@ -27,26 +23,28 @@ import pytz from six import iteritems -from . import benchmarks from . benchmarks import get_benchmark_returns +from . import treasuries, treasuries_can 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') # Mapping from index symbol to appropriate bond data INDEX_MAPPING = { '^GSPC': - ('treasuries', 'treasury_curves.csv', 'data.treasury.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', 'data.treasury.gov'), + (treasuries, 'treasury_curves.csv', 'www.federalreserve.gov'), } @@ -72,158 +70,191 @@ def get_cache_filepath(name): return os.path.join(cr, name) -def dump_treasury_curves(module='treasuries', filename='treasury_curves.csv'): - """ - Dumps data to be used with zipline. - - Puts source treasury and data into zipline. - """ - try: - m = importlib.import_module("." + module, package='zipline.data') - except ImportError: - raise NotImplementedError( - 'Treasury curve {0} module not implemented'.format(module)) - - tr_data = {} - - for curve in m.get_treasury_data(): - # Not ideal but massaging data into expected format - tr_data[curve['date']] = curve - - curves = pd.DataFrame(tr_data).T - - 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 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'): - 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, - ) + trading_days=trading_days_nyse, + bm_symbol='^GSPC'): + """ + Load benchmark returns and treasury yield curves for the given calendar and + benchmark symbol. - dump_benchmarks(bm_symbol) - saved_benchmarks = pd.Series.from_csv(bm_filepath) + Benchmarks are downloaded as a Series from Yahoo Finance. Treasury curves + are US Treasury Bond rates and are downloaded from 'www.federalreserve.gov' + by default. For Canadian exchanges, a loader for Canadian bonds from the + Bank of Canada is also available. - saved_benchmarks = saved_benchmarks.tz_localize('UTC') + Results downloaded from the internet are cached in + ~/.zipline/data. Subsequent loads will attempt to read from the cached + files before falling back to redownload. - 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] + Parameters + ---------- + trading_day : pandas.CustomBusinessDay, optional + A trading_day used to determine the latest day for which we + expect to have data. Defaults to an NYSE trading day. + trading_days : pd.DatetimeIndex, optional + A calendar of trading days. Also used for determining what cached + dates we should expect to have cached. Defaults to the NYSE calendar. + bm_symbol : str, optional + Symbol for the benchmark index to load. Defaults to '^GSPC', the Yahoo + ticker for the S&P 500. - # 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.strftime('%Y/%m/%d')) + Returns + ------- + (benchmark_returns, treasury_curves) : (pd.Series, pd.DataFrame) - # 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') + Notes + ----- - # Get treasury curve module, filename & source from mapping. - # Default to USA. - module, filename, source = INDEX_MAPPING.get( - bm_symbol, INDEX_MAPPING['^GSPC']) + Both return values are DatetimeIndexed with values dated to midnight in UTC + of each stored date. The columns of `treasury_curves` are: - 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, - ) + '1month', '3month', '6month', + '1year','2year','3year','5year','7year','10year','20year','30year' + """ + first_date = trading_days[0] - dump_treasury_curves(module, filename) - saved_curves = pd.DataFrame.from_csv(tr_filepath) + # 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. - # 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') + # We'll attempt to download new data if the latest entry in our cache is + # before this date. + last_date = trading_days[ + trading_days.get_loc(pd.Timestamp.utcnow(), method='ffill') - 2 + ] + benchmark_returns = ensure_benchmark_data( + bm_symbol, + first_date, + last_date, + # We need the trading_day to figure out the close prior to the first + # date so that we can compute returns for the first date. + trading_day, + ) + 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, trading_day): + """ + 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. + trading_day : pd.CustomBusinessDay + A trading day delta. Used to find the day before first_date so we can + get the close of the day prior to first_date. + + 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 - trading_day, 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`. + """ + loader_module, filename, source = INDEX_MAPPING.get( + bm_symbol, INDEX_MAPPING['^GSPC'] + ) + first_date = max(first_date, loader_module.earliest_possible_date()) + 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, + ) + ) + + data = loader_module.get_treasury_data(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 _load_raw_yahoo_data(indexes=None, stocks=None, start=None, end=None): """Load closing prices from yahoo finance. diff --git a/zipline/data/loader_utils.py b/zipline/data/loader_utils.py deleted file mode 100644 index 014a95cb..00000000 --- a/zipline/data/loader_utils.py +++ /dev/null @@ -1,157 +0,0 @@ -# -# Copyright 2012 Quantopian, 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. - - -""" -Various utilites used by different date loaders. - -Could stand to be broken up more into components. -e.g. the mapping utilities. - -""" - -import datetime - -import pytz - -from collections import namedtuple - -from functools import partial - -from six import iteritems - - -def get_utc_from_exchange_time(naive): - local = pytz.timezone('US/Eastern') - local_dt = naive.replace(tzinfo=local) - utc_dt = local_dt.astimezone(pytz.utc) - return utc_dt - - -def get_exchange_time_from_utc(utc_dt): - """ - Takes in result from exchange time. - """ - dt = utc_dt.replace(tzinfo=pytz.utc) - local = pytz.timezone('US/Eastern') - dt = dt.astimezone(local) - - return dt - - -def guarded_conversion(conversion, str_val): - """ - Returns the result of applying the @conversion to @str_val - """ - if str_val in (None, ""): - return None - return conversion(str_val) - - -def safe_int(str_val): - """ - casts the @str_val to a float to handle the occassional - decimal point in int fields from data providers. - """ - f = float(str_val) - i = int(f) - return i - - -def date_conversion(date_str, date_pattern='%m/%d/%Y', to_utc=True): - """ - Convert date strings from TickData (or other source) into epoch values. - - Specify to_utc=False if the input date is already UTC (or is naive). - """ - dt = datetime.datetime.strptime(date_str, date_pattern) - if to_utc: - dt = get_utc_from_exchange_time(dt) - else: - dt = dt.replace(tzinfo=pytz.utc) - return dt - - -# Mapping is a structure for how want to convert the source data into -# the form we insert into the database. -# - conversion, a function used to convert source input to our target value -# - source, the key(s) in the original source to pass to the conversion -# method -# If a single string, then it's a direct lookup into the -# source row by that key -# If an iterator, pass the source to as a list of keys, -# in order, to the conversion function. -# If empty, then the conversion method provides a 'default' value. -Mapping = namedtuple('Mapping', ['conversion', 'source']) - - -def apply_mapping(mapping, row): - """ - Returns the value of a @mapping for a given @row. - - i.e. the @mapping.source values are extracted from @row and fed - into the @mapping.conversion method. - """ - if isinstance(mapping.source, str): - # Do a 'direct' conversion of one key from the source row. - return guarded_conversion(mapping.conversion, row[mapping.source]) - if mapping.source is None: - # For hardcoded values. - # conversion method will return a constant value - return mapping.conversion() - else: - # Assume we are using multiple source values. - # Feed the source values in order prescribed by mapping.source - # to mapping.conversion. - return mapping.conversion(*[row[source] for source in mapping.source]) - - -def _row_cb(mapping, row): - """ - Returns the dict created from our @mapping of the source @row. - - Not intended to be used directly, but rather to be the base of another - function that supplies the mapping value. - """ - return { - target: apply_mapping(mapping, row) - for target, mapping - in iteritems(mapping) - } - - -def make_row_cb(mapping): - """ - Returns a func that can be applied to a dict that returns the - application of the @mapping, which results in a dict. - """ - return partial(_row_cb, mapping) - - -def source_to_records(mappings, - source, - source_wrapper=None, - records_wrapper=None): - if source_wrapper: - source = source_wrapper(source) - - callback = make_row_cb(mappings) - - records = (callback(row) for row in source) - - if records_wrapper: - records = records_wrapper(records) - - return records diff --git a/zipline/data/treasuries.py b/zipline/data/treasuries.py index a23c34c3..65823d45 100644 --- a/zipline/data/treasuries.py +++ b/zipline/data/treasuries.py @@ -12,131 +12,75 @@ # 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 operator import itemgetter import re import numpy as np import pandas as pd -import requests - -from collections import OrderedDict -import xml.etree.ElementTree as ET - -from six import iteritems - -from . loader_utils import ( - guarded_conversion, - safe_int, - Mapping, - date_conversion, - source_to_records -) -def get_treasury_date(dstring): - return date_conversion(dstring.split("T")[0], date_pattern='%Y-%m-%d', - to_utc=False) +get_unit_and_periods = itemgetter('unit', 'periods') -def get_treasury_rate(string_val): - val = guarded_conversion(float, string_val) - if val is not None: - val = round(val / 100.0, 4) - return val - -_CURVE_MAPPINGS = { - 'tid': (safe_int, "Id"), - 'date': (get_treasury_date, "NEW_DATE"), - '1month': (get_treasury_rate, "BC_1MONTH"), - '3month': (get_treasury_rate, "BC_3MONTH"), - '6month': (get_treasury_rate, "BC_6MONTH"), - '1year': (get_treasury_rate, "BC_1YEAR"), - '2year': (get_treasury_rate, "BC_2YEAR"), - '3year': (get_treasury_rate, "BC_3YEAR"), - '5year': (get_treasury_rate, "BC_5YEAR"), - '7year': (get_treasury_rate, "BC_7YEAR"), - '10year': (get_treasury_rate, "BC_10YEAR"), - '20year': (get_treasury_rate, "BC_20YEAR"), - '30year': (get_treasury_rate, "BC_30YEAR"), -} - - -def treasury_mappings(mappings): - return {key: Mapping(*value) - for key, value - in iteritems(mappings)} - - -class iter_to_stream(object): +def parse_treasury_csv_column(column): """ - Exposes an iterable as an i/o stream + Parse a treasury CSV column into a more human-readable format. + + Columns start with 'RIFLGFC', followed by Y or M (year or month), followed + by a two-digit number signifying number of years/months, followed by _N.B. + We only care about the middle two entries, which we turn into a string like + 3month or 30year. """ - def __init__(self, iterable): - self.buffered = "" - self.iter = iter(iterable) + column_re = re.compile( + r"^(?PRIFLGFC)" + "(?P[YM])" + "(?P[0-9]{2})" + "(?P_N.B)$" + ) - def read(self, size): - result = "" - while size > 0: - data = self.buffered or next(self.iter, None) - self.buffered = "" - if data is None: - break - size -= len(data) - if size < 0: - data, self.buffered = data[:size], data[size:] - result += data - return result + match = column_re.match(column) + if match is None: + raise ValueError("Couldn't parse CSV column %r." % column) + unit, periods = get_unit_and_periods(match.groupdict()) + + # Roundtrip through int to coerce '06' into '6'. + return str(int(periods)) + ('year' if unit == 'Y' else 'month') -def get_localname(element): - qtag = ET.QName(element.tag).text - return re.match("(\{.*\})(.*)", qtag).group(2) +def earliest_possible_date(): + """ + The earliest date for which we can load data from this module. + """ + # The US Treasury actually has data going back further than this, but it's + # pretty rare to find pricing data going back that far, and there's no + # reason to make people download benchmarks back to 1950 that they'll never + # be able to use. + return pd.Timestamp('1980', tz='UTC') -def get_treasury_source(): - url = """\ -http://data.treasury.gov/feed.svc/DailyTreasuryYieldCurveRateData\ -""" - res = requests.get(url, stream=True) - stream = iter_to_stream(res.text.splitlines()) - - elements = ET.iterparse(stream, ('end', 'start-ns', 'end-ns')) - - namespaces = OrderedDict() - properties_xpath = [''] - - def updated_namespaces(): - if '' in namespaces and 'm' in namespaces: - properties_xpath[0] = "{%s}content/{%s}properties" % ( - namespaces[''], namespaces['m'] - ) - else: - properties_xpath[0] = '' - - for event, element in elements: - if event == 'end': - tag = get_localname(element) - if tag == "entry": - properties = element.find(properties_xpath[0]) - datum = {get_localname(node): node.text - for node in properties if ET.iselement(node)} - # clear the element after we've dealt with it: - element.clear() - yield datum - - elif event == 'start-ns': - namespaces[element[0]] = element[1] - updated_namespaces() - - elif event == 'end-ns': - namespaces.popitem() - updated_namespaces() - - -def get_treasury_data(): - mappings = treasury_mappings(_CURVE_MAPPINGS) - source = get_treasury_source() - return source_to_records(mappings, source) +def get_treasury_data(start_date, end_date): + return pd.read_csv( + "http://www.federalreserve.gov/datadownload/Output.aspx" + "?rel=H15" + "&series=bf17364827e38702b42a58cf8eaa3f78" + "&lastObs=" + "&from=" # An unbounded query is ~2x faster than specifying dates. + "&to=" + "&filetype=csv" + "&label=omit" + "&layout=seriescolumn" + "&type=package", + skiprows=1, # First row is a useless header. + parse_dates=['Time Period'], + na_values=['ND'], # Presumably this stands for "No Data". + index_col=0, + ).loc[ + start_date:end_date + ].dropna( + how='all' + ).rename( + columns=parse_treasury_csv_column + ).tz_localize('UTC') * 0.01 # Convert from 2.57% to 0.0257. def dataconverter(s): diff --git a/zipline/data/treasuries_can.py b/zipline/data/treasuries_can.py index 6d579aa2..fff249ad 100644 --- a/zipline/data/treasuries_can.py +++ b/zipline/data/treasuries_can.py @@ -13,113 +13,138 @@ # See the License for the specific language governing permissions and # limitations under the License. -import datetime -import requests +import pandas as pd +import six +from toolz import curry +from toolz.curried.operator import add as prepend -from . loader_utils import ( - source_to_records -) - -from zipline.data.treasuries import ( - treasury_mappings, get_treasury_date, get_treasury_rate -) - - -_CURVE_MAPPINGS = { - 'date': (get_treasury_date, "Date"), - '1month': (get_treasury_rate, "V39063"), - '3month': (get_treasury_rate, "V39065"), - '6month': (get_treasury_rate, "V39066"), - '1year': (get_treasury_rate, "V39067"), - '2year': (get_treasury_rate, "V39051"), - '3year': (get_treasury_rate, "V39052"), - '5year': (get_treasury_rate, "V39053"), - '7year': (get_treasury_rate, "V39054"), - '10year': (get_treasury_rate, "V39055"), +COLUMN_NAMES = { + "V39063": '1month', + "V39065": '3month', + "V39066": '6month', + "V39067": '1year', + "V39051": '2year', + "V39052": '3year', + "V39053": '5year', + "V39054": '7year', + "V39055": '10year', # Bank of Canada refers to this as 'Long' Rate, approximately 30 years. - '30year': (get_treasury_rate, "V39056"), + "V39056": '30year', } - -BILLS = ['V39063', 'V39065', 'V39066', 'V39067'] -BONDS = ['V39051', 'V39052', 'V39053', 'V39054', 'V39055', 'V39056'] +BILL_IDS = ['V39063', 'V39065', 'V39066', 'V39067'] +BOND_IDS = ['V39051', 'V39052', 'V39053', 'V39054', 'V39055', 'V39056'] -def get_treasury_source(start_date=None, end_date=None): - - today = datetime.date.today() - # Bank of Canada only has 10 years of data and has this in the URL. - restriction = datetime.date(today.year - 10, today.month, today.day) - - if not end_date: - end_date = today - - if not start_date: - start_date = restriction - - bill_url = ( - "http://www.bankofcanada.ca/stats/results/csv?" - "lP=lookup_tbill_yields.php&sR={restrict}&se=" - "L_V39063-L_V39065-L_V39066-L_V39067&dF={start}&dT={end}" - .format(restrict=restriction.strftime("%Y-%m-%d"), - start=start_date.strftime("%Y-%m-%d"), - end=end_date.strftime("%Y-%m-%d"), - ) +@curry +def _format_url(instrument_type, + instrument_ids, + start_date, + end_date, + earliest_allowed_date): + """ + Format a URL for loading data from Bank of Canada. + """ + return ( + "http://www.bankofcanada.ca/stats/results/csv" + "?lP=lookup_{instrument_type}_yields.php" + "&sR={restrict}" + "&se={instrument_ids}" + "&dF={start}" + "&dT={end}".format( + instrument_type=instrument_type, + instrument_ids='-'.join(map(prepend("L_"), instrument_ids)), + restrict=earliest_allowed_date.strftime("%Y-%m-%d"), + start=start_date.strftime("%Y-%m-%d"), + end=end_date.strftime("%Y-%m-%d"), + ) ) - bond_url = ( - "http://www.bankofcanada.ca/stats/results/csv?" - "lP=lookup_bond_yields.php&sR={restrict}&se=" - "L_V39051-L_V39052-L_V39053-L_V39054-L_V39055-L_V39056" - "&dF={start}&dT={end}" - .format(restrict=restriction.strftime("%Y-%m-%d"), - start=start_date.strftime("%Y-%m-%d"), - end=end_date.strftime("%Y-%m-%d") - ) + +format_bill_url = _format_url('tbill', BILL_IDS) +format_bond_url = _format_url('bond', BOND_IDS) + + +def load_frame(url, skiprows): + """ + Load a DataFrame of data from a Bank of Canada site. + """ + return pd.read_csv( + url, + skiprows=skiprows, + skipinitialspace=True, + na_values=["Bank holiday", "Not available"], + parse_dates=["Date"], + index_col="Date", + ).dropna(how='all') \ + .tz_localize('UTC') \ + .rename(columns=COLUMN_NAMES) + + +def check_known_inconsistencies(bill_data, bond_data): + """ + There are a couple quirks in the data provided by Bank of Canada. + Check that no new quirks have been introduced in the latest download. + """ + inconsistent_dates = bill_data.index.sym_diff(bond_data.index) + known_inconsistencies = [ + # bill_data has an entry for 2010-02-15, which bond_data doesn't. + # bond_data has an entry for 2006-09-04, which bill_data doesn't. + # Both of these dates are bank holidays (Flag Day and Labor Day, + # respectively). + pd.Timestamp('2006-09-04', tz='UTC'), + pd.Timestamp('2010-02-15', tz='UTC'), + # 2013-07-25 comes back as "Not available" from the bills endpoint. + # This date doesn't seem to be a bank holiday, but the previous + # calendar implementation dropped this entry, so we drop it as well. + # If someone cares deeply about the integrity of the Canadian trading + # calendar, they may want to consider forward-filling here rather than + # dropping the row. + pd.Timestamp('2013-07-25', tz='UTC'), + ] + unexpected_inconsistences = inconsistent_dates.drop(known_inconsistencies) + if len(unexpected_inconsistences): + in_bills = bill_data.index.difference(bond_data.index).difference( + known_inconsistencies + ) + in_bonds = bond_data.index.difference(bill_data.index).difference( + known_inconsistencies + ) + raise ValueError( + "Inconsistent dates for Canadian treasury bills vs bonds. \n" + "Dates with bills but not bonds: {in_bills}.\n" + "Dates with bonds but not bills: {in_bonds}.".format( + in_bills=in_bills, + in_bonds=in_bonds, + ) + ) + + +def earliest_possible_date(): + """ + The earliest date for which we can load data from this module. + """ + today = pd.Timestamp('now', tz='UTC').normalize() + # Bank of Canada only has the last 10 years of data at any given time. + return today.replace(year=today.year - 10) + + +def get_treasury_data(start_date, end_date): + bill_data = load_frame( + format_bill_url(start_date, end_date, start_date), + # We skip fewer rows here because we query for fewer bill fields, + # which makes the header smaller. + skiprows=18, ) + bond_data = load_frame( + format_bond_url(start_date, end_date, start_date), + skiprows=22, + ) + check_known_inconsistencies(bill_data, bond_data) - res_bill = requests.get(bill_url, stream=True) - res_bond = requests.get(bond_url, stream=True) - bill_iter = res_bill.iter_lines() - bond_iter = res_bond.iter_lines() + # dropna('any') removes the rows for which we only had data for one of + # bills/bonds. + out = pd.concat([bond_data, bill_data], axis=1).dropna(how='any') + assert set(out.columns) == set(six.itervalues(COLUMN_NAMES)) - bill_row = "" - while ",".join(BILLS) not in bill_row: - bill_row = bill_iter.next() - if 'Daily series:' in bill_row: - bill_end_date = datetime.datetime.strptime( - bill_row.split(' - ')[1].strip(), - "%Y-%m-%d").date() - bill_header = bill_row.split(",") - - bond_row = "" - while ",".join(BONDS) not in bond_row: - bond_row = bond_iter.next() - if 'Daily series:' in bond_row: - bond_end_date = datetime.datetime.strptime( - bond_row.split(' - ')[1].strip(), - "%Y-%m-%d").date() - bond_header = bond_row.split(",") - - # Line up the two dates - if bill_end_date > bond_end_date: - bill_iter.next() - elif bond_end_date > bill_end_date: - bond_iter.next() - - for bill_row in bill_iter: - bond_row = bond_iter.next() - bill_dict = dict(zip(bill_header, bill_row.split(","))) - bond_dict = dict(zip(bond_header, bond_row.split(","))) - if ' Bank holiday' in bond_row.split(",") + bill_row.split(","): - continue - if ' Not available' in bond_row.split(",") + bill_row.split(","): - continue - - bill_dict.update(bond_dict) - yield bill_dict - - -def get_treasury_data(): - mappings = treasury_mappings(_CURVE_MAPPINGS) - source = get_treasury_source() - return source_to_records(mappings, source) + # Multiply by 0.01 to convert from percentages to expected output format. + return out * 0.01