# This is a combination of 8 commits.

# The first commit's message is:
BUG: ignore sids in deltas missing from asset index.

# This is the 2nd commit message:

MAINT: use correct debugger.

# This is the 3rd commit message:

MAINT: fix set add.

# This is the 4th commit message:

WIP: move sid filtering.

# This is the 5th commit message:

WIP: move filtering logic.

# This is the 6th commit message:

WIP: working test.

# This is the 7th commit message:

TST: clean up test.

# This is the 8th commit message:

STY: fix flake8.

tests/pipeline/test_blaze.py

@@ -37,8 +37,11 @@ from zipline.utils.numpy_utils import (
     int64_dtype,
     repeat_last_axis,
 )
-from zipline.utils.test_utils import tmp_asset_finder, make_simple_equity_info
-
+from zipline.utils.test_utils import (
+    tmp_asset_finder,
+    make_simple_equity_info,
+    make_test_handler
+)
 
 nameof = op.attrgetter('name')
 dtypeof = op.attrgetter('dtype')
@@ -1128,3 +1131,82 @@ class BlazeToPipelineTestCase(TestCase):
                 window_length=3,
                 compute_fn=op.itemgetter(-1),
             )
+
+    def test_ignore_nonexistent_delta(self):
+        missing_sid = ord('E')
+        handler = make_test_handler(self)
+        asset_info = asset_infos[0][0]
+        base_dates = pd.DatetimeIndex([
+            pd.Timestamp('2014-01-01'),
+            pd.Timestamp('2014-01-04')
+        ])
+        repeated_dates = base_dates.repeat(4)
+        baseline = pd.DataFrame({
+            'sid': (self.sids + (missing_sid,)) * 2,
+            'value': (0., 1., 2., 3., 1., 2., 3., 4.),
+            'int_value': (0, 1, 2, 3, 1, 2, 3, 4.),
+            'asof_date': repeated_dates,
+            'timestamp': repeated_dates
+        })
+        expr = bz.Data(baseline, name='expr', dshape=self.dshape)
+        deltas = bz.Data(
+            odo(
+                bz.transform(
+                    expr,
+                    value=expr.value + 10,
+                    timestamp=expr.timestamp + timedelta(days=1),
+                ),
+                pd.DataFrame,
+            ),
+            name='delta',
+            dshape=self.dshape,
+        )
+        expected_views = keymap(pd.Timestamp, {
+            '2014-01-03': np.array([[10.0, 11.0, 12.0],
+                                    [10.0, 11.0, 12.0],
+                                    [10.0, 11.0, 12.0]]),
+            '2014-01-06': np.array([[10.0, 11.0, 12.0],
+                                    [10.0, 11.0, 12.0],
+                                    [11.0, 12.0, 13.0]]),
+        })
+        if len(asset_info) == 4:
+            expected_views = valmap(
+                lambda view: np.c_[view, [np.nan, np.nan, np.nan]],
+                expected_views,
+            )
+            expected_output_buffer = [10, 11, 12, np.nan, 11, 12, 13, np.nan]
+        else:
+            expected_output_buffer = [10, 11, 12, 11, 12, 13]
+
+        cal = pd.DatetimeIndex([
+            pd.Timestamp('2014-01-01'),
+            pd.Timestamp('2014-01-02'),
+            pd.Timestamp('2014-01-03'),
+            # omitting the 4th and 5th to simulate a weekend
+            pd.Timestamp('2014-01-06'),
+        ])
+        with handler.applicationbound():
+            with tmp_asset_finder(equities=asset_info) as finder:
+                expected_output = pd.DataFrame(
+                    expected_output_buffer,
+                    index=pd.MultiIndex.from_product((
+                        sorted(expected_views.keys()),
+                        finder.retrieve_all(asset_info.index),
+                    )),
+                    columns=('value',),
+                )
+                self._run_pipeline(
+                    expr,
+                    deltas,
+                    expected_views,
+                    expected_output,
+                    finder,
+                    calendar=cal,
+                    start=cal[2],
+                    end=cal[-1],
+                    window_length=3,
+                    compute_fn=op.itemgetter(-1),
+                )
+            message = handler.records[0].message
+            exp_msg = "Didn't find the following sids in asset index: {}"
+            self.assertIn(exp_msg.format(set([missing_sid])), message)

zipline/pipeline/loaders/blaze/core.py

@@ -129,6 +129,7 @@ from collections import namedtuple, defaultdict
 from copy import copy
 from functools import partial, reduce
 from itertools import count
+import logbook
 import warnings
 from weakref import WeakKeyDictionary
 
@@ -194,7 +195,7 @@ traversable_nodes = (
 )
 is_invalid_deltas_node = complement(flip(isinstance, valid_deltas_node_types))
 get__name__ = op.attrgetter('__name__')
-
+log = logbook.Logger('BlazeLoader')
 
 class ExprData(namedtuple('ExprData', 'expr deltas odo_kwargs')):
     """A pair of expressions and data resources. The expresions will be
@@ -765,8 +766,8 @@ def adjustments_from_deltas_with_sids(dense_dates,
                                       column_name,
                                       asset_idx,
                                       deltas):
-    """Collect all the adjustments that occur in a dataset that does not
-    have a sid column.
+    """Collect all the adjustments that occur in a dataset that has a sid
+    column.
 
     Parameters
     ----------
@@ -952,6 +953,17 @@ class BlazeLoader(dict):
                 columns=added_query_fields + list(map(getname, columns)),
             )
         )
+        if not materialized_deltas.empty and have_sids:
+            missing_sids = materialized_deltas[
+                ~materialized_deltas.sid.isin(assets)
+            ]
+            log.debug(
+                "Didn't find the following sids in asset index: {}".format(
+                    set(missing_sids.sid))
+            )
+            materialized_deltas = materialized_deltas[
+                materialized_deltas.sid.isin(assets)
+            ]
 
         if data_query_time is not None:
             for m in (materialized_expr, materialized_deltas):

zipline/utils/test_utils.py

@@ -12,7 +12,7 @@ import shutil
 from string import ascii_uppercase
 import tempfile
 
-from logbook import FileHandler
+from logbook import FileHandler, TestHandler
 from mock import patch
 from numpy.testing import assert_allclose, assert_array_equal
 import numpy as np
@@ -881,3 +881,9 @@ def parameter_space(**params):
         param_sets = product(*(params[name] for name in argnames))
         return subtest(param_sets, *argnames)(f)
     return decorator
+
+
+def make_test_handler(testcase, *args, **kwargs):
+    handler = TestHandler(*args, **kwargs)
+    testcase.addCleanup(handler.close)
+    return handler