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

ENH: Prep work for supporting '1m' history.

Browse Source 
Overhauls `HistoryContainer` in prep for support of more than one frequency.

Major changes:

   - Methods/variables referring to "day" have been renamed/generalized.
     - `current_day_panel` became `buffer_panel`, which is now a `RollingPanel`
     - `prior_day_panel` became a dictionary mapping `Frequency` objects to
       "digest panels", which are instances of `RollingPanel`.

   - Hard-coded daily rollover replaced with a notion of a "current window" for
     each unique frequency managed by the panel.

     - When the end of the current window is reached for a given frequency, we
       compute an aggregate bar (code refers to this as a "digest"), which is
       appended to a panel associated with that frequency.

     - Window rollover dates are managed by a pair of dictionaries,
       `cur_window_starts` and `cur_window_closes`.  The `Frequency` class is
       responsible for computing window bounds based on the open/close of the
       previous window.

   - Semantic change to the `open_price` field: `open_price` now always
     contains the price of the first trade occurring in the given window.
     Previously it contained the price of the first minute in the window,
     returning NaN it the security happened not to trade in the first minute.
This commit is contained in:
Scott Sanderson
2014-06-05 15:25:48 -04:00
parent b6e5345893
commit bad4c9a439
4 changed files with 556 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_history.py
+2 -4
View File
@@ -145,7 +145,7 @@ class TestHistoryContainer(TestCase):
field='price',
ffill=True
)
specs = {hash(spec): spec}
specs = {spec.key_str: spec}
initial_sids = [1, ]
initial_dt = pd.Timestamp(
'2013-06-28 9:31AM', tz='US/Eastern').tz_convert('UTC')
@@ -154,7 +154,7 @@ class TestHistoryContainer(TestCase):
specs, initial_sids, initial_dt)
bar_data = BarData()
container.update(bar_data, initial_dt)
# Since there was no backfill because of no db.
# And no first bar of data, so all values should be nans.
prices = container.get_history(spec, initial_dt)
@@ -169,7 +169,6 @@ class TestHistoryContainer(TestCase):
'price': 10,
'dt': second_bar_dt
}
container.update(bar_data, second_bar_dt)
prices = container.get_history(spec, second_bar_dt)
@@ -288,7 +287,6 @@ def handle_data(context, data):
# 12 13 14 15 16 17 18
# 19 20 21 22 23 24 25
# 26 27 28 29 30 31
start = pd.Timestamp('2006-03-20', tz='UTC')
end = pd.Timestamp('2006-03-21', tz='UTC')
zipline/history/history.py
+210 -43
View File
@@ -32,19 +32,183 @@ class Frequency(object):
Represents how the data is sampled, as specified by the algoscript
via units like "1d", "1m", etc.
Currently only one frequency is supported, "1d"
"1d" provides data keyed by closing, and the last minute of the current
day.
Currently only two frequencies are supported, "1d" and "1m"
- "1d" provides data at daily frequency, with the latest bar aggregating
the elapsed minutes of the (incomplete) current day
- "1m" provides data at minute frequency
"""
SUPPORTED_FREQUENCIES = frozenset({'1d'})
MAX_MINUTES = {'m': 1, 'd': 390}
def __init__(self, freq_str):
if freq_str not in self.SUPPORTED_FREQUENCIES:
raise ValueError(
"history frequency must be in {supported}".format(
supported=self.SUPPORTED_FREQUENCIES,
))
# The string the at the algoscript specifies.
# Hold onto to use a key for caching.
self.freq_str = freq_str
# num - The number of units of the frequency.
# unit_str - The unit type, e.g. 'd'
self.num, self.unit_str = parse_freq_str(freq_str)
def next_window_start(self, previous_window_close):
"""
Get the first minute of the window starting after a window that
finished on @previous_window_close.
"""
if self.unit_str == 'd':
return self.next_day_window_start(previous_window_close)
elif self.unit_str == 'm':
return self.next_minute_window_start(previous_window_close)
@staticmethod
def next_day_window_start(previous_window_close):
"""
Get the next day window start after @previous_window_close. This is
defined as the first market open strictly greater than
@previous_window_close.
"""
env = trading.environment
next_open, _ = env.next_open_and_close(previous_window_close)
return next_open
@staticmethod
def next_minute_window_start(previous_window_close):
"""
Get the next minute window start after @previous_window_close. This is
defined as the first market minute strictly greater than
@previous_window_close.
"""
env = trading.environment
return env.next_market_minute(previous_window_close)
def window_open(self, window_close):
"""
For a period ending on `window_end`, calculate the date of the first
minute bar that should be used to roll a digest for this frequency.
"""
if self.unit_str == 'd':
return self.day_window_open(window_close, self.num)
elif self.unit_str == 'm':
return self.minute_window_open(window_close, self.num)
def window_close(self, window_start):
"""
For a period starting on `window_start`, calculate the date of the last
minute bar that should be used to roll a digest for this frequency.
"""
if self.unit_str == 'd':
return self.day_window_close(window_start, self.num)
elif self.unit_str == 'm':
return self.minute_window_close(window_start, self.num)
@staticmethod
def day_window_open(window_close, num_days):
"""
Get the first minute for a daily window of length @num_days with last
minute @window_close. This is calculated by searching backward until
@num_days market_closes are encountered.
"""
env = trading.environment
open_ = env.open_close_window(
window_close,
1,
offset=-(num_days - 1)
).market_open.iloc[0]
return open_
@staticmethod
def minute_window_open(window_close, num_minutes):
"""
Get the first minute for a minutely window of length @num_minutes with
last minute @window_close.
This is defined as window_close if num_minutes == 1, and otherwise as
the N-1st market minute after @window_start.
"""
if num_minutes == 1:
# Short circuit this case.
return window_close
env = trading.environment
return env.market_minute_window(window_close, count=-num_minutes)[-1]
@staticmethod
def day_window_close(window_start, num_days):
"""
Get the last minute for a daily window of length @num_days with first
minute @window_start. This is calculated by searching forward until
@num_days market closes are encountered.
Examples:
window_start = Thursday March 2nd, 2006, 9:31 AM EST
num_days = 1
--> window_close = Thursday March 2nd, 2006, 4:00 PM EST
window_start = Thursday March 2nd, 2006, 3:59 AM EST
num_days = 1
--> window_close = Thursday March 2nd, 2006, 4:00 PM EST
window_start = Thursday March 2nd, 2006, 9:31 AM EST
num_days = 2
--> window_close = Friday March 2nd, 2006, 4:00 PM EST
window_start = Thursday March 2nd, 2006, 9:31 AM EST
num_days = 3
--> window_close = Monday March 6th, 2006, 4:00 PM EST
# Day before July 4th is an early close
window_start = Wednesday July 3rd, 2013, 9:31 AM EST
num_days = 1
--> window_close = Wednesday July 3rd, 2013, 1:00 PM EST
"""
env = trading.environment
close = env.open_close_window(
window_start,
1,
offset=num_days - 1
).market_close.iloc[0]
return close
@staticmethod
def minute_window_close(window_start, num_minutes):
"""
Get the last minute for a minutely window of length @num_minutes with
first minute @window_start.
This is defined as window_start if num_minutes == 1, and otherwise as
the N-1st market minute after @window_start.
"""
if num_minutes == 1:
# Short circuit this case.
return window_start
env = trading.environment
return env.market_minute_window(window_start, count=num_minutes)[-1]
@property
def max_minutes(self):
"""
The maximum number of minutes required to roll a bar at this frequency.
"""
return self.MAX_MINUTES[self.unit_str] * self.num
def __eq__(self, other):
return self.freq_str == other.freq_str
def __hash__(self):
return hash(self.freq_str)
def __repr__(self):
return ''.join([str(self.__class__.__name__),
"('", self.freq_str, "')"])
class HistorySpec(object):
"""
@@ -75,61 +239,64 @@ class HistorySpec(object):
# Whether or not to forward fill the nan data.
self.ffill = ffill
# How many trading days the spec needs to look back.
# Used by index creation to see how large of an overarching window
# is needed.
self.days_needed = calculate_days_needed(
self.bar_count, self.frequency)
# Calculate the cache key string once.
self.key_str = self.spec_key(
bar_count, frequency.freq_str, field, ffill)
def __repr__(self):
return ''.join([self.__class__.__name__, "('", self.key_str, "')"])
def calculate_days_needed(bar_count, freq):
""" Returns number trading days needed.
Overshoots so that we more than enough to sample from the current
frequency slot plus previous ones.
def days_index_at_dt(history_spec, algo_dt):
"""
if freq.unit_str == 'd':
return bar_count * freq.num
def days_index_at_dt(days_needed, algo_dt):
"""
The timestamps of previous days closes with the size of @days_needed
at @algo_dt.
Get the index of a frame to be used for a get_history call with daily
frequency.
"""
env = trading.environment
# Get the previous (bar_count - 1) days' worth of market closes.
day_delta = (history_spec.bar_count - 1) * history_spec.frequency.num
market_closes = env.open_close_window(
algo_dt,
day_delta,
offset=(-day_delta),
step=history_spec.frequency.num,
).market_close
latest_algo_dt = algo_dt
current_index = env.open_and_closes.index.searchsorted(algo_dt.date())
previous_days_num = days_needed - 1
previous_days = env.open_and_closes['market_close'][
current_index - previous_days_num:current_index]
# Append the current algo_dt as the last index value.
# Using the 'rawer' numpy array values here because of a bottleneck
# that appeared when using DatetimeIndex
return np.append(previous_days.values, latest_algo_dt)
return np.append(market_closes.values, algo_dt)
def minutes_index_at_dt(history_spec, algo_dt):
"""
Get the index of a frame to be used for a get_history_call with minutely
frequency.
"""
# TODO: This is almost certainly going to be too slow for production.
env = trading.environment
return env.market_minute_window(
algo_dt,
history_spec.bar_count,
step=-1,
)[::-1]
def index_at_dt(history_spec, algo_dt):
"""
The index, including @algo_dt at the given @algo_dt for the count
and frequency of the @history_spec.
Returns index of a frame returned by get_history() with the given
history_spec and algo_dt.
The resulting index `@history_spec.bar_count` bars, increasing in units of
`@history_spec.frequency`, terminating at the given @algo_dt.
Note: The last bar of the returned frame represents an as-of-yet incomplete
time window, so the delta between the last and second-to-last bars is
usually always less than `@history_spec.frequency` for frequencies greater
than 1m.
"""
days_index = days_index_at_dt(history_spec.days_needed, algo_dt)
frequency = history_spec.frequency
if frequency.unit_str == 'd':
index_of_algo_dt = days_index.searchsorted(algo_dt)
start_index = index_of_algo_dt + 1 - history_spec.bar_count
end_index = index_of_algo_dt + 1
return days_index[start_index:end_index]
return days_index_at_dt(history_spec, algo_dt)
elif frequency.unit_str == 'm':
return minutes_index_at_dt(history_spec, algo_dt)
zipline/history/history_container.py
+334 -160
View File
@@ -12,14 +12,15 @@
# 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.
from itertools import groupby
import numpy as np
import pandas as pd
from six import itervalues
from six import itervalues, iteritems, iterkeys
from . history import (
index_at_dt,
days_index_at_dt,
)
from zipline.finance import trading
@@ -30,38 +31,79 @@ from zipline.utils.data import RollingPanel
CLOSING_PRICE_FIELDS = {'price', 'close_price'}
def create_initial_day_panel(days_needed, fields, sids, dt):
index = days_index_at_dt(days_needed, dt)
# Use original index in case of 1 bar.
if days_needed != 1:
index = index[:-1]
window = len(index)
rp = RollingPanel(window, fields, sids)
for i, day in enumerate(index):
rp.index_buf[i] = day
rp.pos = window
return rp
def ffill_buffer_from_prior_values(field,
buffer_frame,
digest_frame,
pre_digest_values):
"""
Forward-fill a buffer frame, falling back to the end-of-period values of a
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]
def iter_nan_sids():
"""
Helper for iterating over the remaining nan sids in first_bar.
"""
return (sid for sid in first_bar[first_bar.isnull()].index)
# Try to fill with the last entry from the digest frame.
if digest_frame is not None:
# We don't store a digest frame for frequencies that only have a bar
# count of 1.
for sid in iter_nan_sids():
buffer_frame[sid][0] = digest_frame.ix[-1, sid]
# If we still have nan sids, try to fill with pre_digest_values.
for sid in iter_nan_sids():
prior_sid_value = pre_digest_values[field].get(sid)
if prior_sid_value:
# If the prior value is greater than the timestamp of our first
# bar.
if prior_sid_value.get('dt', first_bar.name) > first_bar.name:
buffer_frame[sid][0] = prior_sid_value.get('value', np.nan)
return buffer_frame.ffill()
def create_current_day_panel(fields, sids, dt):
# Can't use open_and_close since need to create enough space for a full
# day, even on a half day.
# Can now use mkt open and close, since we don't roll
env = trading.environment
index = env.market_minutes_for_day(dt)
return pd.Panel(items=fields, minor_axis=sids, major_axis=index)
def freq_str_and_bar_count(history_spec):
"""
Helper for getting the frequency string from a history spec.
"""
return (history_spec.frequency.freq_str, history_spec.bar_count)
def ffill_day_frame(field, day_frame, prior_day_frame):
# get values which are nan-at the beginning of the day
# and attempt to fill with the last close
first_bar = day_frame.ix[0]
nan_sids = first_bar[np.isnan(first_bar)]
for sid, _ in nan_sids.iterkv():
day_frame[sid][0] = prior_day_frame.ix[-1, sid]
if field != 'volume':
day_frame = day_frame.ffill()
return day_frame
def group_by_frequency(history_specs):
"""
Takes an iterable of history specs and returns a dictionary mapping unique
frequencies to a list of specs with that frequency.
Within each list, the HistorySpecs are sorted by ascending bar count.
Example:
[HistorySpec(3, '1d', 'price', True),
HistorySpec(2, '2d', 'open', True),
HistorySpec(2, '1d', 'open', False),
HistorySpec(5, '1m', 'open', True)]
yields
{Frequency('1d') : [HistorySpec(2, '1d', 'open', False)],
HistorySpec(3, '1d', 'price', True),
Frequency('2d') : [HistorySpec(2, '2d', 'open', True)],
Frequency('1m') : [HistorySpec(5, '1m', 'open', True)]}
"""
return {key: list(group)
for key, group in groupby(
sorted(history_specs, key=freq_str_and_bar_count),
key=lambda spec: spec.frequency)}
class HistoryContainer(object):
@@ -78,35 +120,105 @@ class HistoryContainer(object):
# History specs to be served by this container.
self.history_specs = history_specs
# The overaching panel needs to be large enough to contain the
# largest history spec
self.max_days_needed = max(spec.days_needed for spec
in itervalues(history_specs))
self.frequency_groups = \
group_by_frequency(itervalues(self.history_specs))
# The set of fields specified by all history specs
self.fields = set(spec.field for spec in itervalues(history_specs))
self.prior_day_panel = create_initial_day_panel(
self.max_days_needed, self.fields, initial_sids, initial_dt)
# This panel contains raw minutes for periods that haven't been fully
# completed. When a frequency period rolls over, these minutes are
# digested using some sort of aggregation call on the panel (e.g. `sum`
# for volume, `max` for high, `min` for low, etc.).
self.buffer_panel = self.create_buffer_panel(
initial_sids,
initial_dt,
)
# This panel contains the minutes for the current day.
# The value that is used is some sort of aggregation call on the
# panel, e.g. `sum` for volume, `max` for high, etc.
self.current_day_panel = create_current_day_panel(
self.fields, initial_sids, initial_dt)
# Dictionaries with Frequency objects as keys.
self.digest_panels, self.cur_window_starts, self.cur_window_closes = \
self.create_digest_panels(initial_sids, initial_dt)
# Populating initial frames here, so that the cost of creating the
# initial frames does not show up when profiling. These frames are
# cached since mid-stream creation of containing data frames on every
# bar is expensive.
self.create_return_frames(initial_dt)
# Helps prop up the prior day panel against having a nan, when the data
# has been seen.
self.last_known_prior_values = {field: {} for field in self.fields}
# Populating initial frames here, so that the cost of creating the
# initial frames does not show up when profiling get_y
# These frames are cached since mid-stream creation of containing
# data frames on every bar is expensive.
self.return_frames = {}
@property
def unique_frequencies(self):
"""
Return an iterator over all the unique frequencies serviced by this
container.
"""
return iterkeys(self.frequency_groups)
self.create_return_frames(initial_dt)
def create_digest_panels(self, initial_sids, initial_dt):
"""
Initialize a RollingPanel for each unique panel frequency being stored
by this container. Each RollingPanel pre-allocates enough storage
space to service the highest bar-count of any history call that it
serves.
Relies on the fact that group_by_frequency sorts the value lists by
ascending bar count.
"""
# Map from frequency -> first/last minute of the next digest to be
# rolled for that frequency.
first_window_starts = {}
first_window_closes = {}
# Map from frequency -> digest_panels.
panels = {}
for freq, specs in iteritems(self.frequency_groups):
# Relying on the sorting of group_by_frequency to get the spec
# requiring the largest number of bars.
largest_spec = specs[-1]
if largest_spec.bar_count == 1:
# No need to allocate a digest panel; this frequency will only
# ever use data drawn from self.buffer_panel.
env = trading.environment
first_window_closes[freq] = \
env.get_open_and_close(initial_dt)[1]
first_window_starts[freq] = \
freq.window_open(first_window_closes[freq])
continue
initial_dates = index_at_dt(largest_spec, initial_dt)
# Set up dates for our first digest roll, which is keyed to the
# close of the first entry in our initial index.
first_window_closes[freq] = initial_dates[0]
first_window_starts[freq] = freq.window_open(initial_dates[0])
rp = RollingPanel(len(initial_dates) - 1,
self.fields,
initial_sids)
panels[freq] = rp
return panels, first_window_starts, first_window_closes
def create_buffer_panel(self, initial_sids, initial_dt):
"""
Initialize a RollingPanel containing enough minutes to service all our
frequencies.
"""
max_bars_needed = max(freq.max_minutes
for freq in self.unique_frequencies)
rp = RollingPanel(
max_bars_needed,
self.fields,
initial_sids,
# Restrict the initial data down to just the fields being used in
# this container.
)
return rp
def create_return_frames(self, algo_dt):
"""
@@ -114,101 +226,146 @@ class HistoryContainer(object):
Called during init and at universe rollovers.
"""
for history_spec in itervalues(self.history_specs):
index = index_at_dt(history_spec, algo_dt)
index = pd.to_datetime(index)
self.return_frames = {}
for spec_key, history_spec in iteritems(self.history_specs):
index = pd.to_datetime(index_at_dt(history_spec, algo_dt))
frame = pd.DataFrame(
index=index,
columns=map(int, self.current_day_panel.minor_axis.values),
columns=map(int, self.buffer_panel.minor_axis.values),
dtype=np.float64)
self.return_frames[history_spec] = frame
self.return_frames[spec_key] = frame
def buffer_panel_minutes(self,
buffer_panel=None,
earliest_minute=None,
latest_minute=None):
"""
Get the minutes in @buffer_panel between @earliest_minute and
@last_minute, inclusive.
@buffer_panel can be a RollingPanel or a plain Panel. If a
RollingPanel is supplied, we call `get_current` to extract a Panel
object. If no panel is supplied, we use self.buffer_panel.
If no value is specified for @earliest_minute, use all the minutes we
have up until @latest minute.
If no value for @latest_minute is specified, use all values up until
the latest minute.
"""
buffer_panel = buffer_panel or self.buffer_panel
if isinstance(buffer_panel, RollingPanel):
buffer_panel = buffer_panel.get_current()
return buffer_panel.ix[:, earliest_minute:latest_minute, :]
def update(self, data, algo_dt):
"""
Takes the bar at @algo_dt's @data and adds to the current day panel.
Takes the bar at @algo_dt's @data, checks to see if we need to roll any
new digests, then adds new data to the buffer panel.
"""
self.check_and_roll(algo_dt)
self.update_digest_panels(algo_dt, self.buffer_panel)
fields = self.fields
field_data = {sid: {field: bar[field] for field in fields}
for sid, bar in data.iteritems()
if (bar
and
bar['dt'] == algo_dt
and
# Only use data which is keyed in the data panel.
# Prevents crashes due to custom data.
sid in self.current_day_panel.minor_axis)}
field_frame = pd.DataFrame(field_data)
self.current_day_panel.ix[:, algo_dt, :] = field_frame.T
frame = pd.DataFrame(
{sid: {field: bar[field] for field in fields}
for sid, bar in data.iteritems()
if (bar
and
bar['dt'] == algo_dt
and
# Only use data which is keyed in the data panel.
# Prevents crashes due to custom data.
sid in self.buffer_panel.minor_axis)})
self.buffer_panel.add_frame(algo_dt, frame)
def update_digest_panels(self, algo_dt, buffer_panel):
"""
Check whether @algo_dt is greater than cur_window_close for any of our
frequencies. If so, roll a digest for that frequency using data drawn
from @buffer panel and insert it into the appropriate digest panels.
"""
for frequency in self.unique_frequencies:
# We don't keep a digest panel if we only have a length-1 history
# spec for a given frequency
digest_panel = self.digest_panels.get(frequency, None)
while algo_dt > self.cur_window_closes[frequency]:
earliest_minute = self.cur_window_starts[frequency]
latest_minute = self.cur_window_closes[frequency]
minutes_to_process = self.buffer_panel_minutes(
buffer_panel,
earliest_minute=earliest_minute,
latest_minute=latest_minute,
)
# Create a digest from minutes_to_process and add it to
# digest_panel.
self.roll(frequency,
digest_panel,
minutes_to_process,
latest_minute)
# Update panel start/close for this frequency.
self.cur_window_starts[frequency] = \
frequency.next_window_start(latest_minute)
self.cur_window_closes[frequency] = \
frequency.window_close(self.cur_window_starts[frequency])
def roll(self, frequency, digest_panel, buffer_minutes, digest_dt):
"""
Package up minutes in @buffer_minutes insert that bar into
@digest_panel at index @last_minute, and update
self.cur_window_{starts|closes} for the given frequency.
"""
if digest_panel is None:
# This happens if the only spec we have at this frequency has a bar
# count of 1.
return
def roll(self, roll_dt):
env = trading.environment
# This should work for price, but not others, e.g.
# open.
# Get the most recent value.
rolled = pd.DataFrame(
index=self.current_day_panel.items,
columns=self.current_day_panel.minor_axis)
index=self.fields,
columns=buffer_minutes.minor_axis)
for field in self.fields:
if field in CLOSING_PRICE_FIELDS:
# Use the last price.
prices = self.current_day_panel.ffill().ix[field, -1, :]
rolled.ix[field] = prices
elif field == 'open_price':
# Use the first price.
opens = self.current_day_panel.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 = self.current_day_panel.ix['volume'].apply(np.sum)
rolled.ix['volume'] = volumes
elif field == 'high':
# Use the highest high.
highs = self.current_day_panel.ix['high'].apply(np.max)
rolled.ix['high'] = highs
elif field == 'low':
# Use the lowest low.
lows = self.current_day_panel.ix['low'].apply(np.min)
rolled.ix['low'] = lows
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 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'] = roll_dt
prior_values['value'] = value
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
self.prior_day_panel.add_frame(roll_dt, rolled)
# Create a new 'current day' collector.
next_day = env.next_trading_day(roll_dt)
if next_day:
# Only create the next panel if there is a next day.
# i.e. don't create the next panel on the last day of
# the backest/current day of live trading.
self.current_day_panel = create_current_day_panel(
self.fields,
# Will break on quarter rollover.
self.current_day_panel.minor_axis,
next_day)
def check_and_roll(self, algo_dt):
"""
Check whether the algo_dt is at the end of a day.
If it is, aggregate the day's minute data and store it in the prior
day panel.
"""
# Use a while loop to account for illiquid bars.
while algo_dt > self.current_day_panel.major_axis[-1]:
roll_dt = self.current_day_panel.major_axis[-1]
self.roll(roll_dt)
digest_panel.add_frame(digest_dt, rolled)
def get_history(self, history_spec, algo_dt):
"""
@@ -217,57 +374,74 @@ class HistoryContainer(object):
Selects from the overarching history panel the values for the
@history_spec at the given @algo_dt.
"""
field = history_spec.field
bar_count = history_spec.bar_count
index = pd.to_datetime(index_at_dt(history_spec, algo_dt))
return_frame = self.return_frames[history_spec.key_str]
index = index_at_dt(history_spec, algo_dt)
index = pd.to_datetime(index)
frame = self.return_frames[history_spec]
# Overwrite the index.
# Not worrying about values here since the values are overwritten
# in the next step.
frame.index = index
return_frame.index = index
prior_day_panel = self.prior_day_panel.get_current()
prior_day_frame = prior_day_panel[field].copy()
if history_spec.ffill:
first_bar = prior_day_frame.ix[0]
nan_sids = first_bar[first_bar.isnull()]
for sid, _ in nan_sids.iterkv():
if bar_count > 1:
# Get the last bar_count - 1 frames from our stored historical
# frames.
digest_panel = self.digest_panels[history_spec.frequency]\
.get_current()
digest_frame = digest_panel[field].copy().ix[1 - bar_count:]
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:
if (
# Only use prior value if it is before the index,
# so that a backfill does not accidentally occur.
# 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'] <=
prior_day_frame.index[0]):
prior_day_frame[sid][0] =\
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
prior_day_frame = prior_day_frame.ffill()
frame.ix[:-1] = prior_day_frame.ix[:]
digest_frame = digest_frame.ffill()
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],
)[field].copy()
# Copy the current day frame, since the fill behavior will mutate
# the values in the panel.
current_day_frame = self.current_day_panel[field][:algo_dt].copy()
if history_spec.ffill:
current_day_frame = ffill_day_frame(field,
current_day_frame,
prior_day_frame)
buffer_frame = ffill_buffer_from_prior_values(
field,
buffer_frame,
digest_frame,
self.last_known_prior_values,
)
if field == 'volume':
# This works for the day rollup, i.e. '1d',
# but '1m' will need to allow for 0 or nan minutes
frame.ix[algo_dt] = current_day_frame.sum()
return_frame.ix[algo_dt] = buffer_frame.sum()
elif field == 'high':
frame.ix[algo_dt] = current_day_frame.max()
return_frame.ix[algo_dt] = buffer_frame.max()
elif field == 'low':
frame.ix[algo_dt] = current_day_frame.min()
return_frame.ix[algo_dt] = buffer_frame.min()
elif field == 'open_price':
frame.ix[algo_dt] = current_day_frame.ix[0]
return_frame.ix[algo_dt] = buffer_frame.ix[0]
else:
frame.ix[algo_dt] = current_day_frame.ix[algo_dt]
return_frame.ix[algo_dt] = buffer_frame.ix[algo_dt]
return frame
return return_frame
zipline/utils/data.py
+10 -5
View File
@@ -32,8 +32,8 @@ class RollingPanel(object):
Restrictions: major_axis can only be a DatetimeIndex for now
"""
def __init__(self, window, items, sids, cap_multiple=2,
dtype=np.float64):
def __init__(self, window, items, sids, cap_multiple=2, dtype=np.float64):
self.pos = 0
self.window = window
@@ -49,9 +49,14 @@ class RollingPanel(object):
self.buffer = self._create_buffer()
def _create_buffer(self):
return pd.Panel(items=self.items, minor_axis=self.minor_axis,
major_axis=range(self.cap),
dtype=self.dtype)
panel = pd.Panel(
items=self.items,
minor_axis=self.minor_axis,
major_axis=range(self.cap),
dtype=self.dtype,
)
return panel
def _update_buffer(self, frame):
# Drop outdated, nan-filled minors (sids) and items (fields)