From 9c37011a3885b5a055f59d12e2ece4ef02787c36 Mon Sep 17 00:00:00 2001 From: llllllllll Date: Tue, 6 Oct 2015 18:07:40 -0400 Subject: [PATCH] BUG: Only simple expressions for array-like dshape --- tests/pipeline/test_blaze.py | 11 +++++---- zipline/pipeline/loaders/blaze.py | 40 +++++++++++++++++++++++-------- 2 files changed, 36 insertions(+), 15 deletions(-) diff --git a/tests/pipeline/test_blaze.py b/tests/pipeline/test_blaze.py index 5a1d39c2..5d642d37 100644 --- a/tests/pipeline/test_blaze.py +++ b/tests/pipeline/test_blaze.py @@ -257,11 +257,12 @@ class BlazeToPipelineTestCase(TestCase): loader=self.garbage_loader, ) - from_blaze( - expr.value + 1, # put an Add in the column - deltas=None, - loader=self.garbage_loader, - ) + with self.assertRaises(TypeError): + from_blaze( + expr.value + 1, # put an Add in the column + deltas=None, + loader=self.garbage_loader, + ) deltas = bz.Data( pd.DataFrame(columns=self.df.columns), diff --git a/zipline/pipeline/loaders/blaze.py b/zipline/pipeline/loaders/blaze.py index a7067c97..9336a8a8 100644 --- a/zipline/pipeline/loaders/blaze.py +++ b/zipline/pipeline/loaders/blaze.py @@ -160,6 +160,10 @@ valid_deltas_node_types = ( bz.expr.ReLabel, bz.expr.Symbol, ) +traversable_nodes = ( + bz.expr.Field, + bz.expr.Label, +) is_invalid_deltas_node = complement(flip(isinstance, valid_deltas_node_types)) getname = attrgetter('__name__') @@ -462,16 +466,30 @@ def from_blaze(expr, # Check if this is a single column out of a dataset. single_column = None + if bz.ndim(expr) != 1: + raise TypeError( + 'expression was not tabular or array-like,' + ' too many dimensions: %d' % bz.ndim(expr) + ) if isscalar(expr.dshape.measure): # This is a single column. Record which column we are to return # but create the entire dataset. - single_column = expr._name - col = expr - for expr in expr._subterms(): - if isrecord(expr.dshape.measure): - break - else: - expr = bz.Data(col, name=single_column) + single_column = rename = expr._name + field_hit = False + if not isinstance(expr, traversable_nodes): + raise TypeError( + "expression '%s' was array-like but not a simple field of" + " some larger table" % str(expr), + ) + while isinstance(expr, traversable_nodes): + if isinstance(expr, bz.expr.Field): + if not field_hit: + field_hit = True + else: + break + rename = expr._name + expr = expr._child + expr = expr.relabel({rename: single_column}) measure = expr.dshape.measure if not isrecord(measure) or AD_FIELD_NAME not in measure.names: @@ -549,7 +567,7 @@ def inline_novel_deltas(base, deltas, dates): (base, deltas.loc[ (get_indexes(deltas[TS_FIELD_NAME].values, 'right') - - get_indexes(deltas[AD_FIELD_NAME].values, 'letf')) <= 1 + get_indexes(deltas[AD_FIELD_NAME].values, 'left')) <= 1 ].drop(AD_FIELD_NAME, 1)), ignore_index=True, ) @@ -613,10 +631,11 @@ def adjustments_from_deltas_no_sids(dates, adjustments : dict[idx -> Float64Overwrite] The adjustments dictionary to feed to the adjusted array. """ + ad_series = deltas.loc[:, AD_FIELD_NAME] return { dates.get_loc(kd): tuple( overwrite_from_dates( - deltas.loc[kd, AD_FIELD_NAME], + ad_series.loc[kd], dates, dense_dates, n, @@ -653,12 +672,13 @@ def adjustments_from_deltas_with_sids(dates, adjustments : dict[idx -> Float64Overwrite] The adjustments dictionary to feed to the adjusted array. """ + ad_series = deltas[AD_FIELD_NAME] adjustments = defaultdict(list) for sid_idx, (sid, per_sid) in enumerate(deltas[column_name].iteritems()): for kd, v in per_sid.iteritems(): adjustments[dates.get_loc(kd)].append( overwrite_from_dates( - deltas[AD_FIELD_NAME].loc[kd, sid], + ad_series.loc[kd, sid], dates, dense_dates, sid_idx,