Merge pull request #1394 from quantopian/downsample

Add Generic Downsampling to Pipeline

tests/pipeline/base.py

@@ -1,24 +1,23 @@
 """
-Base class for Pipeline API  unittests.
+Base class for Pipeline API unit tests.
 """
 from functools import wraps
 
 import numpy as np
 from numpy import arange, prod
-from pandas import date_range, Int64Index, DataFrame
+from pandas import DataFrame, Timestamp
 from six import iteritems
 
-from zipline.assets.synthetic import make_simple_equity_info
 from zipline.pipeline.engine import SimplePipelineEngine
-from zipline.pipeline import TermGraph
-from zipline.pipeline.term import AssetExists
+from zipline.pipeline import ExecutionPlan
+from zipline.pipeline.term import AssetExists, InputDates
 from zipline.testing import (
     check_arrays,
     ExplodingObject,
-    tmp_asset_finder,
 )
 from zipline.testing.fixtures import (
-    WithTradingCalendars,
+    WithAssetFinder,
+    WithTradingSessions,
     ZiplineTestCase,
 )
 
@@ -53,32 +52,26 @@ def with_defaults(**default_funcs):
 with_default_shape = with_defaults(shape=lambda self: self.default_shape)
 
 
-class BasePipelineTestCase(WithTradingCalendars, ZiplineTestCase):
+class BasePipelineTestCase(WithTradingSessions,
+                           WithAssetFinder,
+                           ZiplineTestCase):
+    START_DATE = Timestamp('2014', tz='UTC')
+    END_DATE = Timestamp('2014-12-31', tz='UTC')
+    ASSET_FINDER_EQUITY_SIDS = list(range(20))
 
     @classmethod
     def init_class_fixtures(cls):
         super(BasePipelineTestCase, cls).init_class_fixtures()
 
