From bad4c9a4398d7173ad6cc0f3e3d63f636ecc4c98 Mon Sep 17 00:00:00 2001 From: Scott Sanderson Date: Fri, 30 May 2014 18:30:25 -0400 Subject: [PATCH] ENH: Prep work for supporting '1m' history. 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. --- tests/test_history.py | 6 +- zipline/history/history.py | 253 +++++++++++--- zipline/history/history_container.py | 494 ++++++++++++++++++--------- zipline/utils/data.py | 15 +- 4 files changed, 556 insertions(+), 212 deletions(-) diff --git a/tests/test_history.py b/tests/test_history.py index f4e0e400..3ca7e10b 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -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') diff --git a/zipline/history/history.py b/zipline/history/history.py index dcf49694..9e7853de 100644 --- a/zipline/history/history.py +++ b/zipline/history/history.py @@ -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) diff --git a/zipline/history/history_container.py b/zipline/history/history_container.py index e7b74b86..ce2da464 100644 --- a/zipline/history/history_container.py +++ b/zipline/history/history_container.py @@ -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 diff --git a/zipline/utils/data.py b/zipline/utils/data.py index 67a3c4ae..5dbd8c31 100644 --- a/zipline/utils/data.py +++ b/zipline/utils/data.py @@ -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)