diff --git a/setup.py b/setup.py index cd3b79cc..1189f990 100644 --- a/setup.py +++ b/setup.py @@ -78,18 +78,26 @@ class LazyBuildExtCommandClass(dict): return build_ext +def window_specialization(typename): + """Make an extension for an AdjustedArrayWindow specialization.""" + return Extension( + 'zipline.lib._{name}window'.format(name=typename), + ['zipline/lib/_{name}window.pyx'.format(name=typename)], + depends=['zipline/lib/_windowtemplate.pxi'], + ) + + ext_modules = [ Extension('zipline.assets._assets', ['zipline/assets/_assets.pyx']), Extension('zipline.assets.continuous_futures', ['zipline/assets/continuous_futures.pyx']), Extension('zipline.lib.adjustment', ['zipline/lib/adjustment.pyx']), Extension('zipline.lib._factorize', ['zipline/lib/_factorize.pyx']), - Extension( - 'zipline.lib._float64window', ['zipline/lib/_float64window.pyx'] - ), - Extension('zipline.lib._int64window', ['zipline/lib/_int64window.pyx']), - Extension('zipline.lib._uint8window', ['zipline/lib/_uint8window.pyx']), - Extension('zipline.lib._labelwindow', ['zipline/lib/_labelwindow.pyx']), + window_specialization('float64'), + window_specialization('int64'), + window_specialization('int64'), + window_specialization('uint8'), + window_specialization('label'), Extension('zipline.lib.rank', ['zipline/lib/rank.pyx']), Extension('zipline.data._equities', ['zipline/data/_equities.pyx']), Extension('zipline.data._adjustments', ['zipline/data/_adjustments.pyx']), diff --git a/zipline/assets/assets.py b/zipline/assets/assets.py index d92b5b6a..e3876f59 100644 --- a/zipline/assets/assets.py +++ b/zipline/assets/assets.py @@ -363,7 +363,10 @@ class AssetFinder(object): """ Retrieve the Asset for a given sid. """ - return self.retrieve_all((sid,), default_none=default_none)[0] + try: + return self._asset_cache[sid] + except KeyError: + return self.retrieve_all((sid,), default_none=default_none)[0] def retrieve_all(self, sids, default_none=False): """ diff --git a/zipline/data/data_portal.py b/zipline/data/data_portal.py index 4946cf1d..20f43102 100644 --- a/zipline/data/data_portal.py +++ b/zipline/data/data_portal.py @@ -17,7 +17,7 @@ from operator import mul from logbook import Logger import numpy as np -from numpy import float64, int64 +from numpy import float64, int64, nan import pandas as pd from pandas import isnull from pandas.tslib import normalize_date @@ -634,7 +634,7 @@ class DataPortal(object): if column == "last_traded": last_traded_dt = reader.get_last_traded_dt(asset, dt) - if pd.isnull(last_traded_dt): + if isnull(last_traded_dt): return pd.NaT else: return last_traded_dt @@ -858,34 +858,45 @@ class DataPortal(object): raise Exception( "Only 1d and 1m are supported for forward-filling.") - dt_to_fill = df.index[0] + assets_with_leading_nan = np.where(isnull(df.iloc[0]))[0] - perspective_dt = df.index[-1] - assets_with_leading_nan = np.where(pd.isnull(df.iloc[0]))[0] - for missing_loc in assets_with_leading_nan: - asset = assets[missing_loc] - previous_dt = self.get_last_traded_dt( - asset, dt_to_fill, data_frequency) - if pd.isnull(previous_dt): - continue - previous_value = self.get_adjusted_value( + history_start, history_end = df.index[[0, -1]] + initial_values = [] + for asset in df.columns[assets_with_leading_nan]: + last_traded = self.get_last_traded_dt( asset, - field, - previous_dt, - perspective_dt, + history_start, data_frequency, ) - df.iloc[0, missing_loc] = previous_value + if isnull(last_traded): + initial_values.append(nan) + else: + initial_values.append( + self.get_adjusted_value( + asset, + field, + dt=last_traded, + perspective_dt=history_end, + data_frequency=data_frequency, + ) + ) + # Set leading values for assets that were missing data, then ffill. + df.ix[0, assets_with_leading_nan] = np.array( + initial_values, + dtype=np.float64 + ) df.fillna(method='ffill', inplace=True) + # forward-filling will incorrectly produce values after the end of + # an asset's lifetime, so write NaNs back over the asset's + # end_date. + normed_index = df.index.normalize() for asset in df.columns: - if df.index[-1] >= asset.end_date: + if history_end >= asset.end_date: # if the window extends past the asset's end date, set # all post-end-date values to NaN in that asset's series - series = df[asset] - series[series.index.normalize() > asset.end_date] = np.NaN - + df.loc[normed_index > asset.end_date, asset] = nan return df def _get_minute_window_for_assets(self, assets, field, minutes_for_window): @@ -912,10 +923,6 @@ class DataPortal(object): ------- A numpy array with requested values. """ - return self._get_minute_window_data(assets, field, minutes_for_window) - - def _get_minute_window_data( - self, assets, field, minutes_for_window): return self._minute_history_loader.history(assets, minutes_for_window, field, diff --git a/zipline/data/history_loader.py b/zipline/data/history_loader.py index fb4b062f..61816376 100644 --- a/zipline/data/history_loader.py +++ b/zipline/data/history_loader.py @@ -18,8 +18,8 @@ from abc import ( abstractproperty, ) +from numpy import concatenate from lru import LRU -from numpy import around, hstack from pandas import isnull from pandas.tslib import normalize_date from toolz import sliding_window @@ -268,7 +268,7 @@ class SlidingWindow(object): def __init__(self, window, size, cal_start, offset): self.window = window self.cal_start = cal_start - self.current = around(next(window), 3) + self.current = next(window) self.offset = offset self.most_recent_ix = self.cal_start + size @@ -283,7 +283,7 @@ class SlidingWindow(object): return self.current target = end_ix - self.cal_start - self.offset + 1 - self.current = around(self.window.seek(target), 3) + self.current = self.window.seek(target) self.most_recent_ix = end_ix return self.current @@ -526,7 +526,11 @@ class HistoryLoader(with_metaclass(ABCMeta)): field, is_perspective_after) end_ix = self._calendar.get_loc(dts[-1]) - return hstack([window.get(end_ix) for window in block]) + + return concatenate( + [window.get(end_ix) for window in block], + axis=1, + ).round(3) class DailyHistoryLoader(HistoryLoader): diff --git a/zipline/data/us_equity_pricing.py b/zipline/data/us_equity_pricing.py index 33700d37..5165e19c 100644 --- a/zipline/data/us_equity_pricing.py +++ b/zipline/data/us_equity_pricing.py @@ -636,21 +636,21 @@ class BcolzDailyBarReader(SessionBarReader): try: ix = self.sid_day_index(asset, search_day) except NoDataBeforeDate: - return None + return NaT except NoDataAfterDate: prev_day_ix = self.sessions.get_loc(search_day) - 1 if prev_day_ix > -1: search_day = self.sessions[prev_day_ix] continue except NoDataOnDate: - return None + return NaT if volumes[ix] != 0: return search_day prev_day_ix = self.sessions.get_loc(search_day) - 1 if prev_day_ix > -1: search_day = self.sessions[prev_day_ix] else: - return None + return NaT def sid_day_index(self, sid, day): """ diff --git a/zipline/lib/_windowtemplate.pxi b/zipline/lib/_windowtemplate.pxi index 245df436..5b555be0 100644 --- a/zipline/lib/_windowtemplate.pxi +++ b/zipline/lib/_windowtemplate.pxi @@ -15,6 +15,10 @@ from numpy cimport ndarray from numpy import asanyarray +class Exhausted(Exception): + pass + + cdef class AdjustedArrayWindow: """ An iterator representing a moving view over an AdjustedArray. @@ -34,11 +38,11 @@ cdef class AdjustedArrayWindow: readonly databuffer data readonly dict view_kwargs readonly Py_ssize_t window_length - Py_ssize_t anchor, next_anchor, max_anchor, next_adj + Py_ssize_t anchor, max_anchor, next_adj Py_ssize_t perspective_offset dict adjustments list adjustment_indices - ndarray last_out + ndarray output def __cinit__(self, databuffer data not None, @@ -52,7 +56,7 @@ cdef class AdjustedArrayWindow: self.adjustments = adjustments self.adjustment_indices = sorted(adjustments, reverse=True) self.window_length = window_length - self.anchor = window_length + offset + self.anchor = window_length + offset - 1 if perspective_offset > 1: # Limit perspective_offset to 1. # To support an offset greater than 1, work must be done to @@ -63,11 +67,10 @@ cdef class AdjustedArrayWindow: "is perspective_offset={0}".format( perspective_offset)) self.perspective_offset = perspective_offset - self.next_anchor = self.anchor self.max_anchor = data.shape[0] self.next_adj = self.pop_next_adj() - self.last_out = None + self.output = None cdef pop_next_adj(self): """ @@ -82,54 +85,61 @@ cdef class AdjustedArrayWindow: return self def __next__(self): + try: + self._tick_forward(1) + except Exhausted: + raise StopIteration() + + self._update_output() + return self.output + + def seek(self, Py_ssize_t target_anchor): + cdef: + Py_ssize_t anchor = self.anchor + + if target_anchor < anchor: + raise Exception('Can not access data after window has passed.') + + if target_anchor == anchor: + return self.output + + self._tick_forward(target_anchor - anchor) + self._update_output() + + return self.output + + cdef inline _tick_forward(self, int N): cdef: object adjustment - ndarray out - Py_ssize_t start, anchor - dict view_kwargs + Py_ssize_t anchor = self.anchor + Py_ssize_t target = anchor + N - anchor = self.anchor = self.next_anchor - if anchor > self.max_anchor: - raise StopIteration() + if target > self.max_anchor: + raise Exhausted() # Apply any adjustments that occured before our current anchor. # Equivalently, apply any adjustments known **on or before** the date # for which we're calculating a window. - while self.next_adj < anchor + self.perspective_offset: + while self.next_adj < target + self.perspective_offset: for adjustment in self.adjustments[self.next_adj]: adjustment.mutate(self.data) self.next_adj = self.pop_next_adj() - start = anchor - self.window_length + self.anchor = target - # If our data is a custom subclass of ndarray, preserve that subclass - # by using asanyarray instead of asarray. - out = asanyarray(self.data[start:self.anchor]) - view_kwargs = self.view_kwargs + cdef inline _update_output(self): + cdef: + ndarray new_out + Py_ssize_t anchor = self.anchor + dict view_kwargs = self.view_kwargs + + new_out = asanyarray(self.data[anchor - self.window_length:anchor]) if view_kwargs: - out = out.view(**view_kwargs) - out.setflags(write=False) - - self.next_anchor = self.anchor + 1 - self.last_out = out - return out - - def seek(self, target_anchor): - cdef ndarray out = None - - if target_anchor < self.anchor: - raise Exception('Can not access data after window has passed.') - - if target_anchor == self.anchor: - return self.last_out - - while self.anchor < target_anchor: - out = next(self) - - self.last_out = out - return out + new_out = new_out.view(**view_kwargs) + new_out.setflags(write=False) + self.output = new_out def __repr__(self): return "<%s: window_length=%d, anchor=%d, max_anchor=%d, dtype=%r>" % (