From 34c74346c9f59f328a1bdc93af22f41d3654052a Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Mon, 8 Aug 2016 15:05:19 -0400 Subject: [PATCH] ENH: Data portal reads future asset pricing. Use the future asset equity pricing reader, instead of reading directly from the bcolz table. Required since the format for writing the future data now uses the minute bar reader/writer pair. Add test cases to `test_data_portal` asserting both equity and future `get_spot_value` results. --- tests/test_data_portal.py | 88 +++++++++++++++++++++++++-- zipline/data/data_portal.py | 117 +++++------------------------------- 2 files changed, 99 insertions(+), 106 deletions(-) diff --git a/tests/test_data_portal.py b/tests/test_data_portal.py index 8b36f3ed..1376508f 100644 --- a/tests/test_data_portal.py +++ b/tests/test_data_portal.py @@ -12,8 +12,10 @@ # 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 collections import OrderedDict -from numpy import nan, full, append +from numpy import array, append, nan, full +from numpy.testing import assert_almost_equal import pandas as pd from pandas.tslib import Timedelta @@ -153,7 +155,7 @@ class TestDataPortal(WithDataPortal, index=dts)) yield asset.sid, pd.concat(dfs) - def test_get_last_traded_minute(self): + def test_get_last_traded_equity_minute(self): trading_calendar = self.trading_calendars[Equity] # Case: Missing data at front of data set, and request dt is before # first value. @@ -175,7 +177,7 @@ class TestDataPortal(WithDataPortal, self.data_portal.get_last_traded_dt( asset, dts[5], 'minute')) - def test_get_last_traded_minute_future(self): + def test_get_last_traded_future_minute(self): asset = self.asset_finder.retrieve_asset(10000) trading_calendar = self.trading_calendars[asset.exchange] # Case: Missing data at front of data set, and request dt is before @@ -197,7 +199,7 @@ class TestDataPortal(WithDataPortal, self.data_portal.get_last_traded_dt( asset, dts[5], 'minute')) - def test_get_last_traded_dt_daily(self): + def test_get_last_traded_dt_equity_daily(self): # Case: Missing data at front of data set, and request dt is before # first value. asset = self.asset_finder.retrieve_asset(1) @@ -215,6 +217,84 @@ class TestDataPortal(WithDataPortal, self.data_portal.get_last_traded_dt( asset, self.trading_days[3], 'daily')) + def test_get_spot_value_equity_minute(self): + trading_calendar = self.trading_calendars[Equity] + asset = self.asset_finder.retrieve_asset(1) + dts = trading_calendar.minutes_for_session(self.trading_days[2]) + + # Case: Get data on exact dt. + dt = dts[1] + expected = OrderedDict({ + 'open': 103.5, + 'high': 103.9, + 'low': 103.1, + 'close': 103.3, + 'volume': 1003, + 'price': 103.3 + }) + result = [self.data_portal.get_spot_value(asset, + field, + dt, + 'minute') + for field in expected.keys()] + assert_almost_equal(array(list(expected.values())), result) + + # Case: Get data on empty dt, return nan or most recent data for price. + dt = dts[100] + expected = OrderedDict({ + 'open': nan, + 'high': nan, + 'low': nan, + 'close': nan, + 'volume': 0, + 'price': 101.3 + }) + result = [self.data_portal.get_spot_value(asset, + field, + dt, + 'minute') + for field in expected.keys()] + assert_almost_equal(array(list(expected.values())), result) + + def test_get_spot_value_future_minute(self): + trading_calendar = self.trading_calendars['CME'] + asset = self.asset_finder.retrieve_asset(10000) + dts = trading_calendar.minutes_for_session(self.trading_days[3]) + + # Case: Get data on exact dt. + dt = dts[1] + expected = OrderedDict({ + 'open': 203.5, + 'high': 203.9, + 'low': 203.1, + 'close': 203.3, + 'volume': 2003, + 'price': 203.3 + }) + result = [self.data_portal.get_spot_value(asset, + field, + dt, + 'minute') + for field in expected.keys()] + assert_almost_equal(array(list(expected.values())), result) + + # Case: Get data on empty dt, return nan or most recent data for price. + dt = dts[100] + expected = OrderedDict({ + 'open': nan, + 'high': nan, + 'low': nan, + 'close': nan, + 'volume': 0, + 'price': 201.3 + }) + result = [self.data_portal.get_spot_value(asset, + field, + dt, + 'minute') + for field in expected.keys()] + assert_almost_equal(array(list(expected.values())), result) + def test_bar_count_for_simple_transforms(self): # July 2015 # Su Mo Tu We Th Fr Sa diff --git a/zipline/data/data_portal.py b/zipline/data/data_portal.py index f8685b3b..a2158cf0 100644 --- a/zipline/data/data_portal.py +++ b/zipline/data/data_portal.py @@ -14,7 +14,6 @@ # limitations under the License. from operator import mul -import bcolz from logbook import Logger import numpy as np @@ -276,50 +275,6 @@ class DataPortal(object): self._extra_source_df = extra_source_df - def _open_minute_file(self, field, asset): - sid_str = str(int(asset)) - - try: - carray = self._carrays[field][sid_str] - except KeyError: - carray = self._carrays[field][sid_str] = \ - self._get_ctable(asset)[field] - - return carray - - def _get_ctable(self, asset): - sid = int(asset) - - if isinstance(asset, Future): - if self._future_minute_reader.sid_path_func is not None: - path = self._future_minute_reader.sid_path_func( - self._future_minute_reader.rootdir, sid - ) - else: - path = "{0}/{1}.bcolz".format( - self._future_minute_reader.rootdir, sid) - elif isinstance(asset, Equity): - if self._equity_minute_reader.sid_path_func is not None: - path = self._equity_minute_reader.sid_path_func( - self._equity_minute_reader.rootdir, sid - ) - else: - path = "{0}/{1}.bcolz".format( - self._equity_minute_reader.rootdir, sid) - - else: - # TODO: Figure out if assets should be allowed if neither, and - # why this code path is being hit. - if self._equity_minute_reader.sid_path_func is not None: - path = self._equity_minute_reader.sid_path_func( - self._equity_minute_reader.rootdir, sid - ) - else: - path = "{0}/{1}.bcolz".format( - self._equity_minute_reader.rootdir, sid) - - return bcolz.open(path, mode='r') - def _get_pricing_reader(self, asset, data_frequency): return self._pricing_readers[type(asset)][data_frequency] @@ -402,23 +357,13 @@ class DataPortal(object): if data_frequency == "daily": return self._get_daily_data(asset, field, session_label) else: - if isinstance(asset, Future): - if field == "price": - return self._get_minute_spot_value_future( - asset, "close", dt) - else: - return self._get_minute_spot_value_future( - asset, field, dt) + if field == "last_traded": + return self.get_last_traded_dt(asset, dt, 'minute') + elif field == "price": + return self._get_minute_spot_value(asset, "close", dt, + ffill=True) else: - if field == "last_traded": - return self._equity_minute_reader.get_last_traded_dt( - asset, dt - ) - elif field == "price": - return self._get_minute_spot_value(asset, "close", dt, - True) - else: - return self._get_minute_spot_value(asset, field, dt) + return self._get_minute_spot_value(asset, field, dt) def get_adjustments(self, assets, field, dt, perspective_dt): """ @@ -537,59 +482,27 @@ class DataPortal(object): return spot_value - def _get_minute_spot_value_future(self, asset, column, dt): - # Futures bcolz files have 1440 bars per day (24 hours), 7 days a week. - # The file attributes contain the "start_dt" and "last_dt" fields, - # which represent the time period for this bcolz file. - - # The start_dt is midnight of the first day that this future started - # trading. - - # figure out the # of minutes between dt and this asset's start_dt - start_date = self._get_asset_start_date(asset) - minute_offset = int((dt - start_date).total_seconds() / 60) - - if minute_offset < 0: - # asking for a date that is before the asset's start date, no dice - return 0.0 - - # then just index into the bcolz carray at that offset - carray = self._open_minute_file(column, asset) - result = carray[minute_offset] - - # if there's missing data, go backwards until we run out of file - while result == 0 and minute_offset > 0: - minute_offset -= 1 - result = carray[minute_offset] - - if column != 'volume': - # FIXME switch to a futures reader - return result * 0.001 - else: - return result - def _get_minute_spot_value(self, asset, column, dt, ffill=False): - result = self._equity_minute_reader.get_value( + reader = self._get_pricing_reader(asset, 'minute') + result = reader.get_value( asset.sid, dt, column ) - if column == "volume": - if result == 0: - return 0 - elif not ffill or not np.isnan(result): - # if we're not forward filling, or we found a result, return it + if not ffill: return result # we are looking for price, and didn't find one. have to go hunting. - last_traded_dt = \ - self._equity_minute_reader.get_last_traded_dt(asset, dt) + last_traded_dt = reader.get_last_traded_dt(asset, dt) if last_traded_dt is pd.NaT: # no last traded dt, bail - return np.nan + if column == 'volume': + return 0 + else: + return np.nan # get the value as of the last traded dt - result = self._equity_minute_reader.get_value( + result = reader.get_value( asset.sid, last_traded_dt, column