From 5f9d0a148d703941758eb780b73cc120cadc7d6f Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 15 Apr 2016 15:49:04 -0400 Subject: [PATCH] BUG: Prevent out of order history arrays. Fix a bug where if history were called with assets `[1, 2]` and then subsequently, `[2, 1]`, the loader would return the cached array in order for `[1, 2]`. Instead cache an AdjustedArray for each asset, then when a history window is requested, check if each asset has a sufficient cache, and if not then read values for the assets which are missing or need to be refreshed. An added benefit of this change is that if a subsequent call to history has a smaller number of assets than the previous, no new data needs to be read from disk. e.g. a call with assets `[1, 2, 3]` and then `[1, 2]` would use the cached values for `1` and `2` from the first call. Conversely, if the second call has more assets, then only the data for the new assets needs to be retrieved. e.g. a history with `[1, 2]`, then `[1, 2, 3]` would only need (assuming `1` and `2` have not expired) to retrieve data for `3`. Unfortunately, the benefit here is not great because `load_raw_arrays` is optimized for reading many assets, and pulls the entire daily bar dataset into memory. This change makes tuning `load_raw_arrays` so that faster reads (e.g. by slicing from the carray for each asset, instead of pulling all data into a numpy array), when only a few assets are requested, more beneficial than it would have been previously. --- tests/test_history.py | 30 +++++ zipline/data/us_equity_loader.py | 202 +++++++++++++++--------------- zipline/data/us_equity_pricing.py | 13 +- 3 files changed, 145 insertions(+), 100 deletions(-) diff --git a/tests/test_history.py b/tests/test_history.py index bb59ee03..9343f6c0 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -1584,6 +1584,36 @@ class DailyEquityHistoryTestCase(HistoryTestCaseBase): "close" )[self.ASSET2] + def test_history_window_different_order(self): + """ + Prevent regression on a bug where the passing the same assets, but + in a different order would return a history window with the values, + but not the keys, in order of the first history call. + """ + # Both ASSET1 and ASSET2 have trades on this date. + day = self.ASSET2.end_date + + window_1 = self.data_portal.get_history_window( + [self.ASSET1, self.ASSET2], + day, + 4, + "1d", + "close" + ) + + window_2 = self.data_portal.get_history_window( + [self.ASSET2, self.ASSET1], + day, + 4, + "1d", + "close" + ) + + np.testing.assert_almost_equal(window_1[self.ASSET1].values, + window_2[self.ASSET1].values) + np.testing.assert_almost_equal(window_1[self.ASSET2].values, + window_2[self.ASSET2].values) + class MinuteToDailyAggregationTestCase(WithBcolzMinutes, ZiplineTestCase): diff --git a/zipline/data/us_equity_loader.py b/zipline/data/us_equity_loader.py index cb9a9a14..f3392b63 100644 --- a/zipline/data/us_equity_loader.py +++ b/zipline/data/us_equity_loader.py @@ -19,10 +19,10 @@ from abc import ( ) from cachetools import LRUCache -from numpy import dtype, around +from numpy import dtype, around, hstack from pandas.tslib import normalize_date -from six import iteritems, with_metaclass +from six import with_metaclass from zipline.pipeline.data.equity_pricing import USEquityPricing from zipline.lib._float64window import AdjustedArrayWindow as Float64Window @@ -80,12 +80,16 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): adjustment_reader : SQLiteAdjustmentReader Reader for adjustment data. """ - def __init__(self, env, reader, adjustment_reader): + FIELDS = ('open', 'high', 'low', 'close', 'volume') + + def __init__(self, env, reader, adjustment_reader, sid_cache_size=1000): self.env = env self._reader = reader self._adjustments_reader = adjustment_reader - # TODO: Split cache into 'by column' and 'by sid'. - self._window_blocks = ExpiringCache(LRUCache(maxsize=5)) + self._window_blocks = { + field: ExpiringCache(LRUCache(maxsize=sid_cache_size)) + for field in self.FIELDS + } @abstractproperty def _prefetch_length(self): @@ -99,7 +103,7 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): def _array(self, start, end, assets, field): pass - def _get_adjustments_in_range(self, assets, dts, field): + def _get_adjustments_in_range(self, asset, dts, field): """ Get the Float64Multiply objects to pass to an AdjustedArrayWindow. @@ -116,9 +120,8 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): Parameters ---------- - assets : iterable of Asset + asset : Asset The assets for which to get adjustments. - days : iterable of datetime64-like The days for which adjustment data is needed. field : str @@ -128,79 +131,72 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): ------- out : The adjustments as a dict of loc -> Float64Multiply """ - sids = {int(asset): i for i, asset in enumerate(assets)} + sid = int(asset) start = normalize_date(dts[0]) end = normalize_date(dts[-1]) adjs = {} - for sid, i in iteritems(sids): - if field != 'volume': - mergers = self._adjustments_reader.get_adjustments_for_sid( - 'mergers', sid) - for m in mergers: - dt = m[0] - if start < dt <= end: - end_loc = dts.searchsorted(dt) - mult = Float64Multiply(0, - end_loc - 1, - i, - i, - m[1]) - try: - adjs[end_loc].append(mult) - except KeyError: - adjs[end_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) - mult = Float64Multiply(0, - end_loc - 1, - i, - i, - d[1]) - try: - adjs[end_loc].append(mult) - except KeyError: - adjs[end_loc] = [mult] - splits = self._adjustments_reader.get_adjustments_for_sid( - 'splits', sid) - for s in splits: - dt = s[0] - if field == 'volume': - ratio = 1.0 / s[1] - else: - ratio = s[1] + if field != 'volume': + mergers = self._adjustments_reader.get_adjustments_for_sid( + 'mergers', sid) + for m in mergers: + dt = m[0] if start < dt <= end: end_loc = dts.searchsorted(dt) mult = Float64Multiply(0, end_loc - 1, - i, - i, - ratio) + 0, + 0, + m[1]) try: adjs[end_loc].append(mult) except KeyError: adjs[end_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) + mult = Float64Multiply(0, + end_loc - 1, + 0, + 0, + d[1]) + try: + adjs[end_loc].append(mult) + except KeyError: + adjs[end_loc] = [mult] + splits = self._adjustments_reader.get_adjustments_for_sid( + 'splits', sid) + for s in splits: + dt = s[0] + if field == 'volume': + ratio = 1.0 / s[1] + else: + ratio = s[1] + if start < dt <= end: + end_loc = dts.searchsorted(dt) + mult = Float64Multiply(0, + end_loc - 1, + 0, + 0, + ratio) + try: + adjs[end_loc].append(mult) + except KeyError: + adjs[end_loc] = [mult] return adjs - def _ensure_sliding_window( - self, assets, dts, field): + def _ensure_sliding_windows(self, assets, dts, field): """ - Ensure that there is a Float64Multiply window that can provide data - for the given parameters. + Ensure that there is a Float64Multiply window for each asset that can + provide data for the given parameters. If the corresponding window for the (assets, len(dts), field) does not exist, then create a new one. If a corresponding window does exist for (assets, len(dts), field), but can not provide data for the current dts range, then create a new one and replace the expired window. - WARNING: A simulation with a high variance of assets, may cause - unbounded growth of floating windows stored in `_window_blocks`. - There should be some regular clean up of the cache, if stale windows - prevent simulations from completing because of memory constraints. - Parameters ---------- assets : iterable of Assets @@ -214,48 +210,58 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): Returns ------- - out : Float64Window with sufficient data so that the window can - provide `get` for the index corresponding with the last value in `dts` + out : list of Float64Window with sufficient data so that each asset's + window can provide `get` for the index corresponding with the last + value in `dts` """ end = dts[-1] size = len(dts) - assets_key = frozenset(assets) - try: - return self._window_blocks.get((assets_key, field, size), end) - except KeyError: - pass + asset_windows = {} + needed_assets = [] + for asset in assets: + try: + asset_windows[asset] = self._window_blocks[field].get( + (asset, size), end) + except KeyError: + needed_assets.append(asset) - start = dts[0] + if needed_assets: + start = dts[0] - offset = 0 - start_ix = self._calendar.get_loc(start) - end_ix = self._calendar.get_loc(end) + offset = 0 + start_ix = self._calendar.get_loc(start) + end_ix = self._calendar.get_loc(end) - cal = self._calendar - prefetch_end_ix = min(end_ix + self._prefetch_length, len(cal) - 1) - prefetch_end = cal[prefetch_end_ix] - prefetch_dts = cal[start_ix:prefetch_end_ix + 1] - array = self._array(prefetch_dts, assets, field) - if self._adjustments_reader: - adjs = self._get_adjustments_in_range(assets, prefetch_dts, field) - else: - adjs = {} - if field == 'volume': - array = array.astype('float64') - dtype_ = dtype('float64') + cal = self._calendar + prefetch_end_ix = min(end_ix + self._prefetch_length, len(cal) - 1) + prefetch_end = cal[prefetch_end_ix] + prefetch_dts = cal[start_ix:prefetch_end_ix + 1] + prefetch_len = len(prefetch_dts) + array = self._array(prefetch_dts, needed_assets, field) + if field == 'volume': + array = array.astype('float64') + dtype_ = dtype('float64') - window = Float64Window( - array, - dtype_, - adjs, - offset, - size - ) - block = SlidingWindow(window, size, start_ix, offset) - self._window_blocks.set((assets_key, field, size), - block, - prefetch_end) - return block + for i, asset in enumerate(needed_assets): + if self._adjustments_reader: + adjs = self._get_adjustments_in_range( + asset, prefetch_dts, field) + else: + adjs = {} + window = Float64Window( + array[:, i].reshape(prefetch_len, 1), + dtype_, + adjs, + offset, + size + ) + sliding_window = SlidingWindow(window, size, start_ix, offset) + asset_windows[asset] = sliding_window + self._window_blocks[field].set((asset, size), + sliding_window, + prefetch_end) + + return [asset_windows[asset] for asset in assets] def history(self, assets, dts, field): """ @@ -278,9 +284,9 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): ------- out : np.ndarray with shape(len(days between start, end), len(assets)) """ - block = self._ensure_sliding_window(assets, dts, field) + block = self._ensure_sliding_windows(assets, dts, field) end_ix = self._calendar.get_loc(dts[-1]) - return block.get(end_ix) + return hstack([window.get(end_ix) for window in block]) class USEquityDailyHistoryLoader(USEquityHistoryLoader): diff --git a/zipline/data/us_equity_pricing.py b/zipline/data/us_equity_pricing.py index 5cfefa3b..ac1d3bb6 100644 --- a/zipline/data/us_equity_pricing.py +++ b/zipline/data/us_equity_pricing.py @@ -39,6 +39,7 @@ from numpy import ( issubdtype, nan, uint32, + zeros, ) from pandas import ( DataFrame, @@ -648,8 +649,16 @@ class PanelDailyBarReader(DailyBarReader): col_names = [col.name for col in columns] cal = self._calendar index = cal[cal.slice_indexer(start_date, end_date)] - result = self.panel.loc[assets, start_date:end_date, col_names] - return result.reindex_axis(index, 1).values + shape = (len(index), len(assets)) + results = [] + for col in col_names: + outbuf = zeros(shape=shape) + for i, asset in enumerate(assets): + data = self.panel.loc[asset, start_date:end_date, col] + data = data.reindex_axis(index).values + outbuf[:, i] = data + results.append(outbuf) + return results def spot_price(self, sid, day, colname): """