wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-16 11:18:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

PERF: History Perf Enhancements

Browse Source 
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.
This commit is contained in:
Dale Jung
2015-02-11 06:25:53 -05:00
committed by Eddie Hebert
parent 33cef17396
commit 38e8d5214d
6 changed files with 446 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_history.py
+53
View File
@@ -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(
tests/test_munge.py
+59
View File
@@ -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)
tests/test_rolling_panel.py
+39
View File
@@ -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):
zipline/history/history_container.py
+154 -59
View File
@@ -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,
)
zipline/utils/data.py
+68 -13
View File
@@ -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
zipline/utils/munge.py
+73
View File
@@ -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)