diff --git a/tests/test_history.py b/tests/test_history.py index 84255b27..26519444 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -158,7 +158,7 @@ class WithHistory(WithDataPortal): return pd.DataFrame([ { 'effective_date': str_to_seconds('2015-01-06'), - 'ratio': 0.5, + 'ratio': 0.25, 'sid': cls.SPLIT_ASSET_SID, }, { @@ -173,7 +173,7 @@ class WithHistory(WithDataPortal): return pd.DataFrame([ { 'effective_date': str_to_seconds('2015-01-06'), - 'ratio': 0.5, + 'ratio': 0.25, 'sid': cls.MERGER_ASSET_SID, }, { @@ -482,14 +482,15 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase): # Start values are crafted so that the thousands place are equal when # adjustments are applied correctly. - # The splits and mergers are defined as 2:1 splits, so the prices - # approximate that adjustment by halving the thousands place each day. + # The splits and mergers are defined as 4:1 then 2:1 ratios, so the + # prices approximate that adjustment by quartering and then halving + # the thousands place. data[cls.MERGER_ASSET_SID] = data[cls.SPLIT_ASSET_SID] = pd.concat(( create_minute_df_for_asset( cls.env, pd.Timestamp('2015-01-05', tz='UTC'), pd.Timestamp('2015-01-05', tz='UTC'), - start_val=4000), + start_val=8000), create_minute_df_for_asset( cls.env, pd.Timestamp('2015-01-06', tz='UTC'), @@ -499,6 +500,11 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase): cls.env, pd.Timestamp('2015-01-07', tz='UTC'), pd.Timestamp('2015-01-07', tz='UTC'), + start_val=1000), + create_minute_df_for_asset( + cls.env, + pd.Timestamp('2015-01-08', tz='UTC'), + pd.Timestamp('2015-01-08', tz='UTC'), start_val=1000) )) asset3 = cls.asset_finder.retrieve_asset(3) @@ -546,6 +552,129 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase): with self.assertRaises(HistoryInInitialize): test_algo.initialize() + def test_daily_splits_and_mergers(self): + # self.SPLIT_ASSET and self.MERGER_ASSET had splits/mergers + # on 1/6 and 1/7 + + jan5 = pd.Timestamp('2015-01-05', tz='UTC') + + for asset in [self.SPLIT_ASSET, self.MERGER_ASSET]: + # before any of the adjustments, 1/4 and 1/5 + window1 = self.data_portal.get_history_window( + [asset], + self.env.get_open_and_close(jan5)[1], + 2, + '1d', + 'close' + )[asset] + + np.testing.assert_array_equal(np.array([np.nan, 8389]), window1) + + # straddling the first event + window2 = self.data_portal.get_history_window( + [asset], + pd.Timestamp('2015-01-06 14:35', tz='UTC'), + 2, + '1d', + 'close' + )[asset] + + # Value from 1/5 should be quartered + np.testing.assert_array_equal( + [2097.25, + # Split occurs. The value of the thousands place should + # match. + 2004], + window2 + ) + + # straddling both events! + window3 = self.data_portal.get_history_window( + [asset], + pd.Timestamp('2015-01-07 14:35', tz='UTC'), + 3, + '1d', + 'close' + )[asset] + + np.testing.assert_array_equal( + [1048.625, 1194.50, 1004.0], + window3 + ) + + # after last event + window4 = self.data_portal.get_history_window( + [asset], + pd.Timestamp('2015-01-08 14:40', tz='UTC'), + 2, + '1d', + 'close' + )[asset] + + # should not be adjusted + np.testing.assert_array_equal([1389, 1009], window4) + + def test_daily_dividends(self): + # self.DIVIDEND_ASSET had dividends on 1/6 and 1/7 + + jan5 = pd.Timestamp('2015-01-05', tz='UTC') + asset = self.DIVIDEND_ASSET + + # before any of the dividends + window1 = self.data_portal.get_history_window( + [asset], + self.env.get_open_and_close(jan5)[1], + 2, + '1d', + 'close' + )[asset] + + np.testing.assert_array_equal(np.array([nan, 391]), window1) + + # straddling the first event + window2 = self.data_portal.get_history_window( + [asset], + pd.Timestamp('2015-01-06 14:35', tz='UTC'), + 2, + '1d', + 'close' + )[asset] + + np.testing.assert_array_equal( + [383.18, # 391 (last close) * 0.98 (first div) + # Dividend occurs prior. + 396], + window2 + ) + + # straddling both events! + window3 = self.data_portal.get_history_window( + [asset], + pd.Timestamp('2015-01-07 14:35', tz='UTC'), + 3, + '1d', + 'close' + )[asset] + + np.testing.assert_array_equal( + [367.853, # 391 (last close) * 0.98 * 0.96 (both) + 749.76, # 781 (last_close) * 0.96 (second div) + 786], # no adjustment + window3 + ) + + # after last event + window4 = self.data_portal.get_history_window( + [asset], + pd.Timestamp('2015-01-08 14:40', tz='UTC'), + 2, + '1d', + 'close' + )[asset] + + # should not be adjusted, should be 787 to 791 + np.testing.assert_array_equal([1171, 1181], window4) + def test_minute_before_assets_trading(self): # since asset2 and asset3 both started trading on 1/5/2015, let's do # some history windows that are completely before that @@ -728,7 +857,7 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase): )[asset] np.testing.assert_array_equal( - np.array(range(4380, 4390)), window1) + np.array(range(8380, 8390)), window1) # straddling the first event window2 = self.data_portal.get_history_window( @@ -741,11 +870,11 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase): # five minutes from 1/5 should be halved np.testing.assert_array_equal( - [2192.5, - 2193, - 2193.5, - 2194, - 2194.5, + [2096.25, + 2096.5, + 2096.75, + 2097, + 2097.25, # Split occurs. The value of the thousands place should # match. 2000, @@ -765,9 +894,9 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase): 'close' )[asset] - # first five minutes should be 4385-4390, but quartered + # first five minutes should be 4385-4390, but eigthed np.testing.assert_array_equal( - [1096.25, 1096.5, 1096.75, 1097, 1097.25], + [1048.125, 1048.25, 1048.375, 1048.5, 1048.625], window3[0:5] ) @@ -872,12 +1001,12 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase): bar_data = BarData(self.data_portal, lambda: current_dt, 'minute') adj_expected = { - 'open': np.arange(4381, 4391) / 2.0, - 'high': np.arange(4382, 4392) / 2.0, - 'low': np.arange(4379, 4389) / 2.0, - 'close': np.arange(4380, 4390) / 2.0, - 'volume': np.arange(4380, 4390) * 100 * 2.0, - 'price': np.arange(4380, 4390) / 2.0, + 'open': np.arange(8381, 8391) / 4.0, + 'high': np.arange(8382, 8392) / 4.0, + 'low': np.arange(8379, 8389) / 4.0, + 'close': np.arange(8380, 8390) / 4.0, + 'volume': np.arange(8380, 8390) * 100 * 4.0, + 'price': np.arange(8380, 8390) / 4.0, } expected = { @@ -1390,7 +1519,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase): )[asset] # first value should be halved, second value unadjusted - np.testing.assert_array_equal([1, 3], window2) + np.testing.assert_array_equal([0.5, 3], window2) window2_volume = self.data_portal.get_history_window( [asset], @@ -1402,7 +1531,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase): if asset == self.SPLIT_ASSET: # first value should be doubled, second value unadjusted - np.testing.assert_array_equal(window2_volume, [400, 300]) + np.testing.assert_array_equal(window2_volume, [800, 300]) elif asset == self.MERGER_ASSET: np.testing.assert_array_equal(window2_volume, [200, 300]) @@ -1415,7 +1544,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase): 'close' )[asset] - np.testing.assert_array_equal([0.5, 1.5, 4], window3) + np.testing.assert_array_equal([0.25, 1.5, 4], window3) window3_volume = self.data_portal.get_history_window( [asset], @@ -1426,7 +1555,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase): )[asset] if asset == self.SPLIT_ASSET: - np.testing.assert_array_equal(window3_volume, [800, 600, 400]) + np.testing.assert_array_equal(window3_volume, [1600, 600, 400]) elif asset == self.MERGER_ASSET: np.testing.assert_array_equal(window3_volume, [200, 300, 400]) diff --git a/zipline/data/data_portal.py b/zipline/data/data_portal.py index 2db66024..d5ec13ad 100644 --- a/zipline/data/data_portal.py +++ b/zipline/data/data_portal.py @@ -1336,7 +1336,8 @@ class DataPortal(object): self, assets, field, minutes_for_window): return self._equity_minute_history_loader.history(assets, minutes_for_window, - field) + field, + False) def _apply_all_adjustments(self, data, asset, dts, field, price_adj_factor=1.0): @@ -1452,7 +1453,8 @@ class DataPortal(object): if bar_count != 0: data = self._equity_history_loader.history(assets, days_in_window, - field) + field, + extra_slot) if extra_slot: return_array[:len(return_array) - 1, :] = data else: diff --git a/zipline/data/us_equity_loader.py b/zipline/data/us_equity_loader.py index ffb3d7e2..61043c59 100644 --- a/zipline/data/us_equity_loader.py +++ b/zipline/data/us_equity_loader.py @@ -103,7 +103,8 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): def _array(self, start, end, assets, field): pass - def _get_adjustments_in_range(self, asset, dts, field): + def _get_adjustments_in_range(self, asset, dts, field, + is_perspective_after): """ Get the Float64Multiply objects to pass to an AdjustedArrayWindow. @@ -126,6 +127,12 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): The days for which adjustment data is needed. field : str OHLCV field for which to get the adjustments. + is_perspective_after : bool + see: `USEquityHistoryLoader.history` + If True, the index at which the Multiply object is registered to + be popped is calculated so that it applies to the last slot in the + sliding window when the adjustment occurs immediately after the dt + that slot represents. Returns ------- @@ -142,30 +149,42 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): dt = m[0] if start < dt <= end: end_loc = dts.searchsorted(dt) + adj_loc = end_loc + if is_perspective_after: + # Set adjustment pop location so that it applies + # to last value if adjustment occurs immediately after + # the last slot. + adj_loc -= 1 mult = Float64Multiply(0, end_loc - 1, 0, 0, m[1]) try: - adjs[end_loc].append(mult) + adjs[adj_loc].append(mult) except KeyError: - adjs[end_loc] = [mult] + adjs[adj_loc] = [mult] divs = self._adjustments_reader.get_adjustments_for_sid( 'dividends', sid) for d in divs: dt = d[0] if start < dt <= end: end_loc = dts.searchsorted(dt) + adj_loc = end_loc + if is_perspective_after: + # Set adjustment pop location so that it applies + # to last value if adjustment occurs immediately after + # the last slot. + adj_loc -= 1 mult = Float64Multiply(0, end_loc - 1, 0, 0, d[1]) try: - adjs[end_loc].append(mult) + adjs[adj_loc].append(mult) except KeyError: - adjs[end_loc] = [mult] + adjs[adj_loc] = [mult] splits = self._adjustments_reader.get_adjustments_for_sid( 'splits', sid) for s in splits: @@ -176,18 +195,25 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): ratio = s[1] if start < dt <= end: end_loc = dts.searchsorted(dt) + adj_loc = end_loc + if is_perspective_after: + # Set adjustment pop location so that it applies + # to last value if adjustment occurs immediately after + # the last slot. + adj_loc -= 1 mult = Float64Multiply(0, end_loc - 1, 0, 0, ratio) try: - adjs[end_loc].append(mult) + adjs[adj_loc].append(mult) except KeyError: - adjs[end_loc] = [mult] + adjs[adj_loc] = [mult] return adjs - def _ensure_sliding_windows(self, assets, dts, field): + def _ensure_sliding_windows(self, assets, dts, field, + is_perspective_after): """ Ensure that there is a Float64Multiply window for each asset that can provide data for the given parameters. @@ -207,6 +233,8 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): in the calendar. field : str The OHLCV field for which to retrieve data. + is_perspective_after : bool + see: `USEquityHistoryLoader.history` Returns ------- @@ -218,10 +246,11 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): size = len(dts) asset_windows = {} needed_assets = [] + for asset in assets: try: asset_windows[asset] = self._window_blocks[field].get( - (asset, size), end) + (asset, size, is_perspective_after), end) except KeyError: needed_assets.append(asset) @@ -245,7 +274,7 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): for i, asset in enumerate(needed_assets): if self._adjustments_reader: adjs = self._get_adjustments_in_range( - asset, prefetch_dts, field) + asset, prefetch_dts, field, is_perspective_after) else: adjs = {} window = Float64Window( @@ -257,13 +286,14 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): ) sliding_window = SlidingWindow(window, size, start_ix, offset) asset_windows[asset] = sliding_window - self._window_blocks[field].set((asset, size), - sliding_window, - prefetch_end) + self._window_blocks[field].set( + (asset, size, is_perspective_after), + sliding_window, + prefetch_end) return [asset_windows[asset] for asset in assets] - def history(self, assets, dts, field): + def history(self, assets, dts, field, is_perspective_after): """ A window of pricing data with adjustments applied assuming that the end of the window is the day before the current simulation time. @@ -278,13 +308,70 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): in the calendar. field : str The OHLCV field for which to retrieve data. + is_perspective_after : bool + True, if the window is being viewed immediately after the last dt + in the sliding window. + False, if the window is viewed on the last dt. + This flag is used for handling the case where the last dt in the + requested window immediately precedes a corporate action, e.g.: + + - is_perspective_after is True + + When the viewpoint is after the last dt in the window, as when a + daily history window is accessed from a simulation that uses a + minute data frequency, the history call to this loader will not + include the current simulation dt. At that point in time, the raw + data for the last day in the window will require adjustment, so the + most recent adjustment with respect to the simulation time is + applied to the last dt in the requested window. + + An example equity which has a 0.5 split ratio dated for 05-27, + with the dts for a history call of 5 bars with a '1d' frequency at + 05-27 9:31. Simulation frequency is 'minute'. + + (In this case this function is called with 4 daily dts, and the + calling function is responsible for stitching back on the + 'current' dt) + + | | | | | last dt | <-- viewer is here | + | | 05-23 | 05-24 | 05-25 | 05-26 | 05-27 9:31 | + | raw | 10.10 | 10.20 | 10.30 | 10.40 | | + | adj | 5.05 | 5.10 | 5.15 | 5.25 | | + + The adjustment is applied to the last dt, 05-26, and all previous + dts. + + - is_perspective_after is False, daily + + When the viewpoint is the same point in time as the last dt in the + window, as when a daily history window is accessed from a + simulation that uses a daily data frequency, the history call will + include the current dt. At that point in time, the raw data for the + last day in the window will be post-adjustment, so no adjustment + is applied to the last dt. + + An example equity which has a 0.5 split ratio dated for 05-27, + with the dts for a history call of 5 bars with a '1d' frequency at + 05-27 0:00. Simulation frequency is 'daily'. + + | | | | | | <-- viewer is here | + | | | | | | last dt | + | | 05-23 | 05-24 | 05-25 | 05-26 | 05-27 | + | raw | 10.10 | 10.20 | 10.30 | 10.40 | 5.25 | + | adj | 5.05 | 5.10 | 5.15 | 5.20 | 5.25 | + + Adjustments are applied 05-23 through 05-26 but not to the last dt, + 05-27 Returns ------- out : np.ndarray with shape(len(days between start, end), len(assets)) """ - block = self._ensure_sliding_windows(assets, dts, field) + block = self._ensure_sliding_windows(assets, + dts, + field, + is_perspective_after) end_ix = self._calendar.get_loc(dts[-1]) return hstack([window.get(end_ix) for window in block])