diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index 81de1bed..0d2a158b 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -273,6 +273,11 @@ def uses_ufunc(data, *args, **kwargs): return np.log(data) +@batch_transform +def price_multiple(data, multiplier, extra_arg=1): + return data.price * multiplier * extra_arg + + class BatchTransformAlgorithm(TradingAlgorithm): def initialize(self, *args, **kwargs): self.refresh_period = kwargs.pop('refresh_period', 1) @@ -342,12 +347,6 @@ class BatchTransformAlgorithm(TradingAlgorithm): clean_nans=True ) - self.return_ticks = return_data( - refresh_period=self.refresh_period, - window_length=self.window_length, - create_panel=False - ) - self.return_not_full = return_data( refresh_period=0, window_length=self.window_length, @@ -360,6 +359,12 @@ class BatchTransformAlgorithm(TradingAlgorithm): clean_nans=False ) + self.price_multiple = price_multiple( + refresh_period=self.refresh_period, + window_length=self.window_length, + clean_nans=False + ) + self.iter = 0 self.set_slippage(FixedSlippage()) @@ -372,12 +377,33 @@ class BatchTransformAlgorithm(TradingAlgorithm): self.history_return_args.append( self.return_args_batch.handle_data( data, *self.args, **self.kwargs)) - self.history_return_ticks.append( - self.return_ticks.handle_data(data)) self.history_return_not_full.append( self.return_not_full.handle_data(data)) self.uses_ufunc.handle_data(data) + # check that calling transforms with the same arguments + # is idempotent + self.price_multiple.handle_data(data, 1, extra_arg=1) + + if self.price_multiple.full: + pre = len(self.price_multiple.ticks) + result1 = self.price_multiple.handle_data(data, 1, extra_arg=1) + post = len(self.price_multiple.ticks) + assert pre == post, "batch transform is appending redundant events" + result2 = self.price_multiple.handle_data(data, 1, extra_arg=1) + assert result1 is result2, "batch transform is not idempotent" + + # check that calling transform with the same data, but + # different supplemental arguments results in new + # results. + result3 = self.price_multiple.handle_data(data, 2, extra_arg=1) + assert result1 is not result3, \ + "batch transform is not updating for new args" + + result4 = self.price_multiple.handle_data(data, 1, extra_arg=2) + assert result1 is not result4,\ + "batch transform is not updating for new kwargs" + new_data = deepcopy(data) for sid in new_data: new_data[sid]['arbitrary'] = 123 diff --git a/zipline/transforms/utils.py b/zipline/transforms/utils.py index 72cc1f4c..7b47d9f7 100644 --- a/zipline/transforms/utils.py +++ b/zipline/transforms/utils.py @@ -228,8 +228,6 @@ class EventWindow(object): # Subclasses should override handle_add to define behavior for # adding new ticks. self.handle_add(event) - #if len(self.ticks) > self.window_length: - # import nose.tools; nose.tools.set_trace() # Clear out any expired events. # # oldest newest @@ -313,12 +311,11 @@ class BatchTransform(EventWindow): def __init__(self, func=None, - refresh_period=None, + refresh_period=0, window_length=None, clean_nans=True, sids=None, fields=None, - create_panel=True, compute_only_full=True): """Instantiate new batch_transform object. @@ -329,7 +326,7 @@ class BatchTransform(EventWindow): with the data panel and all args and kwargs supplied to the handle_data() call. refresh_period : int - Interval to call batch_transform function. + Interval to wait between advances in the window. window_length : int How many days the trailing window should have. clean_nans : bool @@ -342,12 +339,6 @@ class BatchTransform(EventWindow): Which fields to include in the moving window (e.g. 'price'). If not supplied, fields will be extracted from incoming events. - create_panel : bool - If True, will create a pandas panel every refresh - period and pass it to the user-defined function. - If False, will pass the underlying deque reference - directly to the function which will be significantly - faster. compute_only_full : bool Only call the user-defined function once the window is full. Returns None if window is not full yet. @@ -361,7 +352,6 @@ class BatchTransform(EventWindow): self.compute_transform_value = self.get_value self.clean_nans = clean_nans - self.create_panel = create_panel self.compute_only_full = compute_only_full self.sids = sids @@ -376,12 +366,15 @@ class BatchTransform(EventWindow): self.window_length = window_length self.trading_days_since_update = 0 self.trading_days_total = 0 + self.window = None self.full = False self.last_dt = None self.updated = False self.cached = None + self.last_args = None + self.last_kwargs = None # Data panel that provides bar information to fill in the window, # when no bar ticks are available from the data source generator @@ -411,9 +404,19 @@ class BatchTransform(EventWindow): # functionality to zipline if len(v)} - # append data frame to window. update() will call handle_add() and - # handle_remove() appropriately - self.update(event) + # only modify the trailing window if this is + # a new event. This is intended to make handle_data + # idempotent. + if event not in self.ticks: + # append data frame to window. update() will call handle_add() and + # handle_remove() appropriately, and self.updated + # will be modified based on the refresh_period + self.update(event) + else: + # we are recalculating based on an old event, so + # there is no change in the contents of the trailing + # window + self.updated = False # return newly computed or cached value return self.get_transform_value(*args, **kwargs) @@ -454,7 +457,6 @@ class BatchTransform(EventWindow): # to call the user-defined batch-transform with the most # recent datapanel self.updated = True - self.trading_days_since_update = 0 else: self.updated = False @@ -518,13 +520,26 @@ class BatchTransform(EventWindow): if self.compute_only_full and not self.full: return None + recalculate_needed = False if self.updated: - # Either create new pandas panel or pass ticks dequeue - # directly - data = self.get_data() if self.create_panel else self.ticks - self.cached = self.compute_transform_value(data, *args, - **kwargs) + # Create new pandas panel + self.window = self.get_data() + # reset our counter for refresh_period + self.trading_days_since_update = 0 + recalculate_needed = True + else: + recalculate_needed = \ + args != self.last_args or kwargs != self.last_kwargs + if recalculate_needed: + self.cached = self.compute_transform_value( + self.window, + *args, + **kwargs + ) + + self.last_args = args + self.last_kwargs = kwargs return self.cached def __call__(self, f):