-        cls.__calendar = date_range('2014', '2015',
-                                    freq=cls.trading_calendar.day)
-        cls.__assets = assets = Int64Index(arange(1, 20))
-        cls.__tmp_finder_ctx = tmp_asset_finder(
-            equities=make_simple_equity_info(
-                assets,
-                cls.__calendar[0],
-                cls.__calendar[-1],
-            )
-        )
-        cls.__finder = cls.__tmp_finder_ctx.__enter__()
-        cls.__mask = cls.__finder.lifetimes(
-            cls.__calendar[-30:],
+        cls.default_asset_exists_mask = cls.asset_finder.lifetimes(
+            cls.nyse_sessions[-30:],
             include_start_date=False,
         )
 
     @property
     def default_shape(self):
         """Default shape for methods that build test data."""
-        return self.__mask.shape
+        return self.default_asset_exists_mask.shape
 
     def run_graph(self, graph, initial_workspace, mask=None):
         """
@@ -103,14 +96,17 @@ class BasePipelineTestCase(WithTradingCalendars, ZiplineTestCase):
         """
         engine = SimplePipelineEngine(
             lambda column: ExplodingObject(),
-            self.__calendar,
-            self.__finder,
+            self.nyse_sessions,
+            self.asset_finder,
         )
         if mask is None:
-            mask = self.__mask
+            mask = self.default_asset_exists_mask
 
         dates, assets, mask_values = explode(mask)
+
         initial_workspace.setdefault(AssetExists(), mask_values)
+        initial_workspace.setdefault(InputDates(), dates)
+
         return engine.compute_chunk(
             graph,
             dates,
@@ -118,15 +114,29 @@ class BasePipelineTestCase(WithTradingCalendars, ZiplineTestCase):
             initial_workspace,
         )
 
-    def check_terms(self, terms, expected, initial_workspace, mask):
+    def check_terms(self,
+                    terms,
+                    expected,
+                    initial_workspace,
+                    mask,
+                    check=check_arrays):
         """
         Compile the given terms into a TermGraph, compute it with
         initial_workspace, and compare the results with ``expected``.
         """
-        graph = TermGraph(terms)
+        start_date, end_date = mask.index[[0, -1]]
+        graph = ExecutionPlan(
+            terms,
+            all_dates=self.nyse_sessions,
+            start_date=start_date,
+            end_date=end_date,
+        )
+
         results = self.run_graph(graph, initial_workspace, mask)
         for key, (res, exp) in dzip_exact(results, expected).items():
-            check_arrays(res, exp)
+            check(res, exp)
+
+        return results
 
     def build_mask(self, array):
         """
@@ -138,13 +148,13 @@ class BasePipelineTestCase(WithTradingCalendars, ZiplineTestCase):
             array,
             # Use the **last** N dates rather than the first N so that we have
             # space for lookbacks.
-            index=self.__calendar[-ndates:],
-            columns=self.__assets[:nassets],
+            index=self.nyse_sessions[-ndates:],
+            columns=self.ASSET_FINDER_EQUITY_SIDS[:nassets],
             dtype=bool,
         )
 
     @with_default_shape
-    def arange_data(self, shape, dtype=float):
+    def arange_data(self, shape, dtype=np.float64):
         """
         Build a block of testing data from numpy.arange.
         """

tests/pipeline/test_downsampling.py

@@ -0,0 +1,686 @@
+"""
+Tests for Downsampled Filters/Factors/Classifiers
+"""
+import pandas as pd
+from pandas.util.testing import assert_frame_equal
+
+from zipline.pipeline import (
+    Pipeline,
+    CustomFactor,
+    CustomFilter,
+    CustomClassifier,
+)
+from zipline.pipeline.data.testing import TestingDataSet
+from zipline.pipeline.factors import SimpleMovingAverage
+from zipline.pipeline.filters.smoothing import All
+from zipline.testing import ZiplineTestCase, parameter_space
+from zipline.testing.fixtures import (
+    WithTradingSessions,
+    WithSeededRandomPipelineEngine,
+)
+from zipline.utils.input_validation import _qualified_name
+from zipline.utils.numpy_utils import int64_dtype
+
+
+class NDaysAgoFactor(CustomFactor):
+    inputs = [TestingDataSet.float_col]
+
+    def compute(self, today, assets, out, floats):
+        out[:] = floats[0]
+
+
+class NDaysAgoFilter(CustomFilter):
+    inputs = [TestingDataSet.bool_col]
+
+    def compute(self, today, assets, out, bools):
+        out[:] = bools[0]
+
+
+class NDaysAgoClassifier(CustomClassifier):
+    inputs = [TestingDataSet.categorical_col]
+    dtype = TestingDataSet.categorical_col.dtype
+
+    def compute(self, today, assets, out, cats):
+        out[:] = cats[0]
+
+
+class ComputeExtraRowsTestcase(WithTradingSessions, ZiplineTestCase):
+
+    DATA_MIN_DAY = pd.Timestamp('2012-06', tz='UTC')
+    DATA_MAX_DAY = pd.Timestamp('2015', tz='UTC')
+    TRADING_CALENDAR_STRS = ('NYSE',)
+
+    # Test with different window_lengths to ensure that window length is not
+    # used when calculating exra rows for the top-level term.
+    factor1 = TestingDataSet.float_col.latest
+    factor11 = NDaysAgoFactor(window_length=11)
+    factor91 = NDaysAgoFactor(window_length=91)
+
+    filter1 = TestingDataSet.bool_col.latest
+    filter11 = NDaysAgoFilter(window_length=11)
+    filter91 = NDaysAgoFilter(window_length=91)
+
+    classifier1 = TestingDataSet.categorical_col.latest
+    classifier11 = NDaysAgoClassifier(window_length=11)
+    classifier91 = NDaysAgoClassifier(window_length=91)
+
+    all_terms = [
+        factor1,
+        factor11,
+        factor91,
+        filter1,
+        filter11,
+        filter91,
+        classifier1,
+        classifier11,
+        classifier91,
+    ]
+
+    @parameter_space(
+        calendar_name=TRADING_CALENDAR_STRS,
+        base_terms=[
+            (factor1, factor11, factor91),
+            (filter1, filter11, filter91),
+            (classifier1, classifier11, classifier91),
+        ],
+        __fail_fast=True
+    )
+    def test_yearly(self, base_terms, calendar_name):
+        downsampled_terms = tuple(
+            t.downsample('year_start') for t in base_terms
+        )
+        all_terms = base_terms + downsampled_terms
+
+        all_sessions = self.trading_sessions[calendar_name]
+        end_session = all_sessions[-1]
+
+        years = all_sessions.year
+        sessions_in_2012 = all_sessions[years == 2012]
+        sessions_in_2013 = all_sessions[years == 2013]
+        sessions_in_2014 = all_sessions[years == 2014]
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the first date in 2014.  We shouldn't request any
+        # additional rows for the regular terms or the downsampled terms.
+        for i in range(0, 30, 5):
+            start_session = sessions_in_2014[i]
+            self.check_extra_row_calculations(
+                all_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land on the second date in 2014.  We should request one more extra
+        # row in the downsampled terms to push us back to the first date in
+        # 2014.
+        for i in range(0, 30, 5):
+            start_session = sessions_in_2014[i + 1]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i + 1,
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land on the last date of 2013. The downsampled terms should request
+        # enough extra rows to push us back to the start of 2013.
+        for i in range(0, 30, 5):
+            start_session = sessions_in_2014[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(sessions_in_2013),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land on the last date of 2012. The downsampled terms should request
+        # enough extra rows to push us back to the first known date, which is
+        # in the middle of 2012
+        for i in range(0, 30, 5):
+            start_session = sessions_in_2013[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(sessions_in_2012),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+    @parameter_space(
+        calendar_name=TRADING_CALENDAR_STRS,
+        base_terms=[
+            (factor1, factor11, factor91),
+            (filter1, filter11, filter91),
+            (classifier1, classifier11, classifier91),
+        ],
+        __fail_fast=True
+    )
+    def test_quarterly(self, calendar_name, base_terms):
+        downsampled_terms = tuple(
+            t.downsample('quarter_start') for t in base_terms
+        )
+        all_terms = base_terms + downsampled_terms
+
+        # This region intersects with Q4 2013, Q1 2014, and Q2 2014.
+        tmp = self.trading_sessions[calendar_name]
+        all_sessions = tmp[tmp.slice_indexer('2013-12-15', '2014-04-30')]
+        end_session = all_sessions[-1]
+
+        months = all_sessions.month
+        Q4_2013 = all_sessions[months == 12]
+        Q1_2014 = all_sessions[(months == 1) | (months == 2) | (months == 3)]
+        Q2_2014 = all_sessions[months == 4]
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the first date in Q2 2014.  We shouldn't request any
+        # additional rows for the regular terms or the downsampled terms.
+        for i in range(0, 15, 5):
+            start_session = Q2_2014[i]
+            self.check_extra_row_calculations(
+                all_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the second date in Q2 2014.
+        # The downsampled terms should request one more extra row.
+        for i in range(0, 15, 5):
+            start_session = Q2_2014[i + 1]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i + 1,
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the last date in Q1 2014.  The downsampled terms
+        # should request enough extra rows to push us back to the first date of
+        # Q1 2014.
+        for i in range(0, 15, 5):
+            start_session = Q2_2014[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(Q1_2014),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the last date in Q4 2013.  The downsampled terms
+        # should request enough extra rows to push us back to the first known
+        # date, which is in the middle of december 2013.
+        for i in range(0, 15, 5):
+            start_session = Q1_2014[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(Q4_2013),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+    @parameter_space(
+        calendar_name=TRADING_CALENDAR_STRS,
+        base_terms=[
+            (factor1, factor11, factor91),
+            (filter1, filter11, filter91),
+            (classifier1, classifier11, classifier91),
+        ],
+        __fail_fast=True
+    )
+    def test_monthly(self, calendar_name, base_terms):
+        downsampled_terms = tuple(
+            t.downsample('month_start') for t in base_terms
+        )
+        all_terms = base_terms + downsampled_terms
+
+        # This region intersects with Dec 2013, Jan 2014, and Feb 2014.
+        tmp = self.trading_sessions[calendar_name]
+        all_sessions = tmp[tmp.slice_indexer('2013-12-15', '2014-02-28')]
+        end_session = all_sessions[-1]
+
+        months = all_sessions.month
+        dec2013 = all_sessions[months == 12]
+        jan2014 = all_sessions[months == 1]
+        feb2014 = all_sessions[months == 2]
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the first date in feb 2014.  We shouldn't request any
+        # additional rows for the regular terms or the downsampled terms.
+        for i in range(0, 10, 2):
+            start_session = feb2014[i]
+            self.check_extra_row_calculations(
+                all_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land on the second date in feb 2014.  We should request one more
+        # extra row in the downsampled terms to push us back to the first date
+        # in 2014.
+        for i in range(0, 10, 2):
+            start_session = feb2014[i + 1]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i + 1,
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land on the last date of jan 2014. The downsampled terms should
+        # request enough extra rows to push us back to the start of jan 2014.
+        for i in range(0, 10, 2):
+            start_session = feb2014[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(jan2014),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land on the last date of dec 2013. The downsampled terms should
+        # request enough extra rows to push us back to the first known date,
+        # which is in the middle of december 2013.
+        for i in range(0, 10, 2):
+            start_session = jan2014[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(dec2013),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+    @parameter_space(
+        calendar_name=TRADING_CALENDAR_STRS,
+        base_terms=[
+            (factor1, factor11, factor91),
+            (filter1, filter11, filter91),
+            (classifier1, classifier11, classifier91),
+        ],
+        __fail_fast=True
+    )
+    def test_weekly(self, calendar_name, base_terms):
+        downsampled_terms = tuple(
+            t.downsample('week_start') for t in base_terms
+        )
+        all_terms = base_terms + downsampled_terms
+
+        #    December 2013
+        # Mo Tu We Th Fr Sa Su
+        #                    1
+        #  2  3  4  5  6  7  8
+        #  9 10 11 12 13 14 15
+        # 16 17 18 19 20 21 22
+        # 23 24 25 26 27 28 29
+        # 30 31
+
+        #     January 2014
+        # Mo Tu We Th Fr Sa Su
+        #        1  2  3  4  5
+        #  6  7  8  9 10 11 12
+        # 13 14 15 16 17 18 19
+        # 20 21 22 23 24 25 26
+        # 27 28 29 30 31
+
+        # This region intersects with the last full week of 2013, the week
+        # shared by 2013 and 2014, and the first full week of 2014.
+        tmp = self.trading_sessions[calendar_name]
+        all_sessions = tmp[tmp.slice_indexer('2013-12-27', '2014-01-12')]
+        end_session = all_sessions[-1]
+
+        week0 = all_sessions[
+            all_sessions.slice_indexer('2013-12-27', '2013-12-29')
+        ]
+        week1 = all_sessions[
+            all_sessions.slice_indexer('2013-12-30', '2014-01-05')
+        ]
+        week2 = all_sessions[
+            all_sessions.slice_indexer('2014-01-06', '2014-01-12')
+        ]
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the first date in week 2.  We shouldn't request any
+        # additional rows for the regular terms or the downsampled terms.
+        for i in range(3):
+            start_session = week2[i]
+            self.check_extra_row_calculations(
+                all_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the second date in week 2.  The downsampled terms
+        # should request one more extra row.
+        for i in range(3):
+            start_session = week2[i + 1]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i + 1,
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i,
+                expected_extra_rows=i,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the last date in week 1.  The downsampled terms
+        # should request enough extra rows to push us back to the first date of
+        # week 1.
+        for i in range(3):
+            start_session = week2[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(week1),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+        # Simulate requesting computation where the unaltered lookback would
+        # land exactly on the last date in week0.  The downsampled terms
+        # should request enough extra rows to push us back to the first known
+        # date, which is in the middle of december 2013.
+        for i in range(3):
+            start_session = week1[i]
+            self.check_extra_row_calculations(
+                downsampled_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + len(week0),
+            )
+            self.check_extra_row_calculations(
+                base_terms,
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows=i + 1,
+                expected_extra_rows=i + 1,
+            )
+
+    def check_extra_row_calculations(self,
+                                     terms,
+                                     all_sessions,
+                                     start_session,
+                                     end_session,
+                                     min_extra_rows,
+                                     expected_extra_rows):
+        """
+        Check that each term in ``terms`` computes an expected number of extra
+        rows for the given parameters.
+        """
+        for term in terms:
+            result = term.compute_extra_rows(
+                all_sessions,
+                start_session,
+                end_session,
+                min_extra_rows,
+            )
+            self.assertEqual(
+                result,
+                expected_extra_rows,
+                "Expected {} extra_rows from {}, but got {}.".format(
+                    expected_extra_rows,
+                    term,
+                    result,
+                )
+            )
+
+
+class DownsampledPipelineTestCase(WithSeededRandomPipelineEngine,
+                                  ZiplineTestCase):
+
+    # Extend into the last few days of 2013 to test year/quarter boundaries.
+    START_DATE = pd.Timestamp('2013-12-15', tz='UTC')
+
+    # Extend into the first few days of 2015 to test year/quarter boundaries.
+    END_DATE = pd.Timestamp('2015-01-06', tz='UTC')
+
+    ASSET_FINDER_EQUITY_SIDS = tuple(range(10))
+
+    def check_downsampled_term(self, term):
+
+        #       June 2014
+        # Mo Tu We Th Fr Sa Su
+        #                    1
+        #  2  3  4  5  6  7  8
+        #  9 10 11 12 13 14 15
+        # 16 17 18 19 20 21 22
+        # 23 24 25 26 27 28 29
+        # 30
+        all_sessions = self.nyse_sessions
+        compute_dates = all_sessions[
+            all_sessions.slice_indexer('2014-06-05', '2015-01-06')
+        ]
+        start_date, end_date = compute_dates[[0, -1]]
+
+        pipe = Pipeline({
+            'year': term.downsample(frequency='year_start'),
+            'quarter': term.downsample(frequency='quarter_start'),
+            'month': term.downsample(frequency='month_start'),
+            'week': term.downsample(frequency='week_start'),
+        })
+
+        # Raw values for term, computed each day from 2014 to the end of the
+        # target period.
+        raw_term_results = self.run_pipeline(
+            Pipeline({'term': term}),
+            start_date=pd.Timestamp('2014-01-02', tz='UTC'),
+            end_date=pd.Timestamp('2015-01-06', tz='UTC'),
+        )['term'].unstack()
+
+        expected_results = {
+            'year': (raw_term_results
+                     .groupby(pd.TimeGrouper('AS'))
+                     .first()
+                     .reindex(compute_dates, method='ffill')),
+            'quarter': (raw_term_results
+                        .groupby(pd.TimeGrouper('QS'))
+                        .first()
+                        .reindex(compute_dates, method='ffill')),
+            'month': (raw_term_results
+                      .groupby(pd.TimeGrouper('MS'))
+                      .first()
+                      .reindex(compute_dates, method='ffill')),
+            'week': (raw_term_results
+                     .groupby(pd.TimeGrouper('W', label='left'))
+                     .first()
+                     .reindex(compute_dates, method='ffill')),
+        }
+
+        results = self.run_pipeline(pipe, start_date, end_date)
+
+        for frequency in expected_results:
+            result = results[frequency].unstack()
+            expected = expected_results[frequency]
+            assert_frame_equal(result, expected)
+
+    def test_downsample_windowed_factor(self):
+        self.check_downsampled_term(
+            SimpleMovingAverage(
+                inputs=[TestingDataSet.float_col],
+                window_length=5,
+            )
+        )
+
+    def test_downsample_non_windowed_factor(self):
+        sma = SimpleMovingAverage(
+            inputs=[TestingDataSet.float_col],
+            window_length=5,
+        )
+
+        self.check_downsampled_term(((sma + sma) / 2).rank())
+
+    def test_downsample_windowed_filter(self):
+        sma = SimpleMovingAverage(
+            inputs=[TestingDataSet.float_col],
+            window_length=5,
+        )
+        self.check_downsampled_term(All(inputs=[sma.top(4)], window_length=5))
+
+    def test_downsample_nonwindowed_filter(self):
+        sma = SimpleMovingAverage(
+            inputs=[TestingDataSet.float_col],
+            window_length=5,
+        )
+        self.check_downsampled_term(sma > 5)
+
+    def test_downsample_windowed_classifier(self):
+
+        class IntSumClassifier(CustomClassifier):
+            inputs = [TestingDataSet.float_col]
+            window_length = 8
+            dtype = int64_dtype
+            missing_value = -1
+
+            def compute(self, today, assets, out, floats):
+                out[:] = floats.sum(axis=0).astype(int) % 4
+
+        self.check_downsampled_term(IntSumClassifier())
+
+    def test_downsample_nonwindowed_classifier(self):
+        sma = SimpleMovingAverage(
+            inputs=[TestingDataSet.float_col],
+            window_length=5,
+        )
+        self.check_downsampled_term(sma.quantiles(5))
+
+    def test_errors_on_bad_downsample_frequency(self):
+
+        f = NDaysAgoFactor(window_length=3)
+        with self.assertRaises(ValueError) as e:
+            f.downsample('bad')
+
+        expected = (
+            "{}() expected a value in "
+            "('month_start', 'quarter_start', 'week_start', 'year_start') "
+            "for argument 'frequency', but got 'bad' instead."
+        ).format(_qualified_name(f.downsample))
+        self.assertEqual(str(e.exception), expected)

tests/pipeline/test_engine.py

@@ -40,6 +40,7 @@ from six import iteritems, itervalues
 from toolz import merge
 
 from zipline.assets.synthetic import make_rotating_equity_info
+from zipline.errors import NoFurtherDataError
 from zipline.lib.adjustment import MULTIPLY
 from zipline.lib.labelarray import LabelArray
 from zipline.pipeline import CustomFactor, Pipeline
@@ -65,6 +66,7 @@ from zipline.pipeline.loaders.synthetic import (
     expected_bar_values_2d,
 )
 from zipline.pipeline.sentinels import NotSpecified
+from zipline.pipeline.term import InputDates
 from zipline.testing import (
     AssetID,
     AssetIDPlusDay,
@@ -81,7 +83,7 @@ from zipline.testing.fixtures import (
     ZiplineTestCase,
 )
 from zipline.utils.memoize import lazyval
-from zipline.utils.numpy_utils import bool_dtype
+from zipline.utils.numpy_utils import bool_dtype, datetime64ns_dtype
 
 
 class RollingSumDifference(CustomFactor):
@@ -206,6 +208,53 @@ class ConstantInputTestCase(WithTradingEnvironment, ZiplineTestCase):
         with self.assertRaisesRegexp(ValueError, msg):
             engine.run_pipeline(p, self.dates[2], self.dates[1])
 
+    def test_fail_usefully_on_insufficient_data(self):
+        loader = self.loader
+        engine = SimplePipelineEngine(
+            lambda column: loader, self.dates, self.asset_finder,
+        )
+
+        class SomeFactor(CustomFactor):
+            inputs = [USEquityPricing.close]
+            window_length = 10
+
+            def compute(self, today, assets, out, closes):
+                pass
+
+        p = Pipeline(columns={'t': SomeFactor()})
+
+        # self.dates[9] is the earliest date we should be able to compute.
+        engine.run_pipeline(p, self.dates[9], self.dates[9])
+
+        # We shouldn't be able to compute dates[8], since we only know about 8
+        # prior dates, and we need a window length of 10.
+        with self.assertRaises(NoFurtherDataError):
+            engine.run_pipeline(p, self.dates[8], self.dates[8])
+
+    def test_input_dates_provided_by_default(self):
+        loader = self.loader
+        engine = SimplePipelineEngine(
+            lambda column: loader, self.dates, self.asset_finder,
+        )
+
+        class TestFactor(CustomFactor):
+            inputs = [InputDates(), USEquityPricing.close]
+            window_length = 10
+            dtype = datetime64ns_dtype
+
+            def compute(self, today, assets, out, dates, closes):
+                first, last = dates[[0, -1], 0]
+                assert last == today.asm8
+                assert len(dates) == len(closes) == self.window_length
+                out[:] = first
+
+        p = Pipeline(columns={'t': TestFactor()})
+        results = engine.run_pipeline(p, self.dates[9], self.dates[10])
+
+        # All results are the same, so just grab one column.
+        column = results.unstack().iloc[:, 0].values
+        check_arrays(column, self.dates[:2].values)
+
     def test_same_day_pipeline(self):
         loader = self.loader
         engine = SimplePipelineEngine(

tests/pipeline/test_factor.py

@@ -26,7 +26,7 @@ from zipline.errors import UnknownRankMethod
 from zipline.lib.labelarray import LabelArray
 from zipline.lib.rank import masked_rankdata_2d
 from zipline.lib.normalize import naive_grouped_rowwise_apply as grouped_apply
-from zipline.pipeline import Classifier, Factor, Filter, TermGraph
+from zipline.pipeline import Classifier, Factor, Filter
 from zipline.pipeline.factors import (
     Returns,
     RSI,
@@ -37,7 +37,6 @@ from zipline.testing import (
     parameter_space,
     permute_rows,
 )
-from zipline.utils.functional import dzip_exact
 from zipline.utils.numpy_utils import (
     categorical_dtype,
     datetime64ns_dtype,
@@ -123,20 +122,18 @@ class FactorTestCase(BasePipelineTestCase):
         data = arange(25).reshape(5, 5)
         data[eye(5, dtype=bool)] = custom_missing_value
 
-        graph = TermGraph(
+        self.check_terms(
             {
                 'isnull': factor.isnull(),
                 'notnull': factor.notnull(),
-            }
-        )
-
-        results = self.run_graph(
-            graph,
+            },
+            {
+                'isnull': eye(5, dtype=bool),
+                'notnull': ~eye(5, dtype=bool),
+            },
             initial_workspace={factor: data},
             mask=self.build_mask(ones((5, 5))),
         )
-        check_arrays(results['isnull'], eye(5, dtype=bool))
-        check_arrays(results['notnull'], ~eye(5, dtype=bool))
 
     def test_isnull_datetime_dtype(self):
         class DatetimeFactor(Factor):
@@ -149,20 +146,18 @@ class FactorTestCase(BasePipelineTestCase):
         data = arange(25).reshape(5, 5).astype('datetime64[ns]')
         data[eye(5, dtype=bool)] = NaTns
 
-        graph = TermGraph(
+        self.check_terms(
             {
                 'isnull': factor.isnull(),
                 'notnull': factor.notnull(),
-            }
-        )
-
-        results = self.run_graph(
-            graph,
+            },
+            {
+                'isnull': eye(5, dtype=bool),
+                'notnull': ~eye(5, dtype=bool),
+            },
             initial_workspace={factor: data},
             mask=self.build_mask(ones((5, 5))),
         )
-        check_arrays(results['isnull'], eye(5, dtype=bool))
-        check_arrays(results['notnull'], ~eye(5, dtype=bool))
 
     @for_each_factor_dtype
     def test_rank_ascending(self, name, factor_dtype):
@@ -206,14 +201,12 @@ class FactorTestCase(BasePipelineTestCase):
         }
 
         def check(terms):
-            graph = TermGraph(terms)
-            results = self.run_graph(
-                graph,
+            self.check_terms(
+                terms,
+                expected={name: expected_ranks[name] for name in terms},
                 initial_workspace={f: data},
                 mask=self.build_mask(ones((5, 5))),
             )
-            for method in terms:
-                check_arrays(results[method], expected_ranks[method])
 
         check({meth: f.rank(method=meth) for meth in expected_ranks})
         check({
@@ -265,14 +258,12 @@ class FactorTestCase(BasePipelineTestCase):
         }
 
         def check(terms):
-            graph = TermGraph(terms)
-            results = self.run_graph(
-                graph,
+            self.check_terms(
+                terms,
+                expected={name: expected_ranks[name] for name in terms},
                 initial_workspace={f: data},
                 mask=self.build_mask(ones((5, 5))),
             )
-            for method in terms:
-                check_arrays(results[method], expected_ranks[method])
 
         check({
             meth: f.rank(method=meth, ascending=False)
@@ -294,14 +285,12 @@ class FactorTestCase(BasePipelineTestCase):
         mask_data = ~eye(5, dtype=bool)
         initial_workspace = {f: data, Mask(): mask_data}
 
-        graph = TermGraph(
-            {
-                "ascending_nomask": f.rank(ascending=True),
-                "ascending_mask": f.rank(ascending=True, mask=Mask()),
-                "descending_nomask": f.rank(ascending=False),
-                "descending_mask": f.rank(ascending=False, mask=Mask()),
-            }
-        )
+        terms = {
+            "ascending_nomask": f.rank(ascending=True),
+            "ascending_mask": f.rank(ascending=True, mask=Mask()),
+            "descending_nomask": f.rank(ascending=False),
+            "descending_mask": f.rank(ascending=False, mask=Mask()),
+        }
 
         expected = {
             "ascending_nomask": array([[1., 3., 4., 5., 2.],
@@ -328,13 +317,12 @@ class FactorTestCase(BasePipelineTestCase):
                                       [4., 3., 2., 1., nan]]),
         }
 
-        results = self.run_graph(
-            graph,
+        self.check_terms(
+            terms,
+            expected,
             initial_workspace,
             mask=self.build_mask(ones((5, 5))),
         )
-        for method in results:
-            check_arrays(expected[method], results[method])
 
     @for_each_factor_dtype
     def test_grouped_rank_ascending(self, name, factor_dtype=float64_dtype):
@@ -363,7 +351,7 @@ class FactorTestCase(BasePipelineTestCase):
             missing_value=None,
         )
 
-        expected_grouped_ranks = {
+        expected_ranks = {
             'ordinal': array(
                 [[1., 1., 3., 2., 2.],
                  [1., 2., 3., 1., 2.],
@@ -402,9 +390,9 @@ class FactorTestCase(BasePipelineTestCase):
         }
 
         def check(terms):
-            graph = TermGraph(terms)
-            results = self.run_graph(
-                graph,
+            self.check_terms(
+                terms,
+                expected={name: expected_ranks[name] for name in terms},
                 initial_workspace={
                     f: data,
                     c: classifier_data,
@@ -413,25 +401,22 @@ class FactorTestCase(BasePipelineTestCase):
                 mask=self.build_mask(ones((5, 5))),
             )
 
-            for method in terms:
-                check_arrays(results[method], expected_grouped_ranks[method])
-
         # Not specifying the value of ascending param should default to True
         check({
             meth: f.rank(method=meth, groupby=c)
-            for meth in expected_grouped_ranks
+            for meth in expected_ranks
         })
         check({
             meth: f.rank(method=meth, groupby=str_c)
-            for meth in expected_grouped_ranks
+            for meth in expected_ranks
         })
         check({
             meth: f.rank(method=meth, groupby=c, ascending=True)
-            for meth in expected_grouped_ranks
+            for meth in expected_ranks
         })
         check({
             meth: f.rank(method=meth, groupby=str_c, ascending=True)
-            for meth in expected_grouped_ranks
+            for meth in expected_ranks
         })
 
         # Not passing a method should default to ordinal
@@ -468,7 +453,7 @@ class FactorTestCase(BasePipelineTestCase):
             missing_value=None,
         )
 
-        expected_grouped_ranks = {
+        expected_ranks = {
             'ordinal': array(
                 [[2., 2., 1., 1., 3.],
                  [2., 1., 1., 2., 3.],
@@ -507,9 +492,9 @@ class FactorTestCase(BasePipelineTestCase):
         }
 
         def check(terms):
-            graph = TermGraph(terms)
-            results = self.run_graph(
-                graph,
+            self.check_terms(
+                terms,
+                expected={name: expected_ranks[name] for name in terms},
                 initial_workspace={
                     f: data,
                     c: classifier_data,
@@ -518,16 +503,13 @@ class FactorTestCase(BasePipelineTestCase):
                 mask=self.build_mask(ones((5, 5))),
             )
 
-            for method in terms:
-                check_arrays(results[method], expected_grouped_ranks[method])
-
         check({
             meth: f.rank(method=meth, groupby=c, ascending=False)
-            for meth in expected_grouped_ranks
+            for meth in expected_ranks
         })
         check({
             meth: f.rank(method=meth, groupby=str_c, ascending=False)
-            for meth in expected_grouped_ranks
+            for meth in expected_ranks
         })
 
         # Not passing a method should default to ordinal
@@ -707,9 +689,9 @@ class FactorTestCase(BasePipelineTestCase):
         expected['grouped_str'] = expected['grouped']
         expected['grouped_masked_str'] = expected['grouped_masked']
 
-        graph = TermGraph(terms)
-        results = self.run_graph(
-            graph,
+        self.check_terms(
+            terms,
+            expected,
             initial_workspace={
                 f: factor_data,
                 c: classifier_data,
@@ -717,20 +699,13 @@ class FactorTestCase(BasePipelineTestCase):
                 m: filter_data,
             },
             mask=self.build_mask(self.ones_mask(shape=factor_data.shape)),
+            # The hand-computed values aren't very precise (in particular,
+            # we truncate repeating decimals at 3 places) This is just
+            # asserting that the example isn't misleading by being totally
+            # wrong.
+            check=partial(check_allclose, atol=0.001),
         )
 
-        for key, (res, exp) in dzip_exact(results, expected).items():
-            check_allclose(
-                res,
-                exp,
-                # The hand-computed values aren't very precise (in particular,
-                # we truncate repeating decimals at 3 places) This is just
-                # asserting that the example isn't misleading by being totally
-                # wrong.
-                atol=0.001,
-                err_msg="Mismatch for %r" % key
-            )
-
     @parameter_space(
         seed_value=range(1, 2),
         normalizer_name_and_func=[

tests/pipeline/test_filter.py

@@ -12,7 +12,6 @@ from numpy import (
     array,
     eye,
     float64,
-    full_like,
     full,
     inf,
     isfinite,
@@ -27,11 +26,11 @@ from numpy import (
 from numpy.random import randn, seed as random_seed
 
 from zipline.errors import BadPercentileBounds
-from zipline.pipeline import Filter, Factor, TermGraph
+from zipline.pipeline import Filter, Factor
 from zipline.pipeline.classifiers import Classifier
 from zipline.pipeline.factors import CustomFactor
 from zipline.pipeline.filters import All, Any, AtLeastN
-from zipline.testing import check_arrays, parameter_space, permute_rows
+from zipline.testing import parameter_space, permute_rows
 from zipline.utils.numpy_utils import float64_dtype, int64_dtype
 from .base import BasePipelineTestCase, with_default_shape
 
@@ -127,7 +126,6 @@ class FilterTestCase(BasePipelineTestCase):
         nan_data[:, 0] = nan
 
         mask = Mask()
-        workspace = {self.f: data, mask: mask_data}
 
         methods = ['top', 'bottom']
         counts = 2, 3, 10
@@ -136,18 +134,6 @@ class FilterTestCase(BasePipelineTestCase):
         def termname(method, count, masked):
             return '_'.join([method, str(count), 'mask' if masked else ''])
 
-        # Add a term for each permutation of top/bottom, count, and
-        # mask/no_mask.
-        terms = {}
-        for method, count, masked in term_combos:
-            kwargs = {'N': count}
-            if masked:
-                kwargs['mask'] = mask
-            term = getattr(self.f, method)(**kwargs)
-            terms[termname(method, count, masked)] = term
-
-        results = self.run_graph(TermGraph(terms), initial_workspace=workspace)
-
         def expected_result(method, count, masked):
             # Ranking with a mask is equivalent to ranking with nans applied on
             # the masked values.
@@ -158,72 +144,55 @@ class FilterTestCase(BasePipelineTestCase):
             elif method == 'bottom':
                 return rowwise_rank(to_rank) < count
 
+        # Add a term for each permutation of top/bottom, count, and
+        # mask/no_mask.
+        terms = {}
+        expected = {}
         for method, count, masked in term_combos:
-            result = results[termname(method, count, masked)]
+            kwargs = {'N': count}
+            if masked:
+                kwargs['mask'] = mask
+            term = getattr(self.f, method)(**kwargs)
+            name = termname(method, count, masked)
+            terms[name] = term
+            expected[name] = expected_result(method, count, masked)
 
-            # Check that `min(c, num_assets)` assets passed each day.
-            passed_per_day = result.sum(axis=1)
-            check_arrays(
-                passed_per_day,
-                full_like(passed_per_day, min(count, data.shape[1])),
-            )
-
-            expected = expected_result(method, count, masked)
-            check_arrays(result, expected)
-
-    def test_bottom(self):
-        counts = 2, 3, 10
-        data = self.randn_data(seed=5)  # Arbitrary seed choice.
-        results = self.run_graph(
-            TermGraph(
-                {'bottom_' + str(c): self.f.bottom(c) for c in counts}
-            ),
-            initial_workspace={self.f: data},
+        self.check_terms(
+            terms,
+            expected,
+            initial_workspace={self.f: data, mask: mask_data},
+            mask=self.build_mask(self.ones_mask()),
         )
-        for c in counts:
-            result = results['bottom_' + str(c)]
-
-            # Check that `min(c, num_assets)` assets passed each day.
-            passed_per_day = result.sum(axis=1)
-            check_arrays(
-                passed_per_day,
-                full_like(passed_per_day, min(c, data.shape[1])),
-            )
-
-            # Check that the bottom `c` assets passed.
-            expected = rowwise_rank(data) < c
-            check_arrays(result, expected)
 
     def test_percentile_between(self):
 
         quintiles = range(5)
         filter_names = ['pct_' + str(q) for q in quintiles]
-        iter_quintiles = zip(filter_names, quintiles)
-
-        graph = TermGraph(
-            {
-                name: self.f.percentile_between(q * 20.0, (q + 1) * 20.0)
-                for name, q in zip(filter_names, quintiles)
-            }
-        )
+        iter_quintiles = list(zip(filter_names, quintiles))
+        terms = {
+            name: self.f.percentile_between(q * 20.0, (q + 1) * 20.0)
+            for name, q in iter_quintiles
+        }
 
         # Test with 5 columns and no NaNs.
         eye5 = eye(5, dtype=float64)
-        results = self.run_graph(
-            graph,
-            initial_workspace={self.f: eye5},
-            mask=self.build_mask(ones((5, 5))),
-        )
+        expected = {}
         for name, quintile in iter_quintiles:
-            result = results[name]
             if quintile < 4:
                 # There are four 0s and one 1 in each row, so the first 4
                 # quintiles should be all the locations with zeros in the input
                 # array.
-                check_arrays(result, ~eye5.astype(bool))
+                expected[name] = ~eye5.astype(bool)
             else:
                 # The top quintile should match the sole 1 in each row.
-                check_arrays(result, eye5.astype(bool))
+                expected[name] = eye5.astype(bool)
+
+        self.check_terms(
+            terms=terms,
+            expected=expected,
+            initial_workspace={self.f: eye5},
+            mask=self.build_mask(ones((5, 5))),
+        )
 
         # Test with 6 columns, no NaNs, and one masked entry per day.
         eye6 = eye(6, dtype=float64)
@@ -233,41 +202,44 @@ class FilterTestCase(BasePipelineTestCase):
                       [1, 1, 0, 1, 1, 1],
                       [1, 1, 1, 0, 1, 1],
                       [1, 1, 1, 1, 0, 1]], dtype=bool)
-
-        results = self.run_graph(
-            graph,
-            initial_workspace={self.f: eye6},
-            mask=self.build_mask(mask)
-        )
+        expected = {}
         for name, quintile in iter_quintiles:
-            result = results[name]
             if quintile < 4:
                 # Should keep all values that were 0 in the base data and were
                 # 1 in the mask.
-                check_arrays(result, mask & (~eye6.astype(bool))),
+                expected[name] = mask & ~eye6.astype(bool)
             else:
-                # Should keep all the 1s in the base data.
-                check_arrays(result, eye6.astype(bool))
+                # The top quintile should match the sole 1 in each row.
+                expected[name] = eye6.astype(bool)
+
+        self.check_terms(
+            terms=terms,
+            expected=expected,
+            initial_workspace={self.f: eye6},
+            mask=self.build_mask(mask),
+        )
 
         # Test with 6 columns, no mask, and one NaN per day.  Should have the
         # same outcome as if we had masked the NaNs.
         # In particular, the NaNs should never pass any filters.
         eye6_withnans = eye6.copy()
         putmask(eye6_withnans, ~mask, nan)
-        results = self.run_graph(
-            graph,
-            initial_workspace={self.f: eye6},
-            mask=self.build_mask(mask)
-        )
+        expected = {}
         for name, quintile in iter_quintiles:
-            result = results[name]
             if quintile < 4:
                 # Should keep all values that were 0 in the base data and were
                 # 1 in the mask.
-                check_arrays(result, mask & (~eye6.astype(bool))),
+                expected[name] = mask & (~eye6.astype(bool))
             else:
                 # Should keep all the 1s in the base data.
-                check_arrays(result, eye6.astype(bool))
+                expected[name] = eye6.astype(bool)
+
+        self.check_terms(
+            terms,
+            expected,
+            initial_workspace={self.f: eye6},
+            mask=self.build_mask(mask),
+        )
 
     def test_percentile_nasty_partitions(self):
         # Test percentile with nasty partitions: divide up 5 assets into
@@ -281,27 +253,26 @@ class FilterTestCase(BasePipelineTestCase):
         quartiles = range(4)
         filter_names = ['pct_' + str(q) for q in quartiles]
 
-        graph = TermGraph(
-            {
-                name: self.f.percentile_between(q * 25.0, (q + 1) * 25.0)
-                for name, q in zip(filter_names, quartiles)
-            }
-        )
-        results = self.run_graph(
-            graph,
-            initial_workspace={self.f: data},
-            mask=self.build_mask(ones((5, 5))),
-        )
+        terms = {
+            name: self.f.percentile_between(q * 25.0, (q + 1) * 25.0)
+            for name, q in zip(filter_names, quartiles)
+        }
 
+        expected = {}
         for name, quartile in zip(filter_names, quartiles):
-            result = results[name]
             lower = quartile * 25.0
             upper = (quartile + 1) * 25.0
-            expected = and_(
+            expected[name] = and_(
                 nanpercentile(data, lower, axis=1, keepdims=True) <= data,
                 data <= nanpercentile(data, upper, axis=1, keepdims=True),
             )
-            check_arrays(result, expected)
+
+        self.check_terms(
+            terms,
+            expected,
+            initial_workspace={self.f: data},
+            mask=self.build_mask(ones((5, 5))),
+        )
 
     def test_percentile_after_mask(self):
         f_input = eye(5)
@@ -312,77 +283,79 @@ class FilterTestCase(BasePipelineTestCase):
         without_mask = self.g.percentile_between(80, 100)
         with_mask = self.g.percentile_between(80, 100, mask=custom_mask)
 
-        graph = TermGraph(
-            {
-                'custom_mask': custom_mask,
-                'without': without_mask,
-                'with': with_mask,
-            }
-        )
+        terms = {
+            'mask': custom_mask,
+            'without_mask': without_mask,
+            'with_mask': with_mask,
+        }
+        expected = {
+            # Mask that accepts everything except the diagonal.
+            'mask': ~eye(5, dtype=bool),
+            # Second should pass the largest value each day.  Each row is
+            # strictly increasing, so we always select the last value.
+            'without_mask': array(
+                [[0, 0, 0, 0, 1],
+                 [0, 0, 0, 0, 1],
+                 [0, 0, 0, 0, 1],
+                 [0, 0, 0, 0, 1],
+                 [0, 0, 0, 0, 1]],
+                dtype=bool,
+            ),
+            # With a mask, we should remove the diagonal as an option before
+            # computing percentiles.  On the last day, we should get the
+            # second-largest value, rather than the largest.
+            'with_mask': array(
+                [[0, 0, 0, 0, 1],
+                 [0, 0, 0, 0, 1],
+                 [0, 0, 0, 0, 1],
+                 [0, 0, 0, 0, 1],
+                 [0, 0, 0, 1, 0]],  # Different from with!
+                dtype=bool,
+            ),
+        }
 
-        results = self.run_graph(
-            graph,
+        self.check_terms(
+            terms,
+            expected,
             initial_workspace={self.f: f_input, self.g: g_input},
             mask=initial_mask,
         )
 
-        # First should pass everything but the diagonal.
-        check_arrays(results['custom_mask'], ~eye(5, dtype=bool))
-
-        # Second should pass the largest value each day.  Each row is strictly
-        # increasing, so we always select the last value.
-        expected_without = array(
-            [[0, 0, 0, 0, 1],
-             [0, 0, 0, 0, 1],
-             [0, 0, 0, 0, 1],
-             [0, 0, 0, 0, 1],
-             [0, 0, 0, 0, 1]],
-            dtype=bool,
-        )
-        check_arrays(results['without'], expected_without)
-
-        # When sequencing, we should remove the diagonal as an option before
-        # computing percentiles.  On the last day, we should get the
-        # second-largest value, rather than the largest.
-        expected_with = array(
-            [[0, 0, 0, 0, 1],
-             [0, 0, 0, 0, 1],
-             [0, 0, 0, 0, 1],
-             [0, 0, 0, 0, 1],
-             [0, 0, 0, 1, 0]],  # Different from previous!
-            dtype=bool,
-        )
-        check_arrays(results['with'], expected_with)
-
     def test_isnan(self):
         data = self.randn_data(seed=10)
         diag = eye(*data.shape, dtype=bool)
         data[diag] = nan
 
-        results = self.run_graph(
-            TermGraph({
+        self.check_terms(
+            terms={
                 'isnan': self.f.isnan(),
                 'isnull': self.f.isnull(),
-            }),
+            },
+            expected={
+                'isnan': diag,
+                'isnull': diag,
+            },
             initial_workspace={self.f: data},
+            mask=self.build_mask(self.ones_mask()),
         )
-        check_arrays(results['isnan'], diag)
-        check_arrays(results['isnull'], diag)
 
     def test_notnan(self):
         data = self.randn_data(seed=10)
         diag = eye(*data.shape, dtype=bool)
         data[diag] = nan
 
-        results = self.run_graph(
-            TermGraph({
+        self.check_terms(
+            terms={
                 'notnan': self.f.notnan(),
                 'notnull': self.f.notnull(),
-            }),
+            },
+            expected={
+                'notnan': ~diag,
+                'notnull': ~diag,
+            },
             initial_workspace={self.f: data},
+            mask=self.build_mask(self.ones_mask()),
         )
-        check_arrays(results['notnan'], ~diag)
-        check_arrays(results['notnull'], ~diag)
 
     def test_isfinite(self):
         data = self.randn_data(seed=10)
@@ -390,11 +363,12 @@ class FilterTestCase(BasePipelineTestCase):
         data[:, 2] = inf
         data[:, 4] = -inf
 
-        results = self.run_graph(
-            TermGraph({'isfinite': self.f.isfinite()}),
+        self.check_terms(
+            terms={'isfinite': self.f.isfinite()},
+            expected={'isfinite': isfinite(data)},
             initial_workspace={self.f: data},
+            mask=self.build_mask(self.ones_mask()),
         )
-        check_arrays(results['isfinite'], isfinite(data))
 
     def test_all(self):
 
@@ -427,18 +401,19 @@ class FilterTestCase(BasePipelineTestCase):
             inputs = ()
             window_length = 0
 
-        results = self.run_graph(
-            TermGraph({
+        self.check_terms(
+            terms={
                 '3': All(inputs=[Input()], window_length=3),
                 '4': All(inputs=[Input()], window_length=4),
-            }),
+            },
+            expected={
+                '3': expected_3,
+                '4': expected_4,
+            },
             initial_workspace={Input(): data},
             mask=self.build_mask(ones(shape=data.shape)),
         )
 
-        check_arrays(results['3'], expected_3)
-        check_arrays(results['4'], expected_4)
-
     def test_any(self):
 
         # FUN FACT: The inputs and outputs here are exactly the negation of
@@ -486,18 +461,19 @@ class FilterTestCase(BasePipelineTestCase):
             inputs = ()
             window_length = 0
 
-        results = self.run_graph(
-            TermGraph({
+        self.check_terms(
+            terms={
                 '3': Any(inputs=[Input()], window_length=3),
                 '4': Any(inputs=[Input()], window_length=4),
-            }),
+            },
+            expected={
+                '3': expected_3,
+                '4': expected_4,
+            },
             initial_workspace={Input(): data},
             mask=self.build_mask(ones(shape=data.shape)),
         )
 
-        check_arrays(results['3'], expected_3)
-        check_arrays(results['4'], expected_4)
-
     def test_at_least_N(self):
 
         # With a window_length of K, AtLeastN should return 1
@@ -553,26 +529,27 @@ class FilterTestCase(BasePipelineTestCase):
                              window_length=4,
                              N=4)
 
-        results = self.run_graph(
-            TermGraph({
+        self.check_terms(
+            terms={
                 'AllButOne': all_but_one,
                 'AllButTwo': all_but_two,
                 'AnyEquiv': any_equiv,
                 'AllEquiv': all_equiv,
                 'Any': Any(inputs=[Input()], window_length=4),
                 'All': All(inputs=[Input()], window_length=4)
-            }),
+            },
+            expected={
+                'Any': expected_1,
+                'AnyEquiv': expected_1,
+                'AllButTwo': expected_2,
+                'AllButOne': expected_3,
+                'All': expected_4,
+                'AllEquiv': expected_4,
+            },
             initial_workspace={Input(): data},
             mask=self.build_mask(ones(shape=data.shape)),
         )
 
-        check_arrays(results['Any'], expected_1)
-        check_arrays(results['AnyEquiv'], expected_1)
-        check_arrays(results['AllButTwo'], expected_2)
-        check_arrays(results['AllButOne'], expected_3)
-        check_arrays(results['All'], expected_4)
-        check_arrays(results['AllEquiv'], expected_4)
-
     @parameter_space(factor_len=[2, 3, 4])
     def test_window_safe(self, factor_len):
         # all true data set of (days, securities)
@@ -591,19 +568,19 @@ class FilterTestCase(BasePipelineTestCase):
                 # sum for each column
                 out[:] = np_sum(filter_, axis=0)
 
-        results = self.run_graph(
-            TermGraph({'windowsafe': TestFactor()}),
-            initial_workspace={InputFilter(): data},
-        )
-
-        # number of days in default_shape
         n = self.default_shape[0]
-
-        # shape of output array
         output_shape = ((n - factor_len + 1), self.default_shape[1])
-        check_arrays(
-            results['windowsafe'],
-            full(output_shape, factor_len, dtype=float64)
+        full(output_shape, factor_len, dtype=float64)
+
+        self.check_terms(
+            terms={
+                'windowsafe': TestFactor(),
+            },
+            expected={
+                'windowsafe': full(output_shape, factor_len, dtype=float64),
+            },
+            initial_workspace={InputFilter(): data},
+            mask=self.build_mask(self.ones_mask()),
         )
 
     @parameter_space(

tests/pipeline/test_pipeline.py

@@ -1,10 +1,14 @@
 """
 Tests for zipline.pipeline.Pipeline
 """
+import inspect
 from unittest import TestCase
 
+from mock import patch
+
 from zipline.pipeline import Factor, Filter, Pipeline
 from zipline.pipeline.data import USEquityPricing
+from zipline.pipeline.graph import display_graph
 from zipline.utils.numpy_utils import float64_dtype
 
 
@@ -137,3 +141,57 @@ class PipelineTestCase(TestCase):
             "expected a value of type bool or int for argument 'overwrite'",
             message,
         )
+
+    def test_show_graph(self):
+        f = SomeFactor()
+        p = Pipeline(columns={'f': SomeFactor()})
+
+        # The real display_graph call shells out to GraphViz, which isn't a
+        # requirement, so patch it out for testing.
+
+        def mock_display_graph(g, format='svg', include_asset_exists=False):
+            return (g, format, include_asset_exists)
+
+        self.assertEqual(
+            inspect.getargspec(display_graph),
+            inspect.getargspec(mock_display_graph),
+            msg="Mock signature doesn't match signature for display_graph."
+        )
+
+        patch_display_graph = patch(
+            'zipline.pipeline.graph.display_graph',
+            mock_display_graph,
+        )
+
+        with patch_display_graph:
+            graph, format, include_asset_exists = p.show_graph()
+            self.assertIs(graph.outputs['f'], f)
+            # '' is a sentinel used for screen if it's not supplied.
+            self.assertEqual(sorted(graph.outputs.keys()), ['', 'f'])
+            self.assertEqual(format, 'svg')
+            self.assertEqual(include_asset_exists, False)
+
+        with patch_display_graph:
+            graph, format, include_asset_exists = p.show_graph(format='png')
+            self.assertIs(graph.outputs['f'], f)
+            # '' is a sentinel used for screen if it's not supplied.
+            self.assertEqual(sorted(graph.outputs.keys()), ['', 'f'])
+            self.assertEqual(format, 'png')
+            self.assertEqual(include_asset_exists, False)
+
+        with patch_display_graph:
+            graph, format, include_asset_exists = p.show_graph(format='jpeg')
+            self.assertIs(graph.outputs['f'], f)
+            # '' is a sentinel used for screen if it's not supplied.
+            self.assertEqual(sorted(graph.outputs.keys()), ['', 'f'])
+            self.assertEqual(format, 'jpeg')
+            self.assertEqual(include_asset_exists, False)
+
+        expected = (
+            r".*\.show_graph\(\) expected a value in "
+            r"\('svg', 'png', 'jpeg'\) for argument 'format', "
+            r"but got 'fizzbuzz' instead."
+        )
+
+        with self.assertRaisesRegexp(ValueError, expected):
+            p.show_graph(format='fizzbuzz')

tests/pipeline/test_technical.py

@@ -7,10 +7,7 @@ import pandas as pd
 import talib
 
 from zipline.lib.adjusted_array import AdjustedArray
-from zipline.pipeline import TermGraph
 from zipline.pipeline.data import USEquityPricing
-from zipline.pipeline.engine import SimplePipelineEngine
-from zipline.pipeline.term import AssetExists
 from zipline.pipeline.factors import (
     BollingerBands,
     Aroon,
@@ -20,61 +17,22 @@ from zipline.pipeline.factors import (
     RateOfChangePercentage,
     TrueRange,
 )
-from zipline.testing import ExplodingObject, parameter_space
-from zipline.testing.fixtures import WithAssetFinder, ZiplineTestCase
+from zipline.testing import parameter_space
+from zipline.testing.fixtures import ZiplineTestCase
 from zipline.testing.predicates import assert_equal
 
-
-class WithTechnicalFactor(WithAssetFinder):
-    """ZiplineTestCase fixture for testing technical factors.
-    """
-    ASSET_FINDER_EQUITY_SIDS = tuple(range(5))
-    START_DATE = pd.Timestamp('2014-01-01', tz='utc')
-
-    @classmethod
-    def init_class_fixtures(cls):
-        super(WithTechnicalFactor, cls).init_class_fixtures()
-        cls.ndays = ndays = 24
-        cls.nassets = nassets = len(cls.ASSET_FINDER_EQUITY_SIDS)
-        cls.dates = dates = pd.date_range(cls.START_DATE, periods=ndays)
-        cls.assets = pd.Index(cls.asset_finder.sids)
-        cls.engine = SimplePipelineEngine(
-            lambda column: ExplodingObject(),
-            dates,
-            cls.asset_finder,
-        )
-        cls.asset_exists = exists = np.full((ndays, nassets), True, dtype=bool)
-        cls.asset_exists_masked = masked = exists.copy()
-        masked[:, -1] = False
-
-    def run_graph(self, graph, initial_workspace, mask_sid):
-        initial_workspace.setdefault(
-            AssetExists(),
-            self.asset_exists_masked if mask_sid else self.asset_exists,
-        )
-        return self.engine.compute_chunk(
-            graph,
-            self.dates,
-            self.assets,
-            initial_workspace,
-        )
+from .base import BasePipelineTestCase
 
 
-class BollingerBandsTestCase(WithTechnicalFactor, ZiplineTestCase):
-    @classmethod
-    def init_class_fixtures(cls):
-        super(BollingerBandsTestCase, cls).init_class_fixtures()
-        cls._closes = closes = (
-            np.arange(cls.ndays, dtype=float)[:, np.newaxis] +
-            np.arange(cls.nassets, dtype=float) * 100
-        )
-        cls._closes_masked = masked = closes.copy()
-        masked[:, -1] = np.nan
+class BollingerBandsTestCase(BasePipelineTestCase):
 
-    def closes(self, masked):
-        return self._closes_masked if masked else self._closes
+    def closes(self, mask_last_sid):
+        data = self.arange_data(dtype=np.float64)
+        if mask_last_sid:
+            data[:, -1] = np.nan
+        return data
 
-    def expected(self, window_length, k, closes):
+    def expected_bbands(self, window_length, k, closes):
         """Compute the expected data (without adjustments) for the given
         window, k, and closes array.
 
@@ -83,11 +41,14 @@ class BollingerBandsTestCase(WithTechnicalFactor, ZiplineTestCase):
         lower_cols = []
         middle_cols = []
         upper_cols = []
-        for n in range(self.nassets):
+
+        ndates, nassets = closes.shape
+
+        for n in range(nassets):
             close_col = closes[:, n]
             if np.isnan(close_col).all():
                 # ta-lib doesn't deal well with all nans.
-                upper, middle, lower = [np.full(self.ndays, np.nan)] * 3
+                upper, middle, lower = [np.full(ndates, np.nan)] * 3
             else:
                 upper, middle, lower = talib.BBANDS(
                     close_col,
@@ -112,38 +73,38 @@ class BollingerBandsTestCase(WithTechnicalFactor, ZiplineTestCase):
     @parameter_space(
         window_length={5, 10, 20},
         k={1.5, 2, 2.5},
-        mask_sid={True, False},
+        mask_last_sid={True, False},
+        __fail_fast=True,
     )
-    def test_bollinger_bands(self, window_length, k, mask_sid):
-        closes = self.closes(mask_sid)
-        result = self.run_graph(
-            TermGraph({
-                'f': BollingerBands(
-                    window_length=window_length,
-                    k=k,
-                ),
-            }),
+    def test_bollinger_bands(self, window_length, k, mask_last_sid):
+        closes = self.closes(mask_last_sid=mask_last_sid)
+        mask = ~np.isnan(closes)
+        bbands = BollingerBands(window_length=window_length, k=k)
+
+        expected = self.expected_bbands(window_length, k, closes)
+
+        self.check_terms(
+            terms={
+                'upper': bbands.upper,
+                'middle': bbands.middle,
+                'lower': bbands.lower,
+            },
+            expected={
+                'upper': expected[0],
+                'middle': expected[1],
+                'lower': expected[2],
+            },
             initial_workspace={
                 USEquityPricing.close: AdjustedArray(
-                    closes,
-                    np.full_like(closes, True, dtype=bool),
-                    {},
-                    np.nan,
+                    data=closes,
+                    mask=mask,
+                    adjustments={},
+                    missing_value=np.nan,
                 ),
             },
-            mask_sid=mask_sid,
-        )['f']
-
-        expected_upper, expected_middle, expected_lower = self.expected(
-            window_length,
-            k,
-            closes,
+            mask=self.build_mask(mask),
         )
 
-        assert_equal(result.upper, expected_upper)
-        assert_equal(result.middle, expected_middle)
-        assert_equal(result.lower, expected_lower)
-
     def test_bollinger_bands_output_ordering(self):
         bbands = BollingerBands(window_length=5, k=2)
         lower, middle, upper = bbands
@@ -185,7 +146,7 @@ class AroonTestCase(ZiplineTestCase):
         assert_equal(out, expected_out)
 
 
-class TestFastStochasticOscillator(WithTechnicalFactor, ZiplineTestCase):
+class TestFastStochasticOscillator(ZiplineTestCase):
     """
     Test the Fast Stochastic Oscillator
     """
@@ -427,7 +388,7 @@ class TestLinearWeightedMovingAverage(ZiplineTestCase):
         assert_equal(out, np.array([30.,  31.,  32.,  33.,  34.]))
 
 
-class TestTrueRange(WithTechnicalFactor, ZiplineTestCase):
+class TestTrueRange(ZiplineTestCase):
 
     def test_tr_basic(self):
         tr = TrueRange()

tests/pipeline/test_term.py

@@ -6,6 +6,7 @@ from itertools import product
 from unittest import TestCase
 
 from toolz import assoc
+import pandas as pd
 
 from zipline.assets import Asset
 from zipline.errors import (
@@ -24,7 +25,7 @@ from zipline.pipeline import (
     CustomFactor,
     Factor,
     Filter,
-    TermGraph,
+    ExecutionPlan,
 )
 from zipline.pipeline.data import Column, DataSet
 from zipline.pipeline.data.testing import TestingDataSet
@@ -33,6 +34,7 @@ from zipline.pipeline.factors import RecarrayField
 from zipline.pipeline.sentinels import NotSpecified
 from zipline.pipeline.term import AssetExists, Slice
 from zipline.testing import parameter_space
+from zipline.testing.fixtures import WithTradingSessions, ZiplineTestCase
 from zipline.testing.predicates import (
     assert_equal,
     assert_raises,
@@ -152,7 +154,14 @@ def to_dict(l):
     return dict(zip(map(str, range(len(l))), l))
 
 
-class DependencyResolutionTestCase(TestCase):
+class DependencyResolutionTestCase(WithTradingSessions, ZiplineTestCase):
+
+    TRADING_CALENDAR_STRS = ('NYSE',)
+    START_DATE = pd.Timestamp('2014-01-02', tz='UTC')
+    END_DATE = pd.Timestamp('2014-12-31', tz='UTC')
+
+    execution_plan_start = pd.Timestamp('2014-06-01', tz='UTC')
+    execution_plan_end = pd.Timestamp('2014-06-30', tz='UTC')
 
     def check_dependency_order(self, ordered_terms):
         seen = set()
@@ -163,6 +172,14 @@ class DependencyResolutionTestCase(TestCase):
 
             seen.add(term)
 
+    def make_execution_plan(self, terms):
+        return ExecutionPlan(
+            terms,
+            self.nyse_sessions,
+            self.execution_plan_start,
+            self.execution_plan_end,
+        )
+
     def test_single_factor(self):
         """
         Test dependency resolution for a single factor.
@@ -182,7 +199,7 @@ class DependencyResolutionTestCase(TestCase):
             self.assertEqual(graph.node[SomeDataSet.bar]['extra_rows'], 4)
 
         for foobar in gen_equivalent_factors():
-            check_output(TermGraph(to_dict([foobar])))
+            check_output(self.make_execution_plan(to_dict([foobar])))
 
     def test_single_factor_instance_args(self):
         """
@@ -190,7 +207,9 @@ class DependencyResolutionTestCase(TestCase):
         the constructor.
         """
         bar, buzz = SomeDataSet.bar, SomeDataSet.buzz
-        graph = TermGraph(to_dict([SomeFactor([bar, buzz], window_length=5)]))
+
+        factor = SomeFactor([bar, buzz], window_length=5)
+        graph = self.make_execution_plan(to_dict([factor]))
 
         resolution_order = list(graph.ordered())
 
@@ -214,7 +233,7 @@ class DependencyResolutionTestCase(TestCase):
         f1 = SomeFactor([SomeDataSet.foo, SomeDataSet.bar])
         f2 = SomeOtherFactor([SomeDataSet.bar, SomeDataSet.buzz])
 
-        graph = TermGraph(to_dict([f1, f2]))
+        graph = self.make_execution_plan(to_dict([f1, f2]))
         resolution_order = list(graph.ordered())
 
         # bar should only appear once.

tests/utils/test_pandas_utils.py

@@ -0,0 +1,79 @@
+"""
+Tests for zipline/utils/pandas_utils.py
+"""
+import pandas as pd
+
+from zipline.testing import parameter_space, ZiplineTestCase
+from zipline.utils.pandas_utils import nearest_unequal_elements
+
+
+class TestNearestUnequalElements(ZiplineTestCase):
+
+    @parameter_space(tz=['UTC', 'US/Eastern'], __fail_fast=True)
+    def test_nearest_unequal_elements(self, tz):
+
+        dts = pd.to_datetime(
+            ['2014-01-01', '2014-01-05', '2014-01-06', '2014-01-09'],
+        ).tz_localize(tz)
+
+        t = lambda s: None if s is None else pd.Timestamp(s, tz=tz)
+
+        for dt, before, after in (('2013-12-30', None, '2014-01-01'),
+                                  ('2013-12-31', None, '2014-01-01'),
+                                  ('2014-01-01', None, '2014-01-05'),
+                                  ('2014-01-02', '2014-01-01', '2014-01-05'),
+                                  ('2014-01-03', '2014-01-01', '2014-01-05'),
+                                  ('2014-01-04', '2014-01-01', '2014-01-05'),
+                                  ('2014-01-05', '2014-01-01', '2014-01-06'),
+                                  ('2014-01-06', '2014-01-05', '2014-01-09'),
+                                  ('2014-01-07', '2014-01-06', '2014-01-09'),
+                                  ('2014-01-08', '2014-01-06', '2014-01-09'),
+                                  ('2014-01-09', '2014-01-06', None),
+                                  ('2014-01-10', '2014-01-09', None),
+                                  ('2014-01-11', '2014-01-09', None)):
+            computed = nearest_unequal_elements(dts, t(dt))
+            expected = (t(before), t(after))
+            self.assertEqual(computed, expected)
+
+    @parameter_space(tz=['UTC', 'US/Eastern'], __fail_fast=True)
+    def test_nearest_unequal_elements_short_dts(self, tz):
+
+        # Length 1.
+        dts = pd.to_datetime(['2014-01-01']).tz_localize(tz)
+        t = lambda s: None if s is None else pd.Timestamp(s, tz=tz)
+
+        for dt, before, after in (('2013-12-31', None, '2014-01-01'),
+                                  ('2014-01-01', None, None),
+                                  ('2014-01-02', '2014-01-01', None)):
+            computed = nearest_unequal_elements(dts, t(dt))
+            expected = (t(before), t(after))
+            self.assertEqual(computed, expected)
+
+        # Length 0
+        dts = pd.to_datetime([]).tz_localize(tz)
+        for dt, before, after in (('2013-12-31', None, None),
+                                  ('2014-01-01', None, None),
+                                  ('2014-01-02', None, None)):
+            computed = nearest_unequal_elements(dts, t(dt))
+            expected = (t(before), t(after))
+            self.assertEqual(computed, expected)
+
+    def test_nearest_unequal_bad_input(self):
+        with self.assertRaises(ValueError) as e:
+            nearest_unequal_elements(
+                pd.to_datetime(['2014', '2014']),
+                pd.Timestamp('2014'),
+            )
+
+        self.assertEqual(str(e.exception), 'dts must be unique')
+
+        with self.assertRaises(ValueError) as e:
+            nearest_unequal_elements(
+                pd.to_datetime(['2014', '2013']),
+                pd.Timestamp('2014'),
+            )
+
+        self.assertEqual(
+            str(e.exception),
+            'dts must be sorted in increasing order',
+        )

tests/utils/test_preprocess.py

@@ -262,7 +262,11 @@ class PreprocessTestCase(TestCase):
         expected_message = (
             "{qualname}() expected a value in {set_!r}"
             " for argument 'a', but got 'c' instead."
-        ).format(set_=set_, qualname=qualname(f))
+        ).format(
+            # We special-case set to show a tuple instead of the set repr.
+            set_=tuple(sorted(set_)),
+            qualname=qualname(f),
+        )
         self.assertEqual(e.exception.args[0], expected_message)
 
     def test_expect_dtypes(self):