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

ENH: Overhaul logic in HistoryContainer.

Browse Source 
Updates `HistoryContainer.roll` to handle cases where no data is present for
the period being rolled.

We now only forward-fill the `price` field when `ffill` is specified.
This commit is contained in:
Scott Sanderson
2014-06-03 15:26:55 -04:00
parent 15f1947652
commit 3a1fc1032e
2 changed files with 104 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/history/history.py
+11 -1
View File
@@ -219,6 +219,8 @@ class HistorySpec(object):
result frames.
"""
FORWARD_FILLABLE = frozenset({'price'})
@classmethod
def spec_key(cls, bar_count, freq_str, field, ffill):
"""
@@ -237,12 +239,20 @@ class HistorySpec(object):
# The field, e.g. 'price', 'volume', etc.
self.field = field
# Whether or not to forward fill the nan data.
self.ffill = ffill
self._ffill = ffill
# Calculate the cache key string once.
self.key_str = self.spec_key(
bar_count, frequency.freq_str, field, ffill)
@property
def ffill(self):
"""
Wrapper around ffill that returns False for fields which are not
forward-fillable.
"""
return self._ffill and self.field in self.FORWARD_FILLABLE
def __repr__(self):
return ''.join([self.__class__.__name__, "('", self.key_str, "')"])
zipline/history/history_container.py
+93 -73
View File
@@ -26,9 +26,10 @@ from . history import (
from zipline.finance import trading
from zipline.utils.data import RollingPanel
# The closing price is referred to by multiple names,
# allow both for price rollover logic etc.
CLOSING_PRICE_FIELDS = {'price', 'close_price'}
CLOSING_PRICE_FIELDS = frozenset({'price', 'close_price'})
def ffill_buffer_from_prior_values(field,
@@ -40,10 +41,6 @@ def ffill_buffer_from_prior_values(field,
digest frame if the buffer frame has leading NaNs.
"""
if field == 'volume':
# Volume is never forward-filled.
return buffer_frame
# Get values which are NaN at the beginning of the period.
first_bar = buffer_frame.iloc[0]
@@ -72,6 +69,35 @@ def ffill_buffer_from_prior_values(field,
return buffer_frame.ffill()
def ffill_digest_frame_from_prior_values(field, digest_frame, prior_values):
"""
Forward-fill a digest frame, falling back to the last known priof values if
necessary.
"""
if digest_frame is not None:
# Digest frame is None in the case that we only have length 1 history
# specs for a given frequency.
# It's possible that the first bar in our digest frame is storing NaN
# values. If so, check if we've tracked an older value and use that as
# an ffill value for the first bar.
first_bar = digest_frame.ix[0]
nan_sids = first_bar[first_bar.isnull()].index
for sid in nan_sids:
try:
# Only use prior value if it is before the index,
# so that a backfill does not accidentally occur.
if prior_values[field][sid]['dt'] <= digest_frame.index[0]:
digest_frame[sid][0] = prior_values[field][sid]['value']
except KeyError:
# Allow case where there is no previous value.
# e.g. with leading nans.
pass
digest_frame = digest_frame.ffill()
return digest_frame
def freq_str_and_bar_count(history_spec):
"""
Helper for getting the frequency string from a history spec.
@@ -329,41 +355,53 @@ class HistoryContainer(object):
index=self.fields,
columns=buffer_minutes.minor_axis)
if len(buffer_minutes.major_axis) > 0:
for field in self.fields:
if field in CLOSING_PRICE_FIELDS:
# Use the last price.
prices = buffer_minutes.ffill().ix[field, -1, :]
rolled.ix[field] = prices
elif field == 'open_price':
# Use the first price.
opens = buffer_minutes.ix['open_price', 0, :]
rolled.ix['open_price'] = opens
elif field == 'volume':
# Volume is the sum of the volumes during the
# course of the day
volumes = buffer_minutes.ix['volume'].apply(np.sum)
rolled.ix['volume'] = volumes
elif field == 'high':
# Use the highest high.
highs = buffer_minutes.ix['high'].apply(np.max)
rolled.ix['high'] = highs
elif field == 'low':
# Use the lowest low.
lows = buffer_minutes.ix['low'].apply(np.min)
rolled.ix['low'] = lows
for field in self.fields:
for sid, value in rolled.ix[field].iterkv():
if not np.isnan(value):
try:
prior_values = \
self.last_known_prior_values[field][sid]
except KeyError:
prior_values = {}
self.last_known_prior_values[field][sid] = \
prior_values
prior_values['dt'] = digest_dt
prior_values['value'] = value
if field in CLOSING_PRICE_FIELDS:
# Use the last close, or NaN if we have no minutes.
try:
prices = buffer_minutes.loc[field].ffill().iloc[-1]
except IndexError:
# Scalar assignment sets the value for all entries.
prices = np.nan
rolled.ix[field] = prices
elif field == 'open_price':
# Use the first open, or NaN if we have no minutes.
try:
opens = buffer_minutes.loc[field].bfill().iloc[0]
except IndexError:
# Scalar assignment sets the value for all entries.
opens = np.nan
rolled.ix['open_price'] = opens
elif field == 'volume':
# Volume is the sum of the volumes during the
# course of the period.
volumes = buffer_minutes.ix['volume'].sum().fillna(0)
rolled.ix['volume'] = volumes
elif field == 'high':
# Use the highest high.
highs = buffer_minutes.ix['high'].max()
rolled.ix['high'] = highs
elif field == 'low':
# Use the lowest low.
lows = buffer_minutes.ix['low'].min()
rolled.ix['low'] = lows
for sid, value in rolled.ix[field].iterkv():
if not np.isnan(value):
try:
prior_values = \
self.last_known_prior_values[field][sid]
except KeyError:
prior_values = {}
self.last_known_prior_values[field][sid] = \
prior_values
prior_values['dt'] = digest_dt
prior_values['value'] = value
digest_panel.add_frame(digest_dt, rolled)
@@ -377,6 +415,8 @@ class HistoryContainer(object):
field = history_spec.field
bar_count = history_spec.bar_count
do_ffill = history_spec.ffill
index = pd.to_datetime(index_at_dt(history_spec, algo_dt))
return_frame = self.return_frames[history_spec.key_str]
@@ -394,56 +434,36 @@ class HistoryContainer(object):
else:
digest_frame = None
if digest_frame is not None and history_spec.ffill:
# It's possible that the first bar in our digest frame is storing
# NaN values. If so, check if we've tracked an older value and use
# that as an ffill value for the first bar.
first_bar = digest_frame.ix[0]
nan_sids = first_bar[first_bar.isnull()].index
for sid in nan_sids:
try:
# Only use prior value if it is before the index,
# so that a backfill does not accidentally occur.
have_pre_frame_value = (
self.last_known_prior_values[field][sid]['dt'] <=
digest_frame.index[0]
)
if have_pre_frame_value:
digest_frame[sid][0] =\
self.last_known_prior_values[field][sid]['value']
except KeyError:
# Allow case where there is no previous value.
# e.g. with leading nans.
pass
digest_frame = digest_frame.ffill()
if digest_frame is not None:
return_frame.ix[:-1] = digest_frame.ix[:]
# Get minutes from our buffer panel to build the last row.
frequency = history_spec.frequency
buffer_frame = self.buffer_panel_minutes(
earliest_minute=self.cur_window_starts[frequency],
earliest_minute=self.cur_window_starts[history_spec.frequency],
)[field].copy()
if history_spec.ffill:
if do_ffill:
digest_frame = ffill_digest_frame_from_prior_values(
field,
digest_frame,
self.last_known_prior_values,
)
buffer_frame = ffill_buffer_from_prior_values(
field,
buffer_frame,
digest_frame,
self.last_known_prior_values,
)
if digest_frame is not None:
return_frame.ix[:-1] = digest_frame.ix[:]
if field == 'volume':
# This works for the day rollup, i.e. '1d',
# but '1m' will need to allow for 0 or nan minutes
return_frame.ix[algo_dt] = buffer_frame.sum()
return_frame.ix[algo_dt] = buffer_frame.fillna(0).sum()
elif field == 'high':
return_frame.ix[algo_dt] = buffer_frame.max()
elif field == 'low':
return_frame.ix[algo_dt] = buffer_frame.min()
elif field == 'open_price':
return_frame.ix[algo_dt] = buffer_frame.ix[0]
return_frame.ix[algo_dt] = buffer_frame.iloc[0]
else:
return_frame.ix[algo_dt] = buffer_frame.ix[algo_dt]
return_frame.ix[algo_dt] = buffer_frame.loc[algo_dt]
return return_frame