From 38e8d5214d46f089020703712dc6b3f4f6ee084d Mon Sep 17 00:00:00 2001 From: Dale Jung Date: Fri, 26 Dec 2014 17:54:04 -0500 Subject: [PATCH] PERF: History Perf Enhancements MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Limited use of `pandas` data structures in both `HistoryContainer` and `RollingPanel`. Where possible, methods were amended to return raw `ndarrays` with the indexing logic done separately. This allows us to cut down the number of times pandas objects are created both as returns and intermediate values. The separation of indexing from data access allowed us to minimize the times we’d make use of pandas indexes. This required that that certain methods like `NDFrame.ffill` be replaced with versions that work with `ndarrays`. Some of this was done via straight numpy methods and others by access pandas internal machinery. Outside of allowing us to use faster ndarrays, many of these function provided speedups over their pandas counterparts as we didn’t require the extra features like handling multiple dtypes. i.e. np.isnan is faster than pd.isnull, but only works with certain dtypes. --- tests/test_history.py | 53 +++++++ tests/test_munge.py | 59 ++++++++ tests/test_rolling_panel.py | 39 +++++ zipline/history/history_container.py | 213 +++++++++++++++++++-------- zipline/utils/data.py | 81 ++++++++-- zipline/utils/munge.py | 73 +++++++++ 6 files changed, 446 insertions(+), 72 deletions(-) create mode 100644 tests/test_munge.py create mode 100644 zipline/utils/munge.py 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)