Merge pull request #1280 from quantopian/bad-pipeline-columns

BUG: Fail fast on invalid pipeline columns

tests/pipeline/test_pipeline.py

@@ -63,6 +63,9 @@ class PipelineTestCase(TestCase):
         with self.assertRaises(TypeError):
             Pipeline({}, SomeFactor())
 
+        with self.assertRaises(TypeError):
+            Pipeline({'open': USEquityPricing.open})
+
         Pipeline({}, SomeFactor() > 5)
 
     def test_add(self):
@@ -78,6 +81,9 @@ class PipelineTestCase(TestCase):
         with self.assertRaises(TypeError):
             p.add(f, 1)
 
+        with self.assertRaises(TypeError):
+            p.add(USEquityPricing.open, 'open')
+
     def test_overwrite(self):
         p = Pipeline()
         f = SomeFactor()

zipline/pipeline/pipeline.py

@@ -1,6 +1,6 @@
 from zipline.utils.input_validation import expect_types, optional
 
-from .term import Term, AssetExists
+from .term import AssetExists, ComputableTerm, Term
 from .filters import Filter
 from .graph import TermGraph
 
@@ -37,6 +37,14 @@ class Pipeline(object):
 
         if columns is None:
             columns = {}
+        for column_name, term in columns.items():
+            if not isinstance(term, ComputableTerm):
+                raise TypeError(
+                    "Column {column_name!r} contains an invalid pipeline term "
+                    "({term}). Did you mean to append '.latest'?".format(
+                        column_name=column_name, term=term,
+                    )
+                )
         self._columns = columns
         self._screen = screen
 
@@ -79,6 +87,12 @@ class Pipeline(object):
             else:
                 raise KeyError("Column '{}' already exists.".format(name))
 
+        if not isinstance(term, ComputableTerm):
+            raise TypeError(
+                "{term} is not a valid pipeline column. Did you mean to "
+                "append '.latest'?".format(term=term)
+            )
+
         self._columns[name] = term
 
     @expect_types(name=str)