ENH: Add single-column input/output capabilities to pipeline terms

tests/pipeline/test_engine.py

@@ -72,9 +72,14 @@ from zipline.pipeline.loaders.synthetic import (
     make_bar_data,
     expected_bar_values_2d,
 )
-from zipline.pipeline.term import NotSpecified
+from zipline.pipeline.sentinels import NotSpecified
 from zipline.testing import (
+    AssetID,
+    AssetIDPlusDay,
     check_arrays,
+    make_alternating_boolean_array,
+    make_cascading_boolean_array,
+    OpenPrice,
     parameter_space,
     product_upper_triangle,
 )
@@ -95,38 +100,6 @@ class RollingSumDifference(CustomFactor):
         out[:] = (open - close).sum(axis=0)
 
 
-class AssetID(CustomFactor):
-    """
-    CustomFactor that returns the AssetID of each asset.
-
-    Useful for providing a Factor that produces a different value for each
-    asset.
-    """
-    window_length = 1
-    # HACK: We currently decide whether to load or compute a Term based on the
-    # length of its inputs.  This means we have to provide a dummy input.
-    inputs = [USEquityPricing.close]
-
-    def compute(self, today, assets, out, close):
-        out[:] = assets
-
-
-class AssetIDPlusDay(CustomFactor):
-    window_length = 1
-    inputs = [USEquityPricing.close]
-
-    def compute(self, today, assets, out, close):
-        out[:] = assets + today.day
-
-
-class OpenPrice(CustomFactor):
-    window_length = 1
-    inputs = [USEquityPricing.open]
-
-    def compute(self, today, assets, out, open):
-        out[:] = open
-
-
 class MultipleOutputs(CustomFactor):
     window_length = 1
     inputs = [USEquityPricing.open, USEquityPricing.close]
@@ -421,6 +394,8 @@ class ConstantInputTestCase(WithTradingEnvironment, ZiplineTestCase):
         assets = self.assets
         asset_ids = self.asset_ids
         constants = self.constants
