diff --git a/tests/test_daily_history_aggregator.py b/tests/data/test_resample.py similarity index 54% rename from tests/test_daily_history_aggregator.py rename to tests/data/test_resample.py index 628e4d59..9326613a 100644 --- a/tests/test_daily_history_aggregator.py +++ b/tests/data/test_resample.py @@ -1,4 +1,3 @@ -# # Copyright 2016 Quantopian, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -12,16 +11,23 @@ # 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 numbers import Real from nose_parameterized import parameterized from numpy.testing import assert_almost_equal -from numpy import nan +from numpy import nan, array import pandas as pd +from pandas import DataFrame +from six import iteritems -from zipline.data.data_portal import DailyHistoryAggregator +from zipline.data.resample import ( + minute_to_session, + DailyHistoryAggregator +) from zipline.testing.fixtures import ( + WithEquityMinuteBarData, WithBcolzEquityMinuteBarReader, ZiplineTestCase, ) @@ -30,6 +36,111 @@ OHLC = ['open', 'high', 'low', 'close'] OHLCV = OHLC + ['volume'] +NYSE_MINUTES = OrderedDict(( + ('day_0_front', pd.date_range('2016-03-15 9:31', + '2016-03-15 9:33', + freq='min', + tz='US/Eastern').tz_convert('UTC')), + ('day_0_back', pd.date_range('2016-03-15 15:58', + '2016-03-15 16:00', + freq='min', + tz='US/Eastern').tz_convert('UTC')), + ('day_1_front', pd.date_range('2016-03-16 9:31', + '2016-03-16 9:33', + freq='min', + tz='US/Eastern').tz_convert('UTC')), + ('day_1_back', pd.date_range('2016-03-16 15:58', + '2016-03-16 16:00', + freq='min', + tz='US/Eastern').tz_convert('UTC')), +)) + + +SCENARIOS = OrderedDict(( + ('none_missing', array([ + [101.5, 101.9, 101.1, 101.3, 1001], + [103.5, 103.9, 103.1, 103.3, 1003], + [102.5, 102.9, 102.1, 102.3, 1002], + ])), + ('all_missing', array([ + [nan, nan, nan, nan, 0], + [nan, nan, nan, nan, 0], + [nan, nan, nan, nan, 0], + ])), + ('missing_first', array([ + [nan, nan, nan, nan, 0], + [103.5, 103.9, 103.1, 103.3, 1003], + [102.5, 102.9, 102.1, 102.3, 1002], + ])), + ('missing_last', array([ + [107.5, 107.9, 107.1, 107.3, 1007], + [108.5, 108.9, 108.1, 108.3, 1008], + [nan, nan, nan, nan, 0], + ])), + ('missing_middle', array([ + [103.5, 103.9, 103.1, 103.3, 1003], + [nan, nan, nan, nan, 0], + [102.5, 102.5, 102.1, 102.3, 1002], + ])), +)) + +OHLCV = ('open', 'high', 'low', 'close', 'volume') + +_EQUITY_CASES = ( + (1, (('none_missing', 'day_0_front'), + ('none_missing', 'day_0_back'))), + (2, (('missing_first', 'day_0_front'), + ('none_missing', 'day_0_back'))), + (3, (('missing_last', 'day_0_back'), + ('missing_first', 'day_1_front'))), +) + +EQUITY_CASES = OrderedDict() + +for sid, combos in _EQUITY_CASES: + frames = [DataFrame(SCENARIOS[s], columns=OHLCV). + set_index(NYSE_MINUTES[m]) + for s, m in combos] + EQUITY_CASES[sid] = pd.concat(frames) + +EXPECTED_AGGREGATION = { + 1: DataFrame({ + 'open': [101.5, 101.5, 101.5, 101.5, 101.5, 101.5], + 'high': [101.9, 103.9, 103.9, 103.9, 103.9, 103.9], + 'low': [101.1, 101.1, 101.1, 101.1, 101.1, 101.1], + 'close': [101.3, 103.3, 102.3, 101.3, 103.3, 102.3], + 'volume': [1001, 2004, 3006, 4007, 5010, 6012], + }, columns=OHLCV), + 2: DataFrame({ + 'open': [nan, 103.5, 103.5, 103.5, 103.5, 103.5], + 'high': [nan, 103.9, 103.9, 103.9, 103.9, 103.9], + 'low': [nan, 103.1, 102.1, 101.1, 101.1, 101.1], + 'close': [nan, 103.3, 102.3, 101.3, 103.3, 102.3], + 'volume': [0, 1003, 2005, 3006, 4009, 5011], + }, columns=OHLCV), + # Equity 3 straddles two days. + 3: DataFrame({ + 'open': [107.5, 107.5, 107.5, nan, 103.5, 103.5], + 'high': [107.9, 108.9, 108.9, nan, 103.9, 103.9], + 'low': [107.1, 107.1, 107.1, nan, 103.1, 102.1], + 'close': [107.3, 108.3, 108.3, nan, 103.3, 102.3], + 'volume': [1007, 2015, 2015, 0, 1003, 2005], + }, columns=OHLCV), +} + +EXPECTED_SESSIONS = { + 1: DataFrame([EXPECTED_AGGREGATION[1].iloc[-1].values], + columns=OHLCV, + index=['2016-03-15']), + 2: DataFrame([EXPECTED_AGGREGATION[2].iloc[-1].values], + columns=OHLCV, + index=['2016-03-15']), + 3: DataFrame(EXPECTED_AGGREGATION[3].iloc[[2, 5]].values, + columns=OHLCV, + index=['2016-03-15', '2016-03-16']), +} + + class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, ZiplineTestCase): @@ -47,71 +158,13 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, TRADING_ENV_MAX_DATE = END_DATE = pd.Timestamp( '2016-03-31', tz='UTC', ) - ASSET_FINDER_EQUITY_SIDS = 1, 2 - - minutes = pd.date_range('2016-03-15 9:31', - '2016-03-15 9:36', - freq='min', - tz='US/Eastern').tz_convert('UTC') + ASSET_FINDER_EQUITY_SIDS = 1, 2, 3 @classmethod def make_equity_minute_bar_data(cls): - # sid data is created so that at least one high is lower than a - # previous high, and the inverse for low - yield 1, pd.DataFrame( - { - 'open': [nan, 103.50, 102.50, 104.50, 101.50, nan], - 'high': [nan, 103.90, 102.90, 104.90, 101.90, nan], - 'low': [nan, 103.10, 102.10, 104.10, 101.10, nan], - 'close': [nan, 103.30, 102.30, 104.30, 101.30, nan], - 'volume': [0, 1003, 1002, 1004, 1001, 0] - }, - index=cls.minutes, - ) - # sid 2 is included to provide data on different bars than sid 1, - # as will as illiquidty mid-day - yield 2, pd.DataFrame( - { - 'open': [201.50, nan, 204.50, nan, 200.50, 202.50], - 'high': [201.90, nan, 204.90, nan, 200.90, 202.90], - 'low': [201.10, nan, 204.10, nan, 200.10, 202.10], - 'close': [201.30, nan, 203.50, nan, 200.30, 202.30], - 'volume': [2001, 0, 2004, 0, 2000, 2002], - }, - index=cls.minutes, - ) - - expected_values = { - 1: pd.DataFrame( - { - 'open': [nan, 103.50, 103.50, 103.50, 103.50, 103.50], - 'high': [nan, 103.90, 103.90, 104.90, 104.90, 104.90], - 'low': [nan, 103.10, 102.10, 102.10, 101.10, 101.10], - 'close': [nan, 103.30, 102.30, 104.30, 101.30, 101.30], - 'volume': [0, 1003, 2005, 3009, 4010, 4010] - }, - index=minutes, - ), - 2: pd.DataFrame( - { - 'open': [201.50, 201.50, 201.50, 201.50, 201.50, 201.50], - 'high': [201.90, 201.90, 204.90, 204.90, 204.90, 204.90], - 'low': [201.10, 201.10, 201.10, 201.10, 200.10, 200.10], - 'close': [201.30, 201.30, 203.50, 203.50, 200.30, 202.30], - 'volume': [2001, 2001, 4005, 4005, 6005, 8007], - }, - index=minutes, - ) - } - - @classmethod - def init_class_fixtures(cls): - super(MinuteToDailyAggregationTestCase, cls).init_class_fixtures() - - cls.EQUITIES = { - 1: cls.env.asset_finder.retrieve_asset(1), - 2: cls.env.asset_finder.retrieve_asset(2) - } + for sid in cls.ASSET_FINDER_EQUITY_SIDS: + frame = EQUITY_CASES[sid] + yield sid, frame def init_instance_fixtures(self): super(MinuteToDailyAggregationTestCase, self).init_instance_fixtures() @@ -134,15 +187,20 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, ('low_2', 'low', 2), ('close_2', 'close', 2), ('volume_2', 'volume', 2), - + ('open_3', 'open', 3), + ('high_3', 'high', 3), + ('low_3', 'low', 3), + ('close_3', 'close', 3), + ('volume_3', 'volume', 3), ]) def test_contiguous_minutes_individual(self, name, field, sid): # First test each minute in order. method_name = field + 's' results = [] repeat_results = [] - asset = self.EQUITIES[sid] - for minute in self.minutes: + asset = self.asset_finder.retrieve_asset(sid) + minutes = EQUITY_CASES[asset].index + for minute in minutes: value = getattr(self.equity_daily_aggregator, method_name)( [asset], minute)[0] # Prevent regression on building an array when scalar is intended. @@ -158,9 +216,9 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, self.assertIsInstance(value, Real) repeat_results.append(value) - assert_almost_equal(results, self.expected_values[asset][field], + assert_almost_equal(results, EXPECTED_AGGREGATION[asset][field], err_msg='sid={0} field={1}'.format(asset, field)) - assert_almost_equal(repeat_results, self.expected_values[asset][field], + assert_almost_equal(repeat_results, EXPECTED_AGGREGATION[asset][field], err_msg='sid={0} field={1}'.format(asset, field)) @parameterized.expand([ @@ -174,21 +232,26 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, ('low_2', 'low', 2), ('close_2', 'close', 2), ('volume_2', 'volume', 2), - + ('open_3', 'open', 3), + ('high_3', 'high', 3), + ('low_3', 'low', 3), + ('close_3', 'close', 3), + ('volume_3', 'volume', 3), ]) def test_skip_minutes_individual(self, name, field, sid): # Test skipping minutes, to exercise backfills. # Tests initial backfill and mid day backfill. method_name = field + 's' + asset = self.asset_finder.retrieve_asset(sid) + minutes = EQUITY_CASES[asset].index for i in [1, 5]: - minute = self.minutes[i] - asset = self.EQUITIES[sid] + minute = minutes[i] value = getattr(self.equity_daily_aggregator, method_name)( [asset], minute)[0] # Prevent regression on building an array when scalar is intended. self.assertIsInstance(value, Real) assert_almost_equal(value, - self.expected_values[sid][field][i], + EXPECTED_AGGREGATION[sid][field][i], err_msg='sid={0} field={1} dt={2}'.format( sid, field, minute)) @@ -200,7 +263,7 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, # Prevent regression on building an array when scalar is intended. self.assertIsInstance(value, Real) assert_almost_equal(value, - self.expected_values[sid][field][i], + EXPECTED_AGGREGATION[sid][field][i], err_msg='sid={0} field={1} dt={2}'.format( sid, field, minute)) @@ -208,10 +271,11 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, def test_contiguous_minutes_multiple(self, field): # First test each minute in order. method_name = field + 's' - assets = sorted(self.EQUITIES.values()) + assets = self.asset_finder.retrieve_all([1, 2]) results = {asset: [] for asset in assets} repeat_results = {asset: [] for asset in assets} - for minute in self.minutes: + minutes = EQUITY_CASES[1].index + for minute in minutes: values = getattr(self.equity_daily_aggregator, method_name)( assets, minute) for j, asset in enumerate(assets): @@ -234,11 +298,11 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, repeat_results[asset].append(value) for asset in assets: assert_almost_equal(results[asset], - self.expected_values[asset][field], + EXPECTED_AGGREGATION[asset][field], err_msg='sid={0} field={1}'.format( asset, field)) assert_almost_equal(repeat_results[asset], - self.expected_values[asset][field], + EXPECTED_AGGREGATION[asset][field], err_msg='sid={0} field={1}'.format( asset, field)) @@ -247,9 +311,10 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, # Test skipping minutes, to exercise backfills. # Tests initial backfill and mid day backfill. method_name = field + 's' - assets = sorted(self.EQUITIES.values()) + assets = self.asset_finder.retrieve_all([1, 2]) + minutes = EQUITY_CASES[1].index for i in [1, 5]: - minute = self.minutes[i] + minute = minutes[i] values = getattr(self.equity_daily_aggregator, method_name)( assets, minute) for j, asset in enumerate(assets): @@ -259,7 +324,7 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, self.assertIsInstance(value, Real) assert_almost_equal( value, - self.expected_values[asset][field][i], + EXPECTED_AGGREGATION[asset][field][i], err_msg='sid={0} field={1} dt={2}'.format( asset, field, minute)) @@ -275,6 +340,46 @@ class MinuteToDailyAggregationTestCase(WithBcolzEquityMinuteBarReader, self.assertIsInstance(value, Real) assert_almost_equal( value, - self.expected_values[asset][field][i], + EXPECTED_AGGREGATION[asset][field][i], err_msg='sid={0} field={1} dt={2}'.format( asset, field, minute)) + + +class TestMinuteToSession(WithEquityMinuteBarData, + ZiplineTestCase): + + # March 2016 + # Su Mo Tu We Th Fr Sa + # 1 2 3 4 5 + # 6 7 8 9 10 11 12 + # 13 14 15 16 17 18 19 + # 20 21 22 23 24 25 26 + # 27 28 29 30 31 + + START_DATE = pd.Timestamp( + '2016-03-15', tz='UTC', + ) + END_DATE = pd.Timestamp( + '2016-03-15', tz='UTC', + ) + ASSET_FINDER_EQUITY_SIDS = 1, 2, 3 + + @classmethod + def make_equity_minute_bar_data(cls): + for sid, frame in iteritems(EQUITY_CASES): + yield sid, frame + + @classmethod + def init_class_fixtures(cls): + super(TestMinuteToSession, cls).init_class_fixtures() + cls.equity_frames = { + sid: frame for sid, frame in cls.make_equity_minute_bar_data()} + + def test_minute_to_session(self): + for sid in self.ASSET_FINDER_EQUITY_SIDS: + frame = self.equity_frames[sid] + expected = EXPECTED_SESSIONS[sid] + result = minute_to_session(frame, self.nyse_calendar) + assert_almost_equal(expected.values, + result.values, + err_msg='sid={0}'.format(sid)) diff --git a/zipline/data/daily_history_aggregator.py b/zipline/data/daily_history_aggregator.py deleted file mode 100644 index 1e710d7b..00000000 --- a/zipline/data/daily_history_aggregator.py +++ /dev/null @@ -1,416 +0,0 @@ -# -# Copyright 2016 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. -import numpy as np -import pandas as pd - - -class DailyHistoryAggregator(object): - """ - Converts minute pricing data into a daily summary, to be used for the - last slot in a call to history with a frequency of `1d`. - - This summary is the same as a daily bar rollup of minute data, with the - distinction that the summary is truncated to the `dt` requested. - i.e. the aggregation slides forward during a the course of simulation day. - - Provides aggregation for `open`, `high`, `low`, `close`, and `volume`. - The aggregation rules for each price type is documented in their respective - - """ - - def __init__(self, market_opens, minute_reader, trading_calendar): - self._market_opens = market_opens - self._minute_reader = minute_reader - self._trading_calendar = trading_calendar - - # The caches are structured as (date, market_open, entries), where - # entries is a dict of asset -> (last_visited_dt, value) - # - # Whenever an aggregation method determines the current value, - # the entry for the respective asset should be overwritten with a new - # entry for the current dt.value (int) and aggregation value. - # - # When the requested dt's date is different from date the cache is - # flushed, so that the cache entries do not grow unbounded. - # - # Example cache: - # cache = (date(2016, 3, 17), - # pd.Timestamp('2016-03-17 13:31', tz='UTC'), - # { - # 1: (1458221460000000000, np.nan), - # 2: (1458221460000000000, 42.0), - # }) - self._caches = { - 'open': None, - 'high': None, - 'low': None, - 'close': None, - 'volume': None - } - - # The int value is used for deltas to avoid extra computation from - # creating new Timestamps. - self._one_min = pd.Timedelta('1 min').value - - def _prelude(self, dt, field): - date = dt.date() - dt_value = dt.value - cache = self._caches[field] - if cache is None or cache[0] != date: - market_open = self._market_opens.loc[date] - cache = self._caches[field] = (dt.date(), market_open, {}) - - _, market_open, entries = cache - market_open = market_open.tz_localize('UTC') - if dt != market_open: - prev_dt = dt_value - self._one_min - else: - prev_dt = None - return market_open, prev_dt, dt_value, entries - - def opens(self, assets, dt): - """ - The open field's aggregation returns the first value that occurs - for the day, if there has been no data on or before the `dt` the open - is `nan`. - - Once the first non-nan open is seen, that value remains constant per - asset for the remainder of the day. - - Returns - ------- - np.array with dtype=float64, in order of assets parameter. - """ - market_open, prev_dt, dt_value, entries = self._prelude(dt, 'open') - - opens = [] - session_label = self._trading_calendar.minute_to_session_label(dt) - - for asset in assets: - if not asset.is_alive_for_session(session_label): - opens.append(np.NaN) - continue - - if prev_dt is None: - val = self._minute_reader.get_value(asset, dt, 'open') - entries[asset] = (dt_value, val) - opens.append(val) - continue - else: - try: - last_visited_dt, first_open = entries[asset] - if last_visited_dt == dt_value: - opens.append(first_open) - continue - elif not pd.isnull(first_open): - opens.append(first_open) - entries[asset] = (dt_value, first_open) - continue - else: - after_last = pd.Timestamp( - last_visited_dt + self._one_min, tz='UTC') - window = self._minute_reader.load_raw_arrays( - ['open'], - after_last, - dt, - [asset], - )[0] - nonnan = window[~pd.isnull(window)] - if len(nonnan): - val = nonnan[0] - else: - val = np.nan - entries[asset] = (dt_value, val) - opens.append(val) - continue - except KeyError: - window = self._minute_reader.load_raw_arrays( - ['open'], - market_open, - dt, - [asset], - )[0] - nonnan = window[~pd.isnull(window)] - if len(nonnan): - val = nonnan[0] - else: - val = np.nan - entries[asset] = (dt_value, val) - opens.append(val) - continue - return np.array(opens) - - def highs(self, assets, dt): - """ - The high field's aggregation returns the largest high seen between - the market open and the current dt. - If there has been no data on or before the `dt` the high is `nan`. - - Returns - ------- - np.array with dtype=float64, in order of assets parameter. - """ - market_open, prev_dt, dt_value, entries = self._prelude(dt, 'high') - - highs = [] - session_label = self._trading_calendar.minute_to_session_label(dt) - - for asset in assets: - if not asset.is_alive_for_session(session_label): - highs.append(np.NaN) - continue - - if prev_dt is None: - val = self._minute_reader.get_value(asset, dt, 'high') - entries[asset] = (dt_value, val) - highs.append(val) - continue - else: - try: - last_visited_dt, last_max = entries[asset] - if last_visited_dt == dt_value: - highs.append(last_max) - continue - elif last_visited_dt == prev_dt: - curr_val = self._minute_reader.get_value( - asset, dt, 'high') - if pd.isnull(curr_val): - val = last_max - elif pd.isnull(last_max): - val = curr_val - else: - val = max(last_max, curr_val) - entries[asset] = (dt_value, val) - highs.append(val) - continue - else: - after_last = pd.Timestamp( - last_visited_dt + self._one_min, tz='UTC') - window = self._minute_reader.load_raw_arrays( - ['high'], - after_last, - dt, - [asset], - )[0].T - val = max(last_max, np.nanmax(window)) - entries[asset] = (dt_value, val) - highs.append(val) - continue - except KeyError: - window = self._minute_reader.load_raw_arrays( - ['high'], - market_open, - dt, - [asset], - )[0].T - val = np.nanmax(window) - entries[asset] = (dt_value, val) - highs.append(val) - continue - return np.array(highs) - - def lows(self, assets, dt): - """ - The low field's aggregation returns the smallest low seen between - the market open and the current dt. - If there has been no data on or before the `dt` the low is `nan`. - - Returns - ------- - np.array with dtype=float64, in order of assets parameter. - """ - market_open, prev_dt, dt_value, entries = self._prelude(dt, 'low') - - lows = [] - session_label = self._trading_calendar.minute_to_session_label(dt) - - for asset in assets: - if not asset.is_alive_for_session(session_label): - lows.append(np.NaN) - continue - - if prev_dt is None: - val = self._minute_reader.get_value(asset, dt, 'low') - entries[asset] = (dt_value, val) - lows.append(val) - continue - else: - try: - last_visited_dt, last_min = entries[asset] - if last_visited_dt == dt_value: - lows.append(last_min) - continue - elif last_visited_dt == prev_dt: - curr_val = self._minute_reader.get_value( - asset, dt, 'low') - val = np.nanmin([last_min, curr_val]) - entries[asset] = (dt_value, val) - lows.append(val) - continue - else: - after_last = pd.Timestamp( - last_visited_dt + self._one_min, tz='UTC') - window = self._minute_reader.load_raw_arrays( - ['low'], - after_last, - dt, - [asset], - )[0].T - window_min = np.nanmin(window) - if pd.isnull(window_min): - val = last_min - else: - val = min(last_min, window_min) - entries[asset] = (dt_value, val) - lows.append(val) - continue - except KeyError: - window = self._minute_reader.load_raw_arrays( - ['low'], - market_open, - dt, - [asset], - )[0].T - val = np.nanmin(window) - entries[asset] = (dt_value, val) - lows.append(val) - continue - return np.array(lows) - - def closes(self, assets, dt): - """ - The close field's aggregation returns the latest close at the given - dt. - If the close for the given dt is `nan`, the most recent non-nan - `close` is used. - If there has been no data on or before the `dt` the close is `nan`. - - Returns - ------- - np.array with dtype=float64, in order of assets parameter. - """ - market_open, prev_dt, dt_value, entries = self._prelude(dt, 'close') - - closes = [] - session_label = self._trading_calendar.minute_to_session_label(dt) - - for asset in assets: - if not asset.is_alive_for_session(session_label): - closes.append(np.NaN) - continue - - if prev_dt is None: - val = self._minute_reader.get_value(asset, dt, 'close') - entries[asset] = (dt_value, val) - closes.append(val) - continue - else: - try: - last_visited_dt, last_close = entries[asset] - if last_visited_dt == dt_value: - closes.append(last_close) - continue - elif last_visited_dt == prev_dt: - val = self._minute_reader.get_value( - asset, dt, 'close') - if pd.isnull(val): - val = last_close - entries[asset] = (dt_value, val) - closes.append(val) - continue - else: - val = self._minute_reader.get_value( - asset, dt, 'close') - if pd.isnull(val): - val = self.closes( - [asset], - pd.Timestamp(prev_dt, tz='UTC'))[0] - entries[asset] = (dt_value, val) - closes.append(val) - continue - except KeyError: - val = self._minute_reader.get_value( - asset, dt, 'close') - if pd.isnull(val): - val = self.closes([asset], - pd.Timestamp(prev_dt, tz='UTC'))[0] - entries[asset] = (dt_value, val) - closes.append(val) - continue - return np.array(closes) - - def volumes(self, assets, dt): - """ - The volume field's aggregation returns the sum of all volumes - between the market open and the `dt` - If there has been no data on or before the `dt` the volume is 0. - - Returns - ------- - np.array with dtype=int64, in order of assets parameter. - """ - market_open, prev_dt, dt_value, entries = self._prelude(dt, 'volume') - - volumes = [] - session_label = self._trading_calendar.minute_to_session_label(dt) - - for asset in assets: - if not asset.is_alive_for_session(session_label): - volumes.append(0) - continue - - if prev_dt is None: - val = self._minute_reader.get_value(asset, dt, 'volume') - entries[asset] = (dt_value, val) - volumes.append(val) - continue - else: - try: - last_visited_dt, last_total = entries[asset] - if last_visited_dt == dt_value: - volumes.append(last_total) - continue - elif last_visited_dt == prev_dt: - val = self._minute_reader.get_value( - asset, dt, 'volume') - val += last_total - entries[asset] = (dt_value, val) - volumes.append(val) - continue - else: - after_last = pd.Timestamp( - last_visited_dt + self._one_min, tz='UTC') - window = self._minute_reader.load_raw_arrays( - ['volume'], - after_last, - dt, - [asset], - )[0] - val = np.nansum(window) + last_total - entries[asset] = (dt_value, val) - volumes.append(val) - continue - except KeyError: - window = self._minute_reader.load_raw_arrays( - ['volume'], - market_open, - dt, - [asset], - )[0] - val = np.nansum(window) - entries[asset] = (dt_value, val) - volumes.append(val) - continue - return np.array(volumes) diff --git a/zipline/data/data_portal.py b/zipline/data/data_portal.py index a2158cf0..8d56a2b1 100644 --- a/zipline/data/data_portal.py +++ b/zipline/data/data_portal.py @@ -23,7 +23,7 @@ from six import iteritems from six.moves import reduce from zipline.assets import Asset, Future, Equity -from zipline.data.daily_history_aggregator import DailyHistoryAggregator +from zipline.data.resample import DailyHistoryAggregator from zipline.data.us_equity_pricing import NoDataOnDate from zipline.data.us_equity_loader import ( USEquityDailyHistoryLoader, diff --git a/zipline/data/resample.py b/zipline/data/resample.py index fc8979d3..05d223e5 100644 --- a/zipline/data/resample.py +++ b/zipline/data/resample.py @@ -11,14 +11,18 @@ # 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 -_MINUTE_TO_SESSION_OHCLV_HOW = { - 'open': 'first', - 'high': 'max', - 'low': 'min', - 'close': 'last', - 'volume': 'sum' -} +import numpy as np +import pandas as pd + +_MINUTE_TO_SESSION_OHCLV_HOW = OrderedDict(( + ('open', 'first'), + ('high', 'max'), + ('low', 'min'), + ('close', 'last'), + ('volume', 'sum'), +)) def minute_to_session(minute_frame, calendar): @@ -46,3 +50,403 @@ def minute_to_session(minute_frame, calendar): # including days with no minute data. return minute_frame.resample(calendar.day, how=_MINUTE_TO_SESSION_OHCLV_HOW) + + +class DailyHistoryAggregator(object): + """ + Converts minute pricing data into a daily summary, to be used for the + last slot in a call to history with a frequency of `1d`. + + This summary is the same as a daily bar rollup of minute data, with the + distinction that the summary is truncated to the `dt` requested. + i.e. the aggregation slides forward during a the course of simulation day. + + Provides aggregation for `open`, `high`, `low`, `close`, and `volume`. + The aggregation rules for each price type is documented in their respective + + """ + + def __init__(self, market_opens, minute_reader, trading_calendar): + self._market_opens = market_opens + self._minute_reader = minute_reader + self._trading_calendar = trading_calendar + + # The caches are structured as (date, market_open, entries), where + # entries is a dict of asset -> (last_visited_dt, value) + # + # Whenever an aggregation method determines the current value, + # the entry for the respective asset should be overwritten with a new + # entry for the current dt.value (int) and aggregation value. + # + # When the requested dt's date is different from date the cache is + # flushed, so that the cache entries do not grow unbounded. + # + # Example cache: + # cache = (date(2016, 3, 17), + # pd.Timestamp('2016-03-17 13:31', tz='UTC'), + # { + # 1: (1458221460000000000, np.nan), + # 2: (1458221460000000000, 42.0), + # }) + self._caches = { + 'open': None, + 'high': None, + 'low': None, + 'close': None, + 'volume': None + } + + # The int value is used for deltas to avoid extra computation from + # creating new Timestamps. + self._one_min = pd.Timedelta('1 min').value + + def _prelude(self, dt, field): + date = dt.date() + dt_value = dt.value + cache = self._caches[field] + if cache is None or cache[0] != date: + market_open = self._market_opens.loc[date] + cache = self._caches[field] = (dt.date(), market_open, {}) + + _, market_open, entries = cache + market_open = market_open.tz_localize('UTC') + if dt != market_open: + prev_dt = dt_value - self._one_min + else: + prev_dt = None + return market_open, prev_dt, dt_value, entries + + def opens(self, assets, dt): + """ + The open field's aggregation returns the first value that occurs + for the day, if there has been no data on or before the `dt` the open + is `nan`. + + Once the first non-nan open is seen, that value remains constant per + asset for the remainder of the day. + + Returns + ------- + np.array with dtype=float64, in order of assets parameter. + """ + market_open, prev_dt, dt_value, entries = self._prelude(dt, 'open') + + opens = [] + session_label = self._trading_calendar.minute_to_session_label(dt) + + for asset in assets: + if not asset.is_alive_for_session(session_label): + opens.append(np.NaN) + continue + + if prev_dt is None: + val = self._minute_reader.get_value(asset, dt, 'open') + entries[asset] = (dt_value, val) + opens.append(val) + continue + else: + try: + last_visited_dt, first_open = entries[asset] + if last_visited_dt == dt_value: + opens.append(first_open) + continue + elif not pd.isnull(first_open): + opens.append(first_open) + entries[asset] = (dt_value, first_open) + continue + else: + after_last = pd.Timestamp( + last_visited_dt + self._one_min, tz='UTC') + window = self._minute_reader.load_raw_arrays( + ['open'], + after_last, + dt, + [asset], + )[0] + nonnan = window[~pd.isnull(window)] + if len(nonnan): + val = nonnan[0] + else: + val = np.nan + entries[asset] = (dt_value, val) + opens.append(val) + continue + except KeyError: + window = self._minute_reader.load_raw_arrays( + ['open'], + market_open, + dt, + [asset], + )[0] + nonnan = window[~pd.isnull(window)] + if len(nonnan): + val = nonnan[0] + else: + val = np.nan + entries[asset] = (dt_value, val) + opens.append(val) + continue + return np.array(opens) + + def highs(self, assets, dt): + """ + The high field's aggregation returns the largest high seen between + the market open and the current dt. + If there has been no data on or before the `dt` the high is `nan`. + + Returns + ------- + np.array with dtype=float64, in order of assets parameter. + """ + market_open, prev_dt, dt_value, entries = self._prelude(dt, 'high') + + highs = [] + session_label = self._trading_calendar.minute_to_session_label(dt) + + for asset in assets: + if not asset.is_alive_for_session(session_label): + highs.append(np.NaN) + continue + + if prev_dt is None: + val = self._minute_reader.get_value(asset, dt, 'high') + entries[asset] = (dt_value, val) + highs.append(val) + continue + else: + try: + last_visited_dt, last_max = entries[asset] + if last_visited_dt == dt_value: + highs.append(last_max) + continue + elif last_visited_dt == prev_dt: + curr_val = self._minute_reader.get_value( + asset, dt, 'high') + if pd.isnull(curr_val): + val = last_max + elif pd.isnull(last_max): + val = curr_val + else: + val = max(last_max, curr_val) + entries[asset] = (dt_value, val) + highs.append(val) + continue + else: + after_last = pd.Timestamp( + last_visited_dt + self._one_min, tz='UTC') + window = self._minute_reader.load_raw_arrays( + ['high'], + after_last, + dt, + [asset], + )[0].T + val = max(last_max, np.nanmax(window)) + entries[asset] = (dt_value, val) + highs.append(val) + continue + except KeyError: + window = self._minute_reader.load_raw_arrays( + ['high'], + market_open, + dt, + [asset], + )[0].T + val = np.nanmax(window) + entries[asset] = (dt_value, val) + highs.append(val) + continue + return np.array(highs) + + def lows(self, assets, dt): + """ + The low field's aggregation returns the smallest low seen between + the market open and the current dt. + If there has been no data on or before the `dt` the low is `nan`. + + Returns + ------- + np.array with dtype=float64, in order of assets parameter. + """ + market_open, prev_dt, dt_value, entries = self._prelude(dt, 'low') + + lows = [] + session_label = self._trading_calendar.minute_to_session_label(dt) + + for asset in assets: + if not asset.is_alive_for_session(session_label): + lows.append(np.NaN) + continue + + if prev_dt is None: + val = self._minute_reader.get_value(asset, dt, 'low') + entries[asset] = (dt_value, val) + lows.append(val) + continue + else: + try: + last_visited_dt, last_min = entries[asset] + if last_visited_dt == dt_value: + lows.append(last_min) + continue + elif last_visited_dt == prev_dt: + curr_val = self._minute_reader.get_value( + asset, dt, 'low') + val = np.nanmin([last_min, curr_val]) + entries[asset] = (dt_value, val) + lows.append(val) + continue + else: + after_last = pd.Timestamp( + last_visited_dt + self._one_min, tz='UTC') + window = self._minute_reader.load_raw_arrays( + ['low'], + after_last, + dt, + [asset], + )[0].T + window_min = np.nanmin(window) + if pd.isnull(window_min): + val = last_min + else: + val = min(last_min, window_min) + entries[asset] = (dt_value, val) + lows.append(val) + continue + except KeyError: + window = self._minute_reader.load_raw_arrays( + ['low'], + market_open, + dt, + [asset], + )[0].T + val = np.nanmin(window) + entries[asset] = (dt_value, val) + lows.append(val) + continue + return np.array(lows) + + def closes(self, assets, dt): + """ + The close field's aggregation returns the latest close at the given + dt. + If the close for the given dt is `nan`, the most recent non-nan + `close` is used. + If there has been no data on or before the `dt` the close is `nan`. + + Returns + ------- + np.array with dtype=float64, in order of assets parameter. + """ + market_open, prev_dt, dt_value, entries = self._prelude(dt, 'close') + + closes = [] + session_label = self._trading_calendar.minute_to_session_label(dt) + + for asset in assets: + if not asset.is_alive_for_session(session_label): + closes.append(np.NaN) + continue + + if prev_dt is None: + val = self._minute_reader.get_value(asset, dt, 'close') + entries[asset] = (dt_value, val) + closes.append(val) + continue + else: + try: + last_visited_dt, last_close = entries[asset] + if last_visited_dt == dt_value: + closes.append(last_close) + continue + elif last_visited_dt == prev_dt: + val = self._minute_reader.get_value( + asset, dt, 'close') + if pd.isnull(val): + val = last_close + entries[asset] = (dt_value, val) + closes.append(val) + continue + else: + val = self._minute_reader.get_value( + asset, dt, 'close') + if pd.isnull(val): + val = self.closes( + [asset], + pd.Timestamp(prev_dt, tz='UTC'))[0] + entries[asset] = (dt_value, val) + closes.append(val) + continue + except KeyError: + val = self._minute_reader.get_value( + asset, dt, 'close') + if pd.isnull(val): + val = self.closes([asset], + pd.Timestamp(prev_dt, tz='UTC'))[0] + entries[asset] = (dt_value, val) + closes.append(val) + continue + return np.array(closes) + + def volumes(self, assets, dt): + """ + The volume field's aggregation returns the sum of all volumes + between the market open and the `dt` + If there has been no data on or before the `dt` the volume is 0. + + Returns + ------- + np.array with dtype=int64, in order of assets parameter. + """ + market_open, prev_dt, dt_value, entries = self._prelude(dt, 'volume') + + volumes = [] + session_label = self._trading_calendar.minute_to_session_label(dt) + + for asset in assets: + if not asset.is_alive_for_session(session_label): + volumes.append(0) + continue + + if prev_dt is None: + val = self._minute_reader.get_value(asset, dt, 'volume') + entries[asset] = (dt_value, val) + volumes.append(val) + continue + else: + try: + last_visited_dt, last_total = entries[asset] + if last_visited_dt == dt_value: + volumes.append(last_total) + continue + elif last_visited_dt == prev_dt: + val = self._minute_reader.get_value( + asset, dt, 'volume') + val += last_total + entries[asset] = (dt_value, val) + volumes.append(val) + continue + else: + after_last = pd.Timestamp( + last_visited_dt + self._one_min, tz='UTC') + window = self._minute_reader.load_raw_arrays( + ['volume'], + after_last, + dt, + [asset], + )[0] + val = np.nansum(window) + last_total + entries[asset] = (dt_value, val) + volumes.append(val) + continue + except KeyError: + window = self._minute_reader.load_raw_arrays( + ['volume'], + market_open, + dt, + [asset], + )[0] + val = np.nansum(window) + entries[asset] = (dt_value, val) + volumes.append(val) + continue + return np.array(volumes)