From 37f4a5a56bc955366cdc0b316a6162f663731e9e Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Mon, 8 Aug 2016 11:28:58 -0400 Subject: [PATCH 1/2] TST: Add tests for Future asset last sale price. In support, also add future asset minute bar data and reader fixtures. --- tests/test_data_portal.py | 94 ++++++++++++++- zipline/testing/fixtures.py | 226 ++++++++++++++++++++++++++++-------- 2 files changed, 270 insertions(+), 50 deletions(-) diff --git a/tests/test_data_portal.py b/tests/test_data_portal.py index 8de79b25..8b36f3ed 100644 --- a/tests/test_data_portal.py +++ b/tests/test_data_portal.py @@ -31,10 +31,28 @@ class TestDataPortal(WithDataPortal, ASSET_FINDER_EQUITY_SIDS = (1,) START_DATE = pd.Timestamp('2016-08-01') - END_DATE = pd.Timestamp('2016-08-04') + END_DATE = pd.Timestamp('2016-08-08') + + TRADING_CALENDAR_STRS = ('NYSE', 'CME') EQUITY_DAILY_BAR_SOURCE_FROM_MINUTE = True + @classmethod + def make_futures_info(cls): + trading_sessions = cls.trading_sessions['CME'] + return pd.DataFrame({ + 'sid': [10000], + 'root_symbol': ['BAR'], + 'symbol': ['BARA'], + 'start_date': [trading_sessions[1]], + 'end_date': [cls.END_DATE], + # TODO: Make separate from 'end_date' + 'notice_date': [cls.END_DATE], + 'expiration_date': [cls.END_DATE], + 'multiplier': [500], + 'exchange': ['CME'], + }) + @classmethod def make_equity_minute_bar_data(cls): trading_calendar = cls.trading_calendars[Equity] @@ -83,6 +101,58 @@ class TestDataPortal(WithDataPortal, index=dts)) yield 1, pd.concat(dfs) + @classmethod + def make_future_minute_bar_data(cls): + asset = cls.asset_finder.retrieve_asset(10000) + trading_calendar = cls.trading_calendars[asset.exchange] + trading_sessions = cls.trading_sessions[asset.exchange] + # No data on first day, future asset intentionally not on the same + # dates as equities, so that cross-wiring of results do not create a + # false positive. + dts = trading_calendar.minutes_for_session(trading_sessions[1]) + dfs = [] + dfs.append(pd.DataFrame( + { + 'open': full(len(dts), nan), + 'high': full(len(dts), nan), + 'low': full(len(dts), nan), + 'close': full(len(dts), nan), + 'volume': full(len(dts), 0), + }, + index=dts)) + dts = trading_calendar.minutes_for_session(trading_sessions[2]) + dfs.append(pd.DataFrame( + { + 'open': append(200.5, full(len(dts) - 1, nan)), + 'high': append(200.9, full(len(dts) - 1, nan)), + 'low': append(200.1, full(len(dts) - 1, nan)), + 'close': append(200.3, full(len(dts) - 1, nan)), + 'volume': append(2000, full(len(dts) - 1, nan)), + }, + index=dts)) + dts = trading_calendar.minutes_for_session(trading_sessions[3]) + dfs.append(pd.DataFrame( + { + 'open': [nan, 203.50, 202.50, 204.50, 201.50, nan], + 'high': [nan, 203.90, 202.90, 204.90, 201.90, nan], + 'low': [nan, 203.10, 202.10, 204.10, 201.10, nan], + 'close': [nan, 203.30, 202.30, 204.30, 201.30, nan], + 'volume': [0, 2003, 2002, 2004, 2001, 0] + }, + index=dts[:6] + )) + dts = trading_calendar.minutes_for_session(trading_sessions[4]) + dfs.append(pd.DataFrame( + { + 'open': full(len(dts), nan), + 'high': full(len(dts), nan), + 'low': full(len(dts), nan), + 'close': full(len(dts), nan), + 'volume': full(len(dts), 0), + }, + index=dts)) + yield asset.sid, pd.concat(dfs) + def test_get_last_traded_minute(self): trading_calendar = self.trading_calendars[Equity] # Case: Missing data at front of data set, and request dt is before @@ -105,6 +175,28 @@ class TestDataPortal(WithDataPortal, self.data_portal.get_last_traded_dt( asset, dts[5], 'minute')) + def test_get_last_traded_minute_future(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 + # first value. + dts = trading_calendar.minutes_for_session(self.trading_days[0]) + self.assertTrue(pd.isnull( + self.data_portal.get_last_traded_dt( + asset, dts[0], 'minute'))) + + # Case: Data on requested dt. + dts = trading_calendar.minutes_for_session(self.trading_days[3]) + + self.assertEqual(dts[1], + self.data_portal.get_last_traded_dt( + asset, dts[1], 'minute')) + + # Case: No data on dt, but data occuring before dt. + self.assertEqual(dts[4], + self.data_portal.get_last_traded_dt( + asset, dts[5], 'minute')) + def test_get_last_traded_dt_daily(self): # Case: Missing data at front of data set, and request dt is before # first value. diff --git a/zipline/testing/fixtures.py b/zipline/testing/fixtures.py index 0f96be5d..325cef88 100644 --- a/zipline/testing/fixtures.py +++ b/zipline/testing/fixtures.py @@ -391,6 +391,7 @@ class WithTradingCalendars(object): """ TRADING_CALENDAR_STRS = ('NYSE',) TRADING_CALENDAR_FOR_ASSET_TYPE = {Equity: 'NYSE'} + TRADING_CALENDAR_FOR_EXCHANGE = {} # For backwards compatibility, exisitng tests and fixtures refer to # `trading_calendar` with the assumption that the value is the NYSE @@ -413,6 +414,9 @@ class WithTradingCalendars(object): cls.TRADING_CALENDAR_FOR_ASSET_TYPE): calendar = get_calendar(cal_str) cls.trading_calendars[asset_type] = calendar + for exchange, cal_str in iteritems(cls.TRADING_CALENDAR_FOR_EXCHANGE): + register_calendar(exchange, get_calendar(cal_str)) + cls.trading_calendars[exchange] = get_calendar(cal_str) class WithTradingEnvironment(WithAssetFinder, WithTradingCalendars): @@ -562,15 +566,17 @@ class WithTradingSessions(WithTradingCalendars): for cal_str in cls.TRADING_CALENDAR_STRS: trading_calendar = cls.trading_calendars[cal_str] - all_sessions = trading_calendar.all_sessions - start_loc = all_sessions.get_loc(cls.DATA_MIN_DAY, 'bfill') - end_loc = all_sessions.get_loc(cls.DATA_MAX_DAY, 'ffill') - - sessions = all_sessions[start_loc:end_loc + 1] + sessions = trading_calendar.sessions_in_range( + cls.DATA_MIN_DAY, cls.DATA_MAX_DAY) # Set name for aliasing. setattr(cls, '{0}_sessions'.format(cal_str.lower()), sessions) cls.trading_sessions[cal_str] = sessions + for exchange, cal_str in iteritems(cls.TRADING_CALENDAR_FOR_EXCHANGE): + trading_calendar = cls.trading_calendars[cal_str] + sessions = trading_calendar.sessions_in_range( + cls.DATA_MIN_DAY, cls.DATA_MAX_DAY) + cls.trading_sessions[exchange] = sessions class WithTmpDir(object): @@ -632,7 +638,7 @@ class WithEquityDailyBarData(WithTradingEnvironment): The end date up to which to create data. This defaults to ``END_DATE``. EQUITY_DAILY_BAR_SOURCE_FROM_MINUTE : bool If this flag is set, `make_equity_daily_bar_data` will read data from - the minute bars defined by `WithMinuteBarData`. + the minute bars defined by `WithEquityMinuteBarData`. The current default is `False`, but could be `True` in the future. Methods @@ -658,7 +664,7 @@ class WithEquityDailyBarData(WithTradingEnvironment): @classmethod def _make_equity_daily_bar_from_minute(cls): - assets = cls.asset_finder.retrieve_all(cls.asset_finder.sids) + assets = cls.asset_finder.retrieve_all(cls.asset_finder.equities_sids) minute_data = dict(cls.make_equity_minute_bar_data()) for asset in assets: yield asset.sid, minute_to_session(minute_data[asset.sid], @@ -792,12 +798,33 @@ class WithBcolzEquityDailyBarReaderFromCSVs(WithBcolzEquityDailyBarReader): _write_method_name = 'write_csvs' -class WithEquityMinuteBarData(WithTradingEnvironment): +def _trading_days_for_minute_bars(calendar, + start_date, + end_date, + lookback_days): + first_session = calendar.minute_to_session_label(start_date) + + if lookback_days > 0: + first_session = calendar.sessions_window( + first_session, + -1 * lookback_days + )[0] + + return calendar.sessions_in_range(first_session, end_date) + + +class _WithMinuteBarDataBase(WithTradingEnvironment): + MINUTE_BAR_LOOKBACK_DAYS = 0 + MINUTE_BAR_START_DATE = alias('START_DATE') + MINUTE_BAR_END_DATE = alias('END_DATE') + + +class WithEquityMinuteBarData(_WithMinuteBarDataBase): """ ZiplineTestCase mixin providing cls.equity_minute_bar_days. After init_class_fixtures has been called: - - `cls.equyt_minute_bar_days` has the range over which data has been + - `cls.equity_minute_bar_days` has the range over which data has been generated. Attributes @@ -806,10 +833,6 @@ class WithEquityMinuteBarData(WithTradingEnvironment): The number of days of data to add before the first day. This is used when a test needs to use history, in which case this should be set to the largest history window that will be requested. - EQUITY_MINUTE_BAR_USE_FULL_CALENDAR : bool - If this flag is set the ``equity_daily_bar_days`` will be the full - set of trading days from the trading environment. This flag overrides - ``EQUITY_MINUTE_BAR_LOOKBACK_DAYS``. EQUITY_MINUTE_BAR_START_DATE : Timestamp The date at to which to start creating data. This defaults to ``START_DATE``. @@ -830,11 +853,9 @@ class WithEquityMinuteBarData(WithTradingEnvironment): WithEquityDailyBarData zipline.testing.create_minute_bar_data """ - - EQUITY_MINUTE_BAR_LOOKBACK_DAYS = 0 - EQUITY_MINUTE_BAR_USE_FULL_CALENDAR = False - EQUITY_MINUTE_BAR_START_DATE = alias('START_DATE') - EQUITY_MINUTE_BAR_END_DATE = alias('END_DATE') + EQUITY_MINUTE_BAR_LOOKBACK_DAYS = alias('MINUTE_BAR_LOOKBACK_DAYS') + EQUITY_MINUTE_BAR_START_DATE = alias('MINUTE_BAR_START_DATE') + EQUITY_MINUTE_BAR_END_DATE = alias('MINUTE_BAR_END_DATE') @classmethod def make_equity_minute_bar_data(cls): @@ -844,32 +865,80 @@ class WithEquityMinuteBarData(WithTradingEnvironment): cls.equity_minute_bar_days[0], cls.equity_minute_bar_days[-1], ), - cls.asset_finder.sids, + cls.asset_finder.equities_sids, ) @classmethod def init_class_fixtures(cls): super(WithEquityMinuteBarData, cls).init_class_fixtures() trading_calendar = cls.trading_calendars[Equity] - if cls.EQUITY_MINUTE_BAR_USE_FULL_CALENDAR: - days = trading_calendar.all_execution_days - else: - first_session = trading_calendar.minute_to_session_label( - pd.Timestamp(cls.EQUITY_MINUTE_BAR_START_DATE) - ) + cls.equity_minute_bar_days = _trading_days_for_minute_bars( + trading_calendar, + pd.Timestamp(cls.EQUITY_MINUTE_BAR_START_DATE), + pd.Timestamp(cls.EQUITY_MINUTE_BAR_END_DATE), + cls.EQUITY_MINUTE_BAR_LOOKBACK_DAYS + ) - if cls.EQUITY_MINUTE_BAR_LOOKBACK_DAYS > 0: - first_session = trading_calendar.sessions_window( - first_session, - -1 * cls.EQUITY_MINUTE_BAR_LOOKBACK_DAYS - )[0] - days = trading_calendar.sessions_in_range( - first_session, - cls.EQUITY_MINUTE_BAR_END_DATE - ) +class WithFutureMinuteBarData(_WithMinuteBarDataBase): + """ + ZiplineTestCase mixin providing cls.future_minute_bar_days. - cls.equity_minute_bar_days = days + After init_class_fixtures has been called: + - `cls.future_minute_bar_days` has the range over which data has been + generated. + + Attributes + ---------- + FUTURE_MINUTE_BAR_LOOKBACK_DAYS : int + The number of days of data to add before the first day. + This is used when a test needs to use history, in which case this + should be set to the largest history window that will be requested. + FUTURE_MINUTE_BAR_START_DATE : Timestamp + The date at to which to start creating data. This defaults to + ``START_DATE``. + FUTURE_MINUTE_BAR_END_DATE = Timestamp + The end date up to which to create data. This defaults to ``END_DATE``. + + Methods + ------- + make_future_minute_bar_data() -> iterable[(int, pd.DataFrame)] + A class method that returns a dict mapping sid to dataframe + which will be written to into the the format of the inherited + class which writes the minute bar data for use by a reader. + By default this creates some simple sythetic data with + :func:`~zipline.testing.create_minute_bar_data` + + See Also + -------- + zipline.testing.create_minute_bar_data + """ + FUTURE_MINUTE_BAR_LOOKBACK_DAYS = alias('MINUTE_BAR_LOOKBACK_DAYS') + FUTURE_MINUTE_BAR_START_DATE = alias('MINUTE_BAR_START_DATE') + FUTURE_MINUTE_BAR_END_DATE = alias('MINUTE_BAR_END_DATE') + + @classmethod + def make_future_minute_bar_data(cls): + trading_calendar = get_calendar('CME') + return create_minute_bar_data( + trading_calendar.minutes_for_sessions_in_range( + cls.future_minute_bar_days[0], + cls.future_minute_bar_days[-1], + ), + cls.asset_finder.futures_sids, + ) + + @classmethod + def init_class_fixtures(cls): + super(WithFutureMinuteBarData, cls).init_class_fixtures() + # To be replaced by quanto calendar. + trading_calendar = get_calendar('CME') + cls.future_minute_bar_days = _trading_days_for_minute_bars( + trading_calendar, + pd.Timestamp(cls.FUTURE_MINUTE_BAR_START_DATE), + pd.Timestamp(cls.FUTURE_MINUTE_BAR_END_DATE), + cls.FUTURE_MINUTE_BAR_LOOKBACK_DAYS + ) class WithBcolzEquityMinuteBarReader(WithEquityMinuteBarData, WithTmpDir): @@ -891,14 +960,6 @@ class WithBcolzEquityMinuteBarReader(WithEquityMinuteBarData, WithTmpDir): ---------- BCOLZ_MINUTE_BAR_PATH : str The path inside the tmpdir where this will be written. - EQUITY_MINUTE_BAR_LOOKBACK_DAYS : int - The number of days of data to add before the first day. - This is used when a test needs to use history, in which case this - should be set to the largest history window that will be requested. - BCOLZ_MINUTE_BAR_USE_FULL_CALENDAR : bool - If this flag is set the ``equity_daily_bar_days`` will be the full - set of trading days from the trading environment. This flag overrides - ``EQUITY_MINUTE_BAR_LOOKBACK_DAYS``. Methods ------- @@ -913,17 +974,17 @@ class WithBcolzEquityMinuteBarReader(WithEquityMinuteBarData, WithTmpDir): WithDataPortal zipline.testing.create_minute_bar_data """ - BCOLZ_MINUTE_BAR_PATH = 'minute_equity_pricing.bcolz' + BCOLZ_EQUITY_MINUTE_BAR_PATH = 'minute_equity_pricing' @classmethod - def make_bcolz_minute_bar_rootdir_path(cls): - return cls.tmpdir.makedir(cls.BCOLZ_MINUTE_BAR_PATH) + def make_bcolz_equity_minute_bar_rootdir_path(cls): + return cls.tmpdir.makedir(cls.BCOLZ_EQUITY_MINUTE_BAR_PATH) @classmethod def init_class_fixtures(cls): super(WithBcolzEquityMinuteBarReader, cls).init_class_fixtures() - cls.bcolz_minute_bar_path = p = \ - cls.make_bcolz_minute_bar_rootdir_path() + cls.bcolz_equity_minute_bar_path = p = \ + cls.make_bcolz_equity_minute_bar_rootdir_path() days = cls.equity_minute_bar_days writer = BcolzMinuteBarWriter( @@ -939,6 +1000,67 @@ class WithBcolzEquityMinuteBarReader(WithEquityMinuteBarData, WithTmpDir): BcolzMinuteBarReader(p) +class WithBcolzFutureMinuteBarReader(WithFutureMinuteBarData, WithTmpDir): + """ + ZiplineTestCase mixin providing cls.bcolz_minute_bar_path, + cls.bcolz_minute_bar_ctable, and cls.bcolz_equity_minute_bar_reader + class level fixtures. + + After init_class_fixtures has been called: + - `cls.bcolz_minute_bar_path` is populated with + `cls.tmpdir.getpath(cls.BCOLZ_MINUTE_BAR_PATH)`. + - `cls.bcolz_minute_bar_ctable` is populated with data returned from + `cls.make_equity_minute_bar_data`. By default this calls + :func:`zipline.pipeline.loaders.synthetic.make_equity_minute_bar_data`. + - `cls.bcolz_equity_minute_bar_reader` is a minute bar reader + pointing to the directory that was just written to. + + Attributes + ---------- + BCOLZ_FUTURE_MINUTE_BAR_PATH : str + The path inside the tmpdir where this will be written. + + Methods + ------- + make_bcolz_minute_bar_rootdir_path() -> string + A class method that returns the path for the directory that contains + the minute bar ctables. By default this is a subdirectory + BCOLZ_MINUTE_BAR_PATH in the shared temp directory. + + See Also + -------- + WithBcolzEquityDailyBarReader + WithDataPortal + zipline.testing.create_minute_bar_data + """ + BCOLZ_FUTURE_MINUTE_BAR_PATH = 'minute_future_pricing' + + @classmethod + def make_bcolz_future_minute_bar_rootdir_path(cls): + return cls.tmpdir.makedir(cls.BCOLZ_FUTURE_MINUTE_BAR_PATH) + + @classmethod + def init_class_fixtures(cls): + super(WithBcolzFutureMinuteBarReader, cls).init_class_fixtures() + trading_calendar = get_calendar('CME') + cls.bcolz_future_minute_bar_path = p = \ + cls.make_bcolz_future_minute_bar_rootdir_path() + days = cls.future_minute_bar_days + + writer = BcolzMinuteBarWriter( + days[0], + p, + trading_calendar.schedule.market_open.loc[days], + trading_calendar.schedule.market_close.loc[days], + # TODO: Make futures minutes per day. + 1440, + ) + writer.write(cls.make_future_minute_bar_data()) + + cls.bcolz_future_minute_bar_reader = \ + BcolzMinuteBarReader(p) + + class WithAdjustmentReader(WithBcolzEquityDailyBarReader): """ ZiplineTestCase mixin providing cls.adjustment_reader as a class level @@ -1100,7 +1222,8 @@ class WithSeededRandomPipelineEngine(WithTradingSessions, WithAssetFinder): class WithDataPortal(WithAdjustmentReader, # Ordered so that bcolz minute reader is used first. - WithBcolzEquityMinuteBarReader): + WithBcolzEquityMinuteBarReader, + WithBcolzFutureMinuteBarReader): """ ZiplineTestCase mixin providing self.data_portal as an instance level fixture. @@ -1162,6 +1285,11 @@ class WithDataPortal(WithAdjustmentReader, if self.DATA_PORTAL_USE_ADJUSTMENTS else None ), + future_minute_reader=( + self.bcolz_future_minute_bar_reader + if self.DATA_PORTAL_USE_MINUTE_DATA else + None + ), ) def init_instance_fixtures(self): From 34c74346c9f59f328a1bdc93af22f41d3654052a Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Mon, 8 Aug 2016 15:05:19 -0400 Subject: [PATCH 2/2] 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