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)