+        num_dates = len(dates)
+        num_assets = len(assets)
         open = USEquityPricing.open
         close = USEquityPricing.close
         engine = SimplePipelineEngine(
@@ -435,19 +410,13 @@ class ConstantInputTestCase(WithTradingEnvironment, ZiplineTestCase):
             return DataFrame(expected_values, index=dates, columns=assets)
 
         cascading_mask = AssetIDPlusDay() < (asset_ids[-1] + dates[0].day)
-        expected_cascading_mask_result = array(
-            [[True,  True,  True, False],
-             [True,  True, False, False],
-             [True, False, False, False]],
-            dtype=bool,
+        expected_cascading_mask_result = make_cascading_boolean_array(
+            shape=(num_dates, num_assets),
         )
 
         alternating_mask = (AssetIDPlusDay() % 2).eq(0)
-        expected_alternating_mask_result = array(
-            [[False,  True, False,  True],
-             [True,  False,  True, False],
-             [False,  True, False,  True]],
-            dtype=bool,
+        expected_alternating_mask_result = make_alternating_boolean_array(
+            shape=(num_dates, num_assets), first_value=False,
         )
 
         masks = cascading_mask, alternating_mask
@@ -592,6 +561,8 @@ class ConstantInputTestCase(WithTradingEnvironment, ZiplineTestCase):
         assets = self.assets
         asset_ids = self.asset_ids
         constants = self.constants
+        num_dates = len(dates)
+        num_assets = len(assets)
         open = USEquityPricing.open
         close = USEquityPricing.close
         engine = SimplePipelineEngine(
@@ -603,32 +574,17 @@ class ConstantInputTestCase(WithTradingEnvironment, ZiplineTestCase):
             return DataFrame(expected_values, index=dates, columns=assets)
 
         cascading_mask = AssetIDPlusDay() < (asset_ids[-1] + dates[0].day)
-        expected_cascading_mask_result = array(
-            [[True,   True,  True, False],
-             [True,   True, False, False],
-             [True,  False, False, False],
-             [False, False, False, False],
-             [False, False, False, False]],
-            dtype=bool,
+        expected_cascading_mask_result = make_cascading_boolean_array(
+            shape=(num_dates, num_assets),
         )
 
         alternating_mask = (AssetIDPlusDay() % 2).eq(0)
-        expected_alternating_mask_result = array(
-            [[False,  True, False,  True],
-             [True,  False,  True, False],
-             [False,  True, False,  True],
-             [True,  False,  True, False],
-             [False,  True, False,  True]],
-            dtype=bool,
+        expected_alternating_mask_result = make_alternating_boolean_array(
+            shape=(num_dates, num_assets), first_value=False,
         )
 
-        expected_no_mask_result = array(
-            [[True, True, True, True],
-             [True, True, True, True],
-             [True, True, True, True],
-             [True, True, True, True],
-             [True, True, True, True]],
-            dtype=bool,
+        expected_no_mask_result = full(
+            shape=(num_dates, num_assets), fill_value=True, dtype=bool,
         )
 
         masks = cascading_mask, alternating_mask, NotSpecified
@@ -1258,19 +1214,39 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
         `RollingSpearmanOfReturns`.
         """
         my_asset_column = 0
-        start_date_index = 6
-        end_date_index = 10
+        start_date_index = 14
+        end_date_index = 18
 
-        assets = self.asset_finder.retrieve_all(self.sids)
+        sids = self.sids
+        dates = self.dates
+        assets = self.asset_finder.retrieve_all(sids)
         my_asset = assets[my_asset_column]
-        my_asset_filter = (AssetID() != (my_asset_column + 1))
         num_days = end_date_index - start_date_index + 1
+        num_assets = len(assets)
 
-        # Our correlation factors require that their target asset is not
-        # filtered out, so make sure that masking out our target asset does not
-        # take effect. That is, a filter which filters out only our target
-        # asset should produce the same result as if no mask was passed at all.
-        for mask in (NotSpecified, my_asset_filter):
+        cascading_mask = \
+            AssetIDPlusDay() < (sids[-1] + dates[start_date_index].day)
+        expected_cascading_mask_result = make_cascading_boolean_array(
+            shape=(num_days, num_assets),
+        )
+
+        alternating_mask = (AssetIDPlusDay() % 2).eq(0)
+        expected_alternating_mask_result = make_alternating_boolean_array(
+            shape=(num_days, num_assets),
+        )
+
+        expected_no_mask_result = full(
+            shape=(num_days, num_assets), fill_value=True, dtype=bool,
+        )
+
+        masks = cascading_mask, alternating_mask, NotSpecified
+        expected_mask_results = (
+            expected_cascading_mask_result,
+            expected_alternating_mask_result,
+            expected_no_mask_result,
+        )
+
+        for mask, expected_mask in zip(masks, expected_mask_results):
             pearson_factor = RollingPearsonOfReturns(
                 target=my_asset,
                 returns_length=returns_length,
@@ -1284,18 +1260,23 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
                 mask=mask,
             )
 
+            pipeline = Pipeline(
+                columns={
+                    'pearson_factor': pearson_factor,
+                    'spearman_factor': spearman_factor,
+                },
+            )
+            if mask is not NotSpecified:
+                pipeline.add(mask, 'mask')
+
             results = self.engine.run_pipeline(
-                Pipeline(
-                    columns={
-                        'pearson_factor': pearson_factor,
-                        'spearman_factor': spearman_factor,
-                    },
-                ),
-                self.dates[start_date_index],
-                self.dates[end_date_index],
+                pipeline, dates[start_date_index], dates[end_date_index],
             )
             pearson_results = results['pearson_factor'].unstack()
             spearman_results = results['spearman_factor'].unstack()
+            if mask is not NotSpecified:
+                mask_results = results['mask'].unstack()
+                check_arrays(mask_results.values, expected_mask)
 
             # Run a separate pipeline that calculates returns starting
             # (correlation_length - 1) days prior to our start date. This is
@@ -1304,8 +1285,8 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
             returns = Returns(window_length=returns_length)
             results = self.engine.run_pipeline(
                 Pipeline(columns={'returns': returns}),
-                self.dates[start_date_index - (correlation_length - 1)],
-                self.dates[end_date_index],
+                dates[start_date_index - (correlation_length - 1)],
+                dates[end_date_index],
             )
             returns_results = results['returns'].unstack()
 
@@ -1328,22 +1309,19 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
                         my_asset_returns, other_asset_returns,
                     )[0]
 
-            assert_frame_equal(
-                pearson_results,
-                DataFrame(
-                    expected_pearson_results,
-                    index=self.dates[start_date_index:end_date_index + 1],
-                    columns=assets,
-                ),
+            expected_pearson_results = DataFrame(
+                data=where(expected_mask, expected_pearson_results, nan),
+                index=dates[start_date_index:end_date_index + 1],
+                columns=assets,
             )
-            assert_frame_equal(
-                spearman_results,
-                DataFrame(
-                    expected_spearman_results,
-                    index=self.dates[start_date_index:end_date_index + 1],
-                    columns=assets,
-                ),
+            assert_frame_equal(pearson_results, expected_pearson_results)
+
+            expected_spearman_results = DataFrame(
+                data=where(expected_mask, expected_spearman_results, nan),
+                index=dates[start_date_index:end_date_index + 1],
+                columns=assets,
             )
+            assert_frame_equal(spearman_results, expected_spearman_results)
 
     @parameter_space(returns_length=[2, 3], regression_length=[3, 4])
     def test_regression_of_returns_factor(self,
@@ -1353,38 +1331,65 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
         Tests for the built-in factor `RollingLinearRegressionOfReturns`.
         """
         my_asset_column = 0
-        start_date_index = 6
-        end_date_index = 10
+        start_date_index = 14
+        end_date_index = 18
 
-        assets = self.asset_finder.retrieve_all(self.sids)
+        sids = self.sids
+        dates = self.dates
+        assets = self.asset_finder.retrieve_all(sids)
         my_asset = assets[my_asset_column]
-        my_asset_filter = (AssetID() != (my_asset_column + 1))
         num_days = end_date_index - start_date_index + 1
+        num_assets = len(assets)
+
+        cascading_mask = \
+            AssetIDPlusDay() < (sids[-1] + dates[start_date_index].day)
+        expected_cascading_mask_result = make_cascading_boolean_array(
+            shape=(num_days, num_assets),
+        )
+
+        alternating_mask = (AssetIDPlusDay() % 2).eq(0)
+        expected_alternating_mask_result = make_alternating_boolean_array(
+            shape=(num_days, num_assets),
+        )
+
+        expected_no_mask_result = full(
+            shape=(num_days, num_assets), fill_value=True, dtype=bool,
+        )
+
+        masks = cascading_mask, alternating_mask, NotSpecified
+        expected_mask_results = (
+            expected_cascading_mask_result,
+            expected_alternating_mask_result,
+            expected_no_mask_result,
+        )
 
         # The order of these is meant to align with the output of `linregress`.
         outputs = ['beta', 'alpha', 'r_value', 'p_value', 'stderr']
 
-        # Our regression factor requires that its target asset is not filtered
-        # out, so make sure that masking out our target asset does not take
-        # effect. That is, a filter which filters out only our target asset
-        # should produce the same result as if no mask was passed at all.
-        for mask in (NotSpecified, my_asset_filter):
+        for mask, expected_mask in zip(masks, expected_mask_results):
             regression_factor = RollingLinearRegressionOfReturns(
                 target=my_asset,
                 returns_length=returns_length,
                 regression_length=regression_length,
                 mask=mask,
             )
-            results = self.engine.run_pipeline(
-                Pipeline(
-                    columns={
-                        output: getattr(regression_factor, output)
-                        for output in outputs
-                    },
-                ),
-                self.dates[start_date_index],
-                self.dates[end_date_index],
+
+            pipeline = Pipeline(
+                columns={
+                    output: getattr(regression_factor, output)
+                    for output in outputs
+                },
             )
+            if mask is not NotSpecified:
+                pipeline.add(mask, 'mask')
+
+            results = self.engine.run_pipeline(
+                pipeline, dates[start_date_index], dates[end_date_index],
+            )
+            if mask is not NotSpecified:
+                mask_results = results['mask'].unstack()
+                check_arrays(mask_results.values, expected_mask)
+
             output_results = {}
             expected_output_results = {}
             for output in outputs:
@@ -1393,15 +1398,15 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
                     output_results[output], nan,
                 )
 
-            # Run a separate pipeline that calculates returns starting 2 days
-            # prior to our start date. This is because we need
-            # (regression_length - 1) extra days of returns to compute our
-            # expected regressions.
+            # Run a separate pipeline that calculates returns starting
+            # (regression_length - 1) days prior to our start date. This is
+            # because we need (regression_length - 1) extra days of returns to
+            # compute our expected regressions.
             returns = Returns(window_length=returns_length)
             results = self.engine.run_pipeline(
                 Pipeline(columns={'returns': returns}),
-                self.dates[start_date_index - (regression_length - 1)],
-                self.dates[end_date_index],
+                dates[start_date_index - (regression_length - 1)],
+                dates[end_date_index],
             )
             returns_results = results['returns'].unstack()
 
@@ -1424,14 +1429,13 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
                             expected_regression_results[i]
 
             for output in outputs:
-                assert_frame_equal(
-                    output_results[output],
-                    DataFrame(
-                        expected_output_results[output],
-                        index=self.dates[start_date_index:end_date_index + 1],
-                        columns=assets,
-                    ),
+                output_result = output_results[output]
+                expected_output_result = DataFrame(
+                    where(expected_mask, expected_output_results[output], nan),
+                    index=dates[start_date_index:end_date_index + 1],
+                    columns=assets,
                 )
+                assert_frame_equal(output_result, expected_output_result)
 
     def test_correlation_and_regression_with_bad_asset(self):
         """
@@ -1439,8 +1443,8 @@ class ParameterizedFactorTestCase(WithTradingEnvironment, ZiplineTestCase):
         `RollingLinearRegressionOfReturns` raise the proper exception when
         given a nonexistent target asset.
         """
-        start_date_index = 6
-        end_date_index = 10
+        start_date_index = 14
+        end_date_index = 18
         my_asset = Equity(0)
 
         # This filter is arbitrary; the important thing is that we test each

tests/pipeline/test_slice.py

@@ -0,0 +1,516 @@
+"""
+Tests for slicing pipeline terms.
+"""
+from numpy import where
+from pandas import Int64Index, Timestamp
+from pandas.util.testing import assert_frame_equal
+
+from zipline.assets import Asset
+from zipline.errors import (
+    NonExistentAssetInTimeFrame,
+    NonSliceableTerm,
+    NonWindowSafeInput,
+    UnsupportedPipelineOutput,
+)
+from zipline.pipeline import CustomFactor, Pipeline
+from zipline.pipeline.data import USEquityPricing
+from zipline.pipeline.data.testing import TestingDataSet
+from zipline.pipeline.factors import (
+    Returns,
+    RollingLinearRegressionOfReturns,
+    RollingPearsonOfReturns,
+    RollingSpearmanOfReturns,
+    SimpleMovingAverage,
+)
+from zipline.testing import (
+    AssetID,
+    AssetIDPlusDay,
+    check_arrays,
+    OpenPrice,
+    parameter_space,
+)
+from zipline.testing.fixtures import (
+    WithSeededRandomPipelineEngine,
+    ZiplineTestCase,
+)
+from zipline.utils.numpy_utils import datetime64ns_dtype
+
+
+class SliceTestCase(WithSeededRandomPipelineEngine, ZiplineTestCase):
+    sids = ASSET_FINDER_EQUITY_SIDS = Int64Index([1, 2, 3])
+    START_DATE = Timestamp('2015-01-31', tz='UTC')
+    END_DATE = Timestamp('2015-03-01', tz='UTC')
+
+    @classmethod
+    def init_class_fixtures(cls):
+        super(SliceTestCase, cls).init_class_fixtures()
+
+        # Using the date at index 14 as the start date because when running
+        # pipelines, especially those involving correlations or regressions, we
+        # want to make sure there are enough days to look back on. The end date
+        # at index 18 is chosen for convenience, as it makes for a contiguous
+        # five day span.
+        cls.pipeline_start_date = cls.trading_days[14]
+        cls.pipeline_end_date = cls.trading_days[18]
+
+        # Random input for factors.
+        cls.col = TestingDataSet.float_col
+
+    @parameter_space(my_asset_column=[0, 1, 2], window_length_=[1, 2, 3])
+    def test_slice(self, my_asset_column, window_length_):
+        """
+        Test that slices can be created by indexing into a term, and that they
+        have the correct shape when used as inputs.
+        """
+        sids = self.sids
+        my_asset = self.asset_finder.retrieve_asset(self.sids[my_asset_column])
+
+        returns = Returns(window_length=2, inputs=[self.col])
+        returns_slice = returns[my_asset]
+
+        class UsesSlicedInput(CustomFactor):
+            window_length = window_length_
+            inputs = [returns, returns_slice]
+
+            def compute(self, today, assets, out, returns, returns_slice):
+                # Make sure that our slice is the correct shape (i.e. has only
+                # one column) and that it has the same values as the original
+                # returns factor from which it is derived.
+                assert returns_slice.shape == (self.window_length, 1)
+                assert returns.shape == (self.window_length, len(sids))
+                check_arrays(returns_slice[:, 0], returns[:, my_asset_column])
+
+        # Assertions about the expected slice data are made in the `compute`
+        # function of our custom factor above.
+        self.run_pipeline(
+            Pipeline(columns={'uses_sliced_input': UsesSlicedInput()}),
+            self.pipeline_start_date,
+            self.pipeline_end_date,
+        )
+
+    @parameter_space(unmasked_column=[0, 1, 2], slice_column=[0, 1, 2])
+    def test_slice_with_masking(self, unmasked_column, slice_column):
+        """
+        Test that masking a factor that uses slices as inputs does not mask the
+        slice data.
+        """
+        sids = self.sids
+        asset_finder = self.asset_finder
+        start_date = self.pipeline_start_date
+        end_date = self.pipeline_end_date
+
+        # Create a filter that masks out all but a single asset.
+        unmasked_asset = asset_finder.retrieve_asset(sids[unmasked_column])
+        unmasked_asset_only = (AssetID().eq(unmasked_asset.sid))
+
+        # Asset used to create our slice. In the cases where this is different
+        # than `unmasked_asset`, our slice should still have non-missing data
+        # when used as an input to our custom factor. That is, it should not be
+        # masked out.
+        slice_asset = asset_finder.retrieve_asset(sids[slice_column])
+
+        returns = Returns(window_length=2, inputs=[self.col])
+        returns_slice = returns[slice_asset]
+
+        returns_results = self.run_pipeline(
+            Pipeline(columns={'returns': returns}), start_date, end_date,
+        )
+        returns_results = returns_results['returns'].unstack()
+
+        class UsesSlicedInput(CustomFactor):
+            window_length = 1
+            inputs = [returns, returns_slice]
+
+            def compute(self, today, assets, out, returns, returns_slice):
+                # Ensure that our mask correctly affects the `returns` input
+                # and does not affect the `returns_slice` input.
+                assert returns.shape == (1, 1)
+                assert returns_slice.shape == (1, 1)
+                assert returns[0, 0] == \
+                    returns_results.loc[today, unmasked_asset]
+                assert returns_slice[0, 0] == \
+                    returns_results.loc[today, slice_asset]
+
+        columns = {'masked': UsesSlicedInput(mask=unmasked_asset_only)}
+
+        # Assertions about the expected data are made in the `compute` function
+        # of our custom factor above.
+        self.run_pipeline(Pipeline(columns=columns), start_date, end_date)
+
+    def test_adding_slice_column(self):
+        """
+        Test that slices cannot be added as a pipeline column.
+        """
+        my_asset = self.asset_finder.retrieve_asset(self.sids[0])
+        open_slice = OpenPrice()[my_asset]
+
+        with self.assertRaises(UnsupportedPipelineOutput):
+            Pipeline(columns={'open_slice': open_slice})
+
+        pipe = Pipeline(columns={})
+        with self.assertRaises(UnsupportedPipelineOutput):
+            pipe.add(open_slice, 'open_slice')
+
+    def test_loadable_term_slices(self):
+        """
+        Test that slicing loadable terms raises the proper error.
+        """
+        my_asset = self.asset_finder.retrieve_asset(self.sids[0])
+
+        with self.assertRaises(NonSliceableTerm):
+            USEquityPricing.close[my_asset]
+
+    def test_non_existent_asset(self):
+        """
+        Test that indexing into a term with a non-existent asset raises the
+        proper exception.
+        """
+        my_asset = Asset(0)
+        returns = Returns(window_length=2, inputs=[self.col])
+        returns_slice = returns[my_asset]
+
+        class UsesSlicedInput(CustomFactor):
+            window_length = 1
+            inputs = [returns_slice]
+
+            def compute(self, today, assets, out, returns_slice):
+                pass
+
+        with self.assertRaises(NonExistentAssetInTimeFrame):
+            self.run_pipeline(
+                Pipeline(columns={'uses_sliced_input': UsesSlicedInput()}),
+                self.pipeline_start_date,
+                self.pipeline_end_date,
+            )
+
+    def test_window_safety_of_slices(self):
+        """
+        Test that slices correctly inherit the `window_safe` property of the
+        term from which they are derived.
+        """
+        col = self.col
+        my_asset = self.asset_finder.retrieve_asset(self.sids[0])
+
+        # SimpleMovingAverage is not window safe.
+        sma = SimpleMovingAverage(inputs=[self.col], window_length=10)
+        sma_slice = sma[my_asset]
+
+        class UsesSlicedInput(CustomFactor):
+            window_length = 1
+            inputs = [sma_slice]
+
+            def compute(self, today, assets, out, sma_slice):
+                pass
+
+        with self.assertRaises(NonWindowSafeInput):
+            self.run_pipeline(
+                Pipeline(columns={'uses_sliced_input': UsesSlicedInput()}),
+                self.pipeline_start_date,
+                self.pipeline_end_date,
+            )
+
+        # Make sure that slices of custom factors are not window safe.
+        class MyUnsafeFactor(CustomFactor):
+            window_length = 1
+            inputs = [col]
+
+            def compute(self, today, assets, out, col):
+                pass
+
+        my_unsafe_factor = MyUnsafeFactor()
+        my_unsafe_factor_slice = my_unsafe_factor[my_asset]
+
+        class UsesSlicedInput(CustomFactor):
+            window_length = 1
+            inputs = [my_unsafe_factor_slice]
+
+            def compute(self, today, assets, out, my_unsafe_factor_slice):
+                pass
+
+        with self.assertRaises(NonWindowSafeInput):
+            self.run_pipeline(
+                Pipeline(columns={'uses_sliced_input': UsesSlicedInput()}),
+                self.pipeline_start_date,
+                self.pipeline_end_date,
+            )
+
+        # Create a window safe factor.
+        class MySafeFactor(CustomFactor):
+            window_length = 1
+            inputs = [col]
+            window_safe = True
+
+            def compute(self, today, assets, out, col):
+                pass
+
+        my_safe_factor = MySafeFactor()
+        my_safe_factor_slice = my_safe_factor[my_asset]
+
+        # Make sure that correlations are not safe if either the factor *or*
+        # the target slice are not window safe.
+        with self.assertRaises(NonWindowSafeInput):
+            my_unsafe_factor.pearsonr(
+                target=my_safe_factor_slice, correlation_length=10,
+            )
+
+        with self.assertRaises(NonWindowSafeInput):
+            my_safe_factor.pearsonr(
+                target=my_unsafe_factor_slice, correlation_length=10,
+            )
+
+    def test_single_column_output(self):
+        """
+        Tests for custom factors that compute a 1D out.
+        """
+        start_date = self.pipeline_start_date
+        end_date = self.pipeline_end_date
+
+        alternating_mask = (AssetIDPlusDay() % 2).eq(0)
+        cascading_mask = AssetIDPlusDay() < (self.sids[-1] + start_date.day)
+
+        class SingleColumnOutput(CustomFactor):
+            window_length = 1
+            inputs = [self.col]
+            window_safe = True
+            ndim = 1
+
+            def compute(self, today, assets, out, col):
+                # Because we specified ndim as 1, `out` should be a singleton
+                # array but `close` should be a regular sized input.
+                assert out.shape == (1,)
+                assert col.shape == (1, 3)
+                out[:] = col.sum()
+
+        # Since we cannot add single column output factors as pipeline
+        # columns, we have to test its output through another factor.
+        class UsesSingleColumnOutput(CustomFactor):
+            window_length = 1
+            inputs = [SingleColumnOutput()]
+
+            def compute(self, today, assets, out, single_column_output):
+                # Make sure that `single_column` has the correct shape. That
+                # is, it should always have one column regardless of any mask
+                # passed to `UsesSingleColumnInput`.
+                assert single_column_output.shape == (1, 1)
+
+        for mask in (alternating_mask, cascading_mask):
+            columns = {
+                'uses_single_column_output': UsesSingleColumnOutput(),
+                'uses_single_column_output_masked': UsesSingleColumnOutput(
+                    mask=mask,
+                ),
+            }
+
+            # Assertions about the expected shapes of our data are made in the
+            # `compute` function of our custom factors above.
+            self.run_pipeline(Pipeline(columns=columns), start_date, end_date)
+
+    def test_masked_single_column_output(self):
+        """
+        Tests for masking custom factors that compute a 1D out.
+        """
+        start_date = self.pipeline_start_date
+        end_date = self.pipeline_end_date
+
+        alternating_mask = (AssetIDPlusDay() % 2).eq(0)
+        cascading_mask = AssetIDPlusDay() < (self.sids[-1] + start_date.day)
+        alternating_mask.window_safe = True
+        cascading_mask.window_safe = True
+
+        for mask in (alternating_mask, cascading_mask):
+            class SingleColumnOutput(CustomFactor):
+                window_length = 1
+                inputs = [self.col, mask]
+                window_safe = True
+                ndim = 1
+
+                def compute(self, today, assets, out, col, mask):
+                    # Because we specified ndim as 1, `out` should always be a
+                    # singleton array but `close` should be a sized based on
+                    # the mask we passed.
+                    assert out.shape == (1,)
+                    assert col.shape == (1, mask.sum())
+                    out[:] = col.sum()
+
+            # Since we cannot add single column output factors as pipeline
+            # columns, we have to test its output through another factor.
+            class UsesSingleColumnInput(CustomFactor):
+                window_length = 1
+                inputs = [self.col, mask, SingleColumnOutput(mask=mask)]
+
+                def compute(self,
+                            today,
+                            assets,
+                            out,
+                            col,
+                            mask,
+                            single_column_output):
+                    # Make sure that `single_column` has the correct value
+                    # based on the masked it used.
+                    assert single_column_output.shape == (1, 1)
+                    single_column_output_value = single_column_output[0][0]
+                    expected_value = where(mask, col, 0).sum()
+                    assert single_column_output_value == expected_value
+
+            columns = {'uses_single_column_input': UsesSingleColumnInput()}
+
+            # Assertions about the expected shapes of our data are made in the
+            # `compute` function of our custom factors above.
+            self.run_pipeline(Pipeline(columns=columns), start_date, end_date)
+
+    @parameter_space(returns_length=[2, 3], correlation_length=[3, 4])
+    def test_factor_correlation_methods(self,
+                                        returns_length,
+                                        correlation_length):
+        """
+        Ensure that `Factor.pearsonr` and `Factor.spearmanr` are consistent
+        with the built-in factors `RollingPearsonOfReturns` and
+        `RollingSpearmanOfReturns`.
+        """
+        my_asset = self.asset_finder.retrieve_asset(self.sids[0])
+
+        returns = Returns(window_length=returns_length, inputs=[self.col])
+        returns_slice = returns[my_asset]
+
+        pearson = returns.pearsonr(
+            target=returns_slice, correlation_length=correlation_length,
+        )
+        spearman = returns.spearmanr(
+            target=returns_slice, correlation_length=correlation_length,
+        )
+        expected_pearson = RollingPearsonOfReturns(
+            target=my_asset,
+            returns_length=returns_length,
+            correlation_length=correlation_length,
+        )
+        expected_spearman = RollingSpearmanOfReturns(
+            target=my_asset,
+            returns_length=returns_length,
+            correlation_length=correlation_length,
+        )
+
+        # These built-ins construct their own Returns factor to use as inputs,
+        # so the only way to set our own inputs is to do so after the fact.
+        # This should not be done in practice. It is necessary here because we
+        # want Returns to use our random data as an input, but by default it is
+        # using USEquityPricing.close.
+        expected_pearson.inputs = [returns, returns_slice]
+        expected_spearman.inputs = [returns, returns_slice]
+
+        columns = {
+            'pearson': pearson,
+            'spearman': spearman,
+            'expected_pearson': expected_pearson,
+            'expected_spearman': expected_spearman,
+        }
+
+        results = self.run_pipeline(
+            Pipeline(columns=columns),
+            self.pipeline_start_date,
+            self.pipeline_end_date,
+        )
+        pearson_results = results['pearson'].unstack()
+        spearman_results = results['spearman'].unstack()
+        expected_pearson_results = results['expected_pearson'].unstack()
+        expected_spearman_results = results['expected_spearman'].unstack()
+
+        assert_frame_equal(pearson_results, expected_pearson_results)
+        assert_frame_equal(spearman_results, expected_spearman_results)
+
+        # Make sure we cannot call the correlation methods on factors or slices
+        # of dtype `datetime64[ns]`.
+        class DateFactor(CustomFactor):
+            window_length = 1
+            inputs = []
+            dtype = datetime64ns_dtype
+            window_safe = True
+
+            def compute(self, today, assets, out):
+                pass
+
+        date_factor = DateFactor()
+        date_factor_slice = date_factor[my_asset]
+
+        with self.assertRaises(TypeError):
+            date_factor.pearsonr(
+                target=returns_slice, correlation_length=correlation_length,
+            )
+        with self.assertRaises(TypeError):
+            date_factor.spearmanr(
+                target=returns_slice, correlation_length=correlation_length,
+            )
+        with self.assertRaises(TypeError):
+            returns.pearsonr(
+                target=date_factor_slice,
+                correlation_length=correlation_length,
+            )
+        with self.assertRaises(TypeError):
+            returns.pearsonr(
+                target=date_factor_slice,
+                correlation_length=correlation_length,
+            )
+
+    @parameter_space(returns_length=[2, 3], regression_length=[3, 4])
+    def test_factor_regression_method(self, returns_length, regression_length):
+        """
+        Ensure that `Factor.linear_regression` is consistent with the built-in
+        factor `RollingLinearRegressionOfReturns`.
+        """
+        my_asset = self.asset_finder.retrieve_asset(self.sids[0])
+
+        returns = Returns(window_length=returns_length, inputs=[self.col])
+        returns_slice = returns[my_asset]
+
+        regression = returns.linear_regression(
+            target=returns_slice, regression_length=regression_length,
+        )
+        expected_regression = RollingLinearRegressionOfReturns(
+            target=my_asset,
+            returns_length=returns_length,
+            regression_length=regression_length,
+        )
+
+        # These built-ins construct their own Returns factor to use as inputs,
+        # so the only way to set our own inputs is to do so after the fact.
+        # This should not be done in practice. It is necessary here because we
+        # want Returns to use our random data as an input, but by default it is
+        # using USEquityPricing.close.
+        expected_regression.inputs = [returns, returns_slice]
+
+        columns = {
+            'regression': regression,
+            'expected_regression': expected_regression,
+        }
+
+        results = self.run_pipeline(
+            Pipeline(columns=columns),
+            self.pipeline_start_date,
+            self.pipeline_end_date,
+        )
+        regression_results = results['regression'].unstack()
+        expected_regression_results = results['expected_regression'].unstack()
+
+        assert_frame_equal(regression_results, expected_regression_results)
+
+        # Make sure we cannot call the linear regression method on factors or
+        # slices of dtype `datetime64[ns]`.
+        class DateFactor(CustomFactor):
+            window_length = 1
+            inputs = []
+            dtype = datetime64ns_dtype
+            window_safe = True
+
+            def compute(self, today, assets, out):
+                pass
+
+        date_factor = DateFactor()
+        date_factor_slice = date_factor[my_asset]
+
+        with self.assertRaises(TypeError):
+            date_factor.linear_regression(
+                target=returns_slice, regression_length=regression_length,
+            )
+        with self.assertRaises(TypeError):
+            returns.linear_regression(
+                target=date_factor_slice, regression_length=regression_length,
+            )

tests/pipeline/test_term.py

@@ -5,6 +5,7 @@ from collections import Counter
 from itertools import product
 from unittest import TestCase
 
+from zipline.assets import Asset
 from zipline.errors import (
     DTypeNotSpecified,
     InvalidOutputName,
@@ -25,9 +26,10 @@ from zipline.pipeline import (
 )
 from zipline.pipeline.data import Column, DataSet
 from zipline.pipeline.data.testing import TestingDataSet
-from zipline.pipeline.factors import RecarrayField
-from zipline.pipeline.term import AssetExists, NotSpecified
 from zipline.pipeline.expression import NUMEXPR_MATH_FUNCS
+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.predicates import assert_equal, assert_raises
 from zipline.utils.numpy_utils import (
@@ -96,6 +98,18 @@ class MultipleOutputs(CustomFactor):
         return
 
 
+class GenericFilter(Filter):
+    dtype = bool_dtype
+    window_length = 0
+    inputs = []
+
+
+class GenericClassifier(Classifier):
+    dtype = categorical_dtype
+    window_length = 0
+    inputs = []
+
+
 def gen_equivalent_factors():
     """
     Return an iterator of SomeFactor instances that should all be the same
@@ -268,6 +282,21 @@ class ObjectIdentityTestCase(TestCase):
         self.assertIs(alpha, multiple_outputs.alpha)
         self.assertIs(beta, multiple_outputs.beta)
 
+    def test_instance_caching_of_slices(self):
+        my_asset = Asset(1)
+
+        f = GenericCustomFactor()
+        f_slice = f[my_asset]
+        self.assertIs(f_slice, Slice(GenericCustomFactor(), my_asset))
+
+        f = GenericFilter()
+        f_slice = f[my_asset]
+        self.assertIs(f_slice, Slice(GenericFilter(), my_asset))
+
+        c = GenericClassifier()
+        c_slice = c[my_asset]
+        self.assertIs(c_slice, Slice(GenericClassifier(), my_asset))
+
     def test_instance_non_caching(self):
 
         f = SomeFactor()

tests/test_testing.py

@@ -4,7 +4,15 @@ Tests for our testing utilities.
 from itertools import product
 from unittest import TestCase
 
-from zipline.testing import parameter_space
+from numpy import array, empty
+
+from zipline.testing import (
+    check_arrays,
+    make_alternating_boolean_array,
+    make_cascading_boolean_array,
+    parameter_space,
+)
+from zipline.utils.numpy_utils import bool_dtype
 
 
 class TestParameterSpace(TestCase):
@@ -38,3 +46,66 @@ class TestParameterSpace(TestCase):
         # our {setUp,tearDown}Class won't be called if, for example,
         # `parameter_space` returns None.
         pass
+
+
+class TestMakeBooleanArray(TestCase):
+
+    def test_make_alternating_boolean_array(self):
+        check_arrays(
+            make_alternating_boolean_array((3, 3)),
+            array(
+                [[True,  False,  True],
+                 [False,  True, False],
+                 [True,  False,  True]]
+            ),
+        )
+        check_arrays(
+            make_alternating_boolean_array((3, 3), first_value=False),
+            array(
+                [[False,  True, False],
+                 [True,  False,  True],
+                 [False,  True, False]]
+            ),
+        )
+        check_arrays(
+            make_alternating_boolean_array((1, 3)),
+            array([[True, False, True]]),
+        )
+        check_arrays(
+            make_alternating_boolean_array((3, 1)),
+            array([[True], [False], [True]]),
+        )
+        check_arrays(
+            make_alternating_boolean_array((3, 0)),
+            empty((3, 0), dtype=bool_dtype),
+        )
+
+    def test_make_cascading_boolean_array(self):
+        check_arrays(
+            make_cascading_boolean_array((3, 3)),
+            array(
+                [[True,   True, False],
+                 [True,  False, False],
+                 [False, False, False]]
+            ),
+        )
+        check_arrays(
+            make_cascading_boolean_array((3, 3), first_value=False),
+            array(
+                [[False, False, True],
+                 [False,  True, True],
+                 [True,   True, True]]
+            ),
+        )
+        check_arrays(
+            make_cascading_boolean_array((1, 3)),
+            array([[True, True, False]]),
+        )
+        check_arrays(
+            make_cascading_boolean_array((3, 1)),
+            array([[False], [False], [False]]),
+        )
+        check_arrays(
+            make_cascading_boolean_array((3, 0)),
+            empty((3, 0), dtype=bool_dtype),
+        )