wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-15 12:15:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #793 from quantopian/treasury-cleanup

Browse Source 
Treasury cleanup
This commit is contained in:
Scott Sanderson
2015-10-25 17:55:42 -04:00
parent 653e7cbbf2 01888918dd
commit e49cc3a6d1
5 changed files with 391 additions and 623 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/data/benchmarks.py
+37 -112
View File
@@ -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:]
zipline/data/loader.py
+175 -144
View File
@@ -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.
zipline/data/loader_utils.py
-157
View File
@@ -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
zipline/data/treasuries.py
+54 -110
View File
@@ -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"^(?P<prefix>RIFLGFC)"
"(?P<unit>[YM])"
"(?P<periods>[0-9]{2})"
"(?P<suffix>_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):
zipline/data/treasuries_can.py
+125 -100
View File
@@ -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