diff --git a/tests/test_history.py b/tests/test_history.py index 8bc3842d..6b7aec4d 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -220,6 +220,59 @@ class TestHistoryContainer(TestCase): check_frame_type=True, ) + def test_multiple_specs_on_same_bar(self): + """ + Test that a ffill and non ffill spec both get + the correct results when called on the same tick + """ + spec = history.HistorySpec( + bar_count=3, + frequency='1m', + field='price', + ffill=True, + data_frequency='minute' + ) + no_fill_spec = history.HistorySpec( + bar_count=3, + frequency='1m', + field='price', + ffill=False, + data_frequency='minute' + ) + + specs = {spec.key_str: spec, no_fill_spec.key_str: no_fill_spec} + initial_sids = [1, ] + initial_dt = pd.Timestamp( + '2013-06-28 9:31AM', tz='US/Eastern').tz_convert('UTC') + + container = HistoryContainer( + specs, initial_sids, initial_dt, 'minute' + ) + + bar_data = BarData() + container.update(bar_data, initial_dt) + # Add data on bar two of first day. + second_bar_dt = pd.Timestamp( + '2013-06-28 9:32AM', tz='US/Eastern').tz_convert('UTC') + bar_data[1] = { + 'price': 10, + 'dt': second_bar_dt + } + container.update(bar_data, second_bar_dt) + + third_bar_dt = pd.Timestamp( + '2013-06-28 9:33AM', tz='US/Eastern').tz_convert('UTC') + + del bar_data[1] + + # add nan for 3rd bar + container.update(bar_data, third_bar_dt) + prices = container.get_history(spec, third_bar_dt) + no_fill_prices = container.get_history(no_fill_spec, third_bar_dt) + self.assertEqual(prices.values[-1], 10) + self.assertTrue(np.isnan(no_fill_prices.values[-1]), + "Last price should be np.nan") + def test_container_nans_and_daily_roll(self): spec = history.HistorySpec( diff --git a/tests/test_munge.py b/tests/test_munge.py new file mode 100644 index 00000000..3fabc295 --- /dev/null +++ b/tests/test_munge.py @@ -0,0 +1,59 @@ +# +# Copyright 2015 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 random + +import pandas as pd +import numpy as np +from numpy.testing import assert_almost_equal + +from zipline.utils.munge import bfill, ffill + + +def test_bfill(): + # test ndim=1 + N = 100 + s = pd.Series(np.random.randn(N)) + mask = random.sample(range(N), 10) + s.iloc[mask] = np.nan + + correct = s.bfill().values + test = bfill(s.values) + assert_almost_equal(correct, test) + + # test ndim=2 + df = pd.DataFrame(np.random.randn(N, N)) + df.iloc[mask] = np.nan + correct = df.bfill().values + test = bfill(df.values) + assert_almost_equal(correct, test) + + +def test_ffill(): + # test ndim=1 + N = 100 + s = pd.Series(np.random.randn(N)) + mask = random.sample(range(N), 10) + s.iloc[mask] = np.nan + + correct = s.ffill().values + test = ffill(s.values) + assert_almost_equal(correct, test) + + # test ndim=2 + df = pd.DataFrame(np.random.randn(N, N)) + df.iloc[mask] = np.nan + correct = df.ffill().values + test = ffill(df.values) + assert_almost_equal(correct, test) diff --git a/tests/test_rolling_panel.py b/tests/test_rolling_panel.py index 60f42cfa..a8a3e3cf 100644 --- a/tests/test_rolling_panel.py +++ b/tests/test_rolling_panel.py @@ -90,6 +90,45 @@ class TestRollingPanel(unittest.TestCase): expected, ) + @with_environment() + def test_get_current_multiple_call_same_tick(self, env): + """ + In old get_current, each call the get_current would copy the data. Thus + changing that object would have no side effects. + + To keep the same api, make sure that the raw option returns a copy too. + """ + data_id = lambda values: values.__array_interface__['data'] + + items = ('a', 'b') + sids = (1, 2) + + dts = env.market_minute_window( + env.open_and_closes.market_open[0], 4, + ).values + rp = RollingPanel(2, items, sids, initial_dates=dts[1:-1]) + + frame = pd.DataFrame( + data=np.arange(4).reshape((2, 2)), + columns=sids, + index=items, + ) + + nan_arr = np.empty((2, 6)) + nan_arr.fill(np.nan) + + rp.add_frame(dts[-1], frame) + + # each get_current call makea a copy + cur = rp.get_current() + cur2 = rp.get_current() + assert data_id(cur.values) != data_id(cur2.values) + + # make sure raw follow same logic + raw = rp.get_current(raw=True) + raw2 = rp.get_current(raw=True) + assert data_id(raw) != data_id(raw2) + class TestMutableIndexRollingPanel(unittest.TestCase): diff --git a/zipline/history/history_container.py b/zipline/history/history_container.py index a71a52a6..87c352d8 100644 --- a/zipline/history/history_container.py +++ b/zipline/history/history_container.py @@ -25,6 +25,7 @@ from . history import HistorySpec from zipline.finance.trading import with_environment from zipline.utils.data import RollingPanel, _ensure_index +from zipline.utils.munge import ffill, bfill logger = logbook.Logger('History Container') @@ -38,25 +39,39 @@ def ffill_buffer_from_prior_values(freq, field, buffer_frame, digest_frame, - pv_frame): + pv_frame, + raw=False): """ Forward-fill a buffer frame, falling back to the end-of-period values of a digest frame if the buffer frame has leading NaNs. """ - nan_sids = buffer_frame.iloc[0].isnull() - if any(nan_sids) and len(digest_frame): + # convert to ndarray if necessary + digest_values = digest_frame + if raw and isinstance(digest_frame, pd.DataFrame): + digest_values = digest_frame.values + + buffer_values = buffer_frame + if raw and isinstance(buffer_frame, pd.DataFrame): + buffer_values = buffer_frame.values + + nan_sids = pd.isnull(buffer_values[0]) + if np.any(nan_sids) and len(digest_values): # If we have any leading nans in the buffer and we have a non-empty # digest frame, use the oldest digest values as the initial buffer # values. - buffer_frame.ix[0, nan_sids] = digest_frame.ix[-1, nan_sids] + buffer_values[0, nan_sids] = digest_values[-1, nan_sids] - nan_sids = buffer_frame.iloc[0].isnull() - if any(nan_sids): + nan_sids = pd.isnull(buffer_values[0]) + if np.any(nan_sids): # If we still have leading nans, fall back to the last known values # from before the digest. - buffer_frame.ix[0, nan_sids] = pv_frame.loc[ - (freq.freq_str, field), nan_sids - ] + key_loc = pv_frame.index.get_loc((freq.freq_str, field)) + filler = pv_frame.values[key_loc, nan_sids] + buffer_values[0, nan_sids] = filler + + if raw: + filled = ffill(buffer_values) + return filled return buffer_frame.ffill() @@ -64,18 +79,28 @@ def ffill_buffer_from_prior_values(freq, def ffill_digest_frame_from_prior_values(freq, field, digest_frame, - pv_frame): + pv_frame, + raw=False): """ Forward-fill a digest frame, falling back to the last known prior values if necessary. """ - nan_sids = digest_frame.iloc[0].isnull() - if any(nan_sids): + # convert to ndarray if necessary + values = digest_frame + if raw and isinstance(digest_frame, pd.DataFrame): + values = digest_frame.values + + nan_sids = pd.isnull(values[0]) + if np.any(nan_sids): # If we have any leading nans in the frame, use values from pv_frame to # seed values for those sids. - digest_frame.ix[0, nan_sids] = pv_frame.loc[ - (freq.freq_str, field), nan_sids - ] + key_loc = pv_frame.index.get_loc((freq.freq_str, field)) + filler = pv_frame.values[key_loc, nan_sids] + values[0, nan_sids] = filler + + if raw: + filled = ffill(values) + return filled return digest_frame.ffill() @@ -247,9 +272,14 @@ class HistoryContainer(object): dtype=np.float64, ) + _ffillable_fields = None + @property def ffillable_fields(self): - return self.fields.intersection(HistorySpec.FORWARD_FILLABLE) + if self._ffillable_fields is None: + fillables = self.fields.intersection(HistorySpec.FORWARD_FILLABLE) + self._ffillable_fields = fillables + return self._ffillable_fields @property def prior_values_index(self): @@ -344,6 +374,8 @@ class HistoryContainer(object): ls = list(self.fields) insort_left(ls, field) self.fields = pd.Index(ls) + # unset fillable fields cache + self._ffillable_fields = None self._realign_fields() self.last_known_prior_values = self.last_known_prior_values.reindex( @@ -616,32 +648,39 @@ class HistoryContainer(object): if bar_count == 1: # slicing with [1 - bar_count:] doesn't work when bar_count == 1, # so special-casing this. - return pd.DataFrame(index=[], columns=self.sids) + res = pd.DataFrame(index=[], columns=self.sids) + return res.values, res.index field = history_spec.field # Panel axes are (field, dates, sids). We want just the entries for # the requested field, the last (bar_count - 1) data points, and all # sids. - panel = self.digest_panels[history_spec.frequency].get_current() + digest_panel = self.digest_panels[history_spec.frequency] + frame = digest_panel.get_current(field, raw=True) if do_ffill: # Do forward-filling *before* truncating down to the requested # number of bars. This protects us from losing data if an illiquid # stock has a gap in its price history. - return ffill_digest_frame_from_prior_values( + filled = ffill_digest_frame_from_prior_values( history_spec.frequency, history_spec.field, - panel.loc[field], + frame, self.last_known_prior_values, + raw=True # Truncate only after we've forward-filled - ).iloc[1 - bar_count:] + ) + indexer = slice(1 - bar_count, None) + return filled[indexer], digest_panel.current_dates()[indexer] else: - return panel.ix[field, 1 - bar_count:, :] + indexer = slice(1 - bar_count, None) + return frame[indexer, :], digest_panel.current_dates()[indexer] def buffer_panel_minutes(self, buffer_panel, earliest_minute=None, - latest_minute=None): + latest_minute=None, + raw=False): """ Get the minutes in @buffer_panel between @earliest_minute and @latest_minute, inclusive. @@ -657,8 +696,10 @@ class HistoryContainer(object): the latest minute. """ if isinstance(buffer_panel, RollingPanel): - buffer_panel = buffer_panel.get_current() - + buffer_panel = buffer_panel.get_current(start=earliest_minute, + end=latest_minute, + raw=raw) + return buffer_panel # Using .ix here rather than .loc because loc requires that the keys # are actually in the index, whereas .ix returns all the values between # earliest_minute and latest_minute, which is what we want. @@ -724,14 +765,22 @@ class HistoryContainer(object): buffer_panel, earliest_minute=earliest_minute, latest_minute=latest_minute, + raw=True ) if digest_panel is not None: # Create a digest from minutes_to_process and add it to # digest_panel. + digest_frame = self.create_new_digest_frame( + minutes_to_process, + self.fields, + self.sids + ) digest_panel.add_frame( latest_minute, - self.create_new_digest_frame(minutes_to_process) + digest_frame, + self.fields, + self.sids ) # Update panel start/close for this frequency. @@ -740,51 +789,73 @@ class HistoryContainer(object): self.cur_window_closes[frequency] = \ frequency.window_close(self.cur_window_starts[frequency]) - def frame_to_series(self, field, frame): + def frame_to_series(self, field, frame, columns=None): """ Convert a frame with a DatetimeIndex and sid columns into a series with a sid index, using the aggregator defined by the given field. """ + if isinstance(frame, pd.DataFrame): + columns = frame.columns + frame = frame.values + if not len(frame): return pd.Series( data=(0 if field == 'volume' else np.nan), - index=frame.columns, - ) + index=columns, + ).values if field in ['price', 'close_price']: - return frame.ffill().iloc[-1].values + # shortcircuit for full last row + vals = frame[-1] + if np.all(~np.isnan(vals)): + return vals + return ffill(frame)[-1] elif field == 'open_price': - return frame.bfill().iloc[0].values + return bfill(frame)[0] elif field == 'volume': - return frame.sum().values + return np.nansum(frame, axis=0) elif field == 'high': - return frame.max().values + return np.nanmax(frame, axis=0) elif field == 'low': - return frame.min().values + return np.nanmin(frame, axis=0) else: raise ValueError("Unknown field {}".format(field)) - def aggregate_ohlcv_panel(self, fields, ohlcv_panel): + def aggregate_ohlcv_panel(self, + fields, + ohlcv_panel, + items=None, + minor_axis=None): """ Convert an OHLCV Panel into a DataFrame by aggregating each field's frame into a Series. """ - return pd.DataFrame( - [ - self.frame_to_series(field, ohlcv_panel.loc[field]) - for field in fields - ], - index=fields, - columns=ohlcv_panel.minor_axis, - ) + vals = ohlcv_panel + if isinstance(ohlcv_panel, pd.Panel): + vals = ohlcv_panel.values + items = ohlcv_panel.items + minor_axis = ohlcv_panel.minor_axis - def create_new_digest_frame(self, buffer_minutes): + data = [ + self.frame_to_series( + field, + vals[items.get_loc(field)], + minor_axis + ) + for field in fields + ] + return np.array(data) + + def create_new_digest_frame(self, buffer_minutes, items=None, + minor_axis=None): """ Package up minutes in @buffer_minutes into a single digest frame. """ return self.aggregate_ohlcv_panel( self.fields, buffer_minutes, + items=items, + minor_axis=minor_axis ) def update_last_known_values(self): @@ -798,15 +869,22 @@ class HistoryContainer(object): for frequency in self.unique_frequencies: digest_panel = self.digest_panels.get(frequency, None) if digest_panel: - oldest_known_values = digest_panel.oldest_frame() + oldest_known_values = digest_panel.oldest_frame(raw=True) else: - oldest_known_values = self.buffer_panel.oldest_frame() + oldest_known_values = self.buffer_panel.oldest_frame(raw=True) + oldest_vals = oldest_known_values + oldest_columns = self.fields for field in ffillable: - non_nan_sids = oldest_known_values[field].notnull() - self.last_known_prior_values.loc[ - (frequency.freq_str, field), non_nan_sids - ] = oldest_known_values[field].dropna() + f_idx = oldest_columns.get_loc(field) + field_vals = oldest_vals[f_idx] + # isnan would be fast, possible to use? + non_nan_sids = np.where(pd.notnull(field_vals)) + key = (frequency.freq_str, field) + key_loc = self.last_known_prior_values.index.get_loc(key) + self.last_known_prior_values.values[ + key_loc, non_nan_sids + ] = field_vals[non_nan_sids] def get_history(self, history_spec, algo_dt): """ @@ -819,14 +897,16 @@ class HistoryContainer(object): do_ffill = history_spec.ffill # Get our stored values from periods prior to the current period. - digest_frame = self.digest_bars(history_spec, do_ffill) + digest_frame, index = self.digest_bars(history_spec, do_ffill) # Get minutes from our buffer panel to build the last row of the # returned frame. - buffer_frame = self.buffer_panel_minutes( + buffer_panel = self.buffer_panel_minutes( self.buffer_panel, earliest_minute=self.cur_window_starts[history_spec.frequency], - )[field] + raw=True + ) + buffer_frame = buffer_panel[self.fields.get_loc(field)] if do_ffill: buffer_frame = ffill_buffer_from_prior_values( @@ -835,30 +915,45 @@ class HistoryContainer(object): buffer_frame, digest_frame, self.last_known_prior_values, + raw=True ) - last_period = self.frame_to_series(field, buffer_frame) - return fast_build_history_output(digest_frame, last_period, algo_dt) + last_period = self.frame_to_series(field, buffer_frame, self.sids) + return fast_build_history_output(digest_frame, + last_period, + algo_dt, + index=index, + columns=self.sids) -def fast_build_history_output(buffer_frame, last_period, algo_dt): +def fast_build_history_output(buffer_frame, + last_period, + algo_dt, + index=None, + columns=None): """ Optimized concatenation of DataFrame and Series for use in HistoryContainer.get_history. Relies on the fact that the input arrays have compatible shapes. """ + buffer_values = buffer_frame + if isinstance(buffer_frame, pd.DataFrame): + buffer_values = buffer_frame.values + index = buffer_frame.index + columns = buffer_frame.columns + return pd.DataFrame( data=np.vstack( [ - buffer_frame.values, + buffer_values, last_period, ] ), index=fast_append_date_to_index( - buffer_frame.index, + index, pd.Timestamp(algo_dt) ), - columns=buffer_frame.columns, + columns=columns, ) diff --git a/zipline/utils/data.py b/zipline/utils/data.py index 797493b8..d81fd87d 100644 --- a/zipline/utils/data.py +++ b/zipline/utils/data.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +import datetime + import numpy as np import pandas as pd from copy import deepcopy @@ -78,10 +80,12 @@ class RollingPanel(object): def start_date(self): return self.date_buf[self._start_index] - def oldest_frame(self): + def oldest_frame(self, raw=False): """ Get the oldest frame in the panel. """ + if raw: + return self.buffer.values[:, self._start_index, :] return self.buffer.iloc[:, self._start_index, :] def set_minor_axis(self, minor_axis): @@ -144,27 +148,71 @@ class RollingPanel(object): where = slice(self._start_index, self._start_index + delta) self.date_buf[where] = missing_dts - def add_frame(self, tick, frame): + def add_frame(self, tick, frame, minor_axis=None, items=None): """ """ if self._pos == self.cap: self._roll_data() - self.buffer.loc[:, self._pos, :] = frame.T.astype(self.dtype) + values = frame + if isinstance(frame, pd.DataFrame): + values = frame.values + + self.buffer.values[:, self._pos, :] = values.astype(self.dtype) self.date_buf[self._pos] = tick self._pos += 1 - def get_current(self): + def get_current(self, item=None, raw=False, start=None, end=None): """ Get a Panel that is the current data in view. It is not safe to persist these objects because internal data might change """ + item_indexer = slice(None) + if item: + item_indexer = self.items.get_loc(item) - where = slice(self._start_index, self._pos) - major_axis = pd.DatetimeIndex(deepcopy(self.date_buf[where]), tz='utc') - return pd.Panel(self.buffer.values[:, where, :], self.items, - major_axis, self.minor_axis, dtype=self.dtype) + start_index = self._start_index + end_index = self._pos + + # get inital date window + where = slice(start_index, end_index) + current_dates = self.date_buf[where] + + def convert_datelike_to_long(dt): + if isinstance(dt, pd.Timestamp): + return dt.asm8 + if isinstance(dt, datetime.datetime): + return np.datetime64(dt) + return dt + + # constrict further by date + if start: + start = convert_datelike_to_long(start) + start_index += current_dates.searchsorted(start) + + if end: + end = convert_datelike_to_long(end) + _end = current_dates.searchsorted(end, 'right') + end_index -= len(current_dates) - _end + + where = slice(start_index, end_index) + + values = self.buffer.values[item_indexer, where, :] + current_dates = self.date_buf[where] + + if raw: + # return copy so we can change it without side effects here + return values.copy() + + major_axis = pd.DatetimeIndex(deepcopy(current_dates), tz='utc') + if values.ndim == 3: + return pd.Panel(values, self.items, major_axis, self.minor_axis, + dtype=self.dtype) + + elif values.ndim == 2: + return pd.DataFrame(values, major_axis, self.minor_axis, + dtype=self.dtype) def set_current(self, panel): """ @@ -223,10 +271,12 @@ class MutableIndexRollingPanel(object): def _oldest_frame_idx(self): return max(self._pos - self._window, 0) - def oldest_frame(self): + def oldest_frame(self, raw=False): """ Get the oldest frame in the panel. """ + if raw: + return self.buffer.values[:, self._oldest_frame_idx(), :] return self.buffer.iloc[:, self._oldest_frame_idx(), :] def set_sids(self, sids): @@ -277,17 +327,22 @@ class MutableIndexRollingPanel(object): self.date_buf[:self._window] = self.date_buf[-self._window:] self._pos = self._window - def add_frame(self, tick, frame): + def add_frame(self, tick, frame, minor_axis=None, items=None): """ """ if self._pos == self.cap: self._roll_data() - if set(frame.columns).difference(set(self.minor_axis)) or \ - set(frame.index).difference(set(self.items)): + if isinstance(frame, pd.DataFrame): + minor_axis = frame.columns + items = frame.index + + if set(minor_axis).difference(set(self.minor_axis)) or \ + set(items).difference(set(self.items)): self._update_buffer(frame) - self.buffer.loc[:, self._pos, :] = frame.T.astype(self.dtype) + vals = frame.T.astype(self.dtype) + self.buffer.loc[:, self._pos, :] = vals self.date_buf[self._pos] = tick self._pos += 1 diff --git a/zipline/utils/munge.py b/zipline/utils/munge.py new file mode 100644 index 00000000..c2d13588 --- /dev/null +++ b/zipline/utils/munge.py @@ -0,0 +1,73 @@ +# +# Copyright 2015 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 pandas.core.common as com + + +def _interpolate(values, method, axis=None): + if values.ndim == 1: + axis = 0 + elif values.ndim == 2: + axis = 1 + else: + raise Exception("Cannot interpolate array with more than 2 dims") + + values = values.copy() + values = interpolate_2d(values, method, axis=axis) + return values + + +def interpolate_2d(values, method='pad', axis=0, limit=None, fill_value=None): + """ + Copied from the 0.15.2. This did not exist in 0.12.0. + + Differences: + - Don't depend on pad_2d and backfill_2d to return values + - Removed dtype kwarg. 0.12.0 did not have this option. + """ + transf = (lambda x: x) if axis == 0 else (lambda x: x.T) + + # reshape a 1 dim if needed + ndim = values.ndim + if values.ndim == 1: + if axis != 0: # pragma: no cover + raise AssertionError("cannot interpolate on a ndim == 1 with " + "axis != 0") + values = values.reshape(tuple((1,) + values.shape)) + + if fill_value is None: + mask = None + else: # todo create faster fill func without masking + mask = com.mask_missing(transf(values), fill_value) + + # Note: pad_2d and backfill_2d work inplace in 0.12.0 and 0.15.2 + # in 0.15.2 they also return a reference to values + if method == 'pad': + com.pad_2d(transf(values), limit=limit, mask=mask) + else: + com.backfill_2d(transf(values), limit=limit, mask=mask) + + # reshape back + if ndim == 1: + values = values[0] + + return values + + +def ffill(values, axis=None): + return _interpolate(values, 'pad', axis=axis) + + +def bfill(values, axis=None): + return _interpolate(values, 'bfill', axis=axis)