diff --git a/tests/pipeline/test_blaze.py b/tests/pipeline/test_blaze.py index cf5cb1c8..29885a30 100644 --- a/tests/pipeline/test_blaze.py +++ b/tests/pipeline/test_blaze.py @@ -21,7 +21,7 @@ from toolz.curried import operator as op from zipline.assets.synthetic import make_simple_equity_info from zipline.pipeline import Pipeline, CustomFactor -from zipline.pipeline.data import DataSet, BoundColumn +from zipline.pipeline.data import DataSet, BoundColumn, Column from zipline.pipeline.engine import SimplePipelineEngine from zipline.pipeline.loaders.blaze import ( from_blaze, @@ -29,16 +29,21 @@ from zipline.pipeline.loaders.blaze import ( NoMetaDataWarning, ) from zipline.pipeline.loaders.blaze.core import ( + ExprData, NonPipelineField, ) -from zipline.testing import parameter_space +from zipline.testing import ( + ZiplineTestCase, + parameter_space, + tmp_asset_finder, +) from zipline.testing.fixtures import WithAssetFinder from zipline.utils.numpy_utils import ( float64_dtype, int64_dtype, repeat_last_axis, ) -from zipline.testing import tmp_asset_finder, ZiplineTestCase +from zipline.testing.predicates import assert_equal, assert_isidentical nameof = op.attrgetter('name') dtypeof = op.attrgetter('dtype') @@ -208,6 +213,108 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase): self.assertIn("'asof_date'", str(e.exception)) self.assertIn(repr(str(expr.dshape.measure)), str(e.exception)) + def test_missing_timestamp(self): + expr = bz.data( + self.df.loc[:, ['sid', 'value', 'asof_date']], + name='expr', + dshape=""" + var * { + sid: ?int64, + value: float64, + asof_date: datetime, + }""", + ) + + loader = BlazeLoader() + + from_blaze( + expr, + loader=loader, + no_deltas_rule='ignore', + no_checkpoints_rule='ignore', + ) + + self.assertEqual(len(loader), 1) + exprdata, = loader.values() + + assert_isidentical( + exprdata.expr, + bz.transform(expr, timestamp=expr.asof_date), + ) + + def test_from_blaze_no_resources_dataset_expr(self): + expr = bz.symbol('expr', self.dshape) + + with self.assertRaises(ValueError) as e: + from_blaze( + expr, + loader=self.garbage_loader, + no_deltas_rule='ignore', + no_checkpoints_rule='ignore', + missing_values=self.missing_values, + ) + assert_equal( + str(e.exception), + 'no resources provided to compute expr', + ) + + @parameter_space(metadata={'deltas', 'checkpoints'}) + def test_from_blaze_no_resources_metadata_expr(self, metadata): + expr = bz.data(self.df, name='expr', dshape=self.dshape) + metadata_expr = bz.symbol('metadata', self.dshape) + + with self.assertRaises(ValueError) as e: + from_blaze( + expr, + loader=self.garbage_loader, + no_deltas_rule='ignore', + no_checkpoints_rule='ignore', + missing_values=self.missing_values, + **{metadata: metadata_expr} + ) + assert_equal( + str(e.exception), + 'no resources provided to compute %s' % metadata, + ) + + def test_from_blaze_mixed_resources_dataset_expr(self): + expr = bz.data(self.df, name='expr', dshape=self.dshape) + + with self.assertRaises(ValueError) as e: + from_blaze( + expr, + resources={expr: self.df}, + loader=self.garbage_loader, + no_deltas_rule='ignore', + no_checkpoints_rule='ignore', + missing_values=self.missing_values, + ) + assert_equal( + str(e.exception), + 'explicit and implicit resources provided to compute expr', + ) + + @parameter_space(metadata={'deltas', 'checkpoints'}) + def test_from_blaze_mixed_resources_metadata_expr(self, metadata): + expr = bz.symbol('expr', self.dshape) + metadata_expr = bz.data(self.df, name=metadata, dshape=self.dshape) + + with self.assertRaises(ValueError) as e: + from_blaze( + expr, + resources={metadata_expr: self.df}, + loader=self.garbage_loader, + no_deltas_rule='ignore', + no_checkpoints_rule='ignore', + missing_values=self.missing_values, + **{metadata: metadata_expr} + ) + assert_equal( + str(e.exception), + 'explicit and implicit resources provided to compute %s' % + metadata, + ) + @parameter_space(deltas={True, False}, checkpoints={True, False}) def test_auto_metadata(self, deltas, checkpoints): select_level = op.getitem(('ignore', 'raise')) @@ -1486,3 +1593,43 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase): window_length=1, compute_fn=op.itemgetter(-1), ) + + +class MiscTestCase(ZiplineTestCase): + def test_exprdata_repr(self): + strd = set() + + class BadRepr(object): + """A class which cannot be repr'd. + """ + def __init__(self, name): + self._name = name + + def __repr__(self): # pragma: no cover + raise AssertionError('ayy') + + def __str__(self): + strd.add(self) + return self._name + + assert_equal( + repr(ExprData( + expr=BadRepr('expr'), + deltas=BadRepr('deltas'), + checkpoints=BadRepr('checkpoints'), + odo_kwargs={'a': 'b'}, + )), + "ExprData(expr='expr', deltas='deltas'," + " checkpoints='checkpoints', odo_kwargs={'a': 'b'})", + ) + + def test_blaze_loader_repr(self): + assert_equal(repr(BlazeLoader()), '') + + def test_blaze_loader_lookup_failure(self): + class D(DataSet): + c = Column(dtype='float64') + + with self.assertRaises(KeyError) as e: + BlazeLoader()(D.c) + assert_equal(str(e.exception), 'D.c::float64') diff --git a/zipline/pipeline/loaders/blaze/core.py b/zipline/pipeline/loaders/blaze/core.py index a23d5b70..47cf5967 100644 --- a/zipline/pipeline/loaders/blaze/core.py +++ b/zipline/pipeline/loaders/blaze/core.py @@ -140,7 +140,6 @@ from __future__ import division, absolute_import from abc import ABCMeta, abstractproperty from collections import namedtuple, defaultdict from copy import copy -import datetime from functools import partial from itertools import count import warnings @@ -240,7 +239,7 @@ class ExprData(namedtuple('ExprData', 'expr deltas checkpoints odo_kwargs')): return super(ExprData, cls).__repr__(cls( str(self.expr), str(self.deltas), - str(self.checkpoint), + str(self.checkpoints), self.odo_kwargs, )) @@ -257,7 +256,7 @@ class InvalidField(with_metaclass(ABCMeta)): The shape of the field. """ @abstractproperty - def error_format(self): + def error_format(self): # pragma: no cover raise NotImplementedError('error_format') def __init__(self, field, type_): @@ -282,14 +281,6 @@ class NonPipelineField(InvalidField): ) -class NotPipelineCompatible(TypeError): - """Exception used to indicate that a dshape is not Pipeline API - compatible. - """ - def __str__(self): - return "'%s' is a non Pipeline API compatible type'" % self.args - - _new_names = ('BlazeDataSet_%d' % n for n in count()) @@ -369,7 +360,7 @@ def new_dataset(expr, deltas, missing_values): # unicode is a name error in py3 but the branch is only hit # when we are in python 2. - if PY2 and isinstance(name, unicode): # noqa + if PY2 and isinstance(name, unicode): # pragma: no cover # noqa name = name.encode('utf-8') return type(name, (DataSet,), columns) @@ -674,8 +665,9 @@ def from_blaze(expr, ) # Ensure that we have a data resource to execute the query against. - _check_resources('dataset_expr', dataset_expr, resources) + _check_resources('expr', dataset_expr, resources) _check_resources('deltas', deltas, resources) + _check_resources('checkpoints', checkpoints, resources) # Create or retrieve the Pipeline API dataset. if missing_values is None: @@ -883,31 +875,6 @@ def adjustments_from_deltas_with_sids(dense_dates, return dict(adjustments) # no subclasses of dict -def _checkpoint_ts(lower_dt): - """Given a lower time bound for a query, get the date in the checkpoint - table to query for. - - Parameters - ---------- - lower_dt : datetime - The lower time bound for the query. - - Returns - ------- - checkpoint_ts : pd.Timestamp - The date in the checkpoint table to query for. - """ - date = lower_dt.date() - return pd.Timestamp.combine( - date.replace( - day=1, - month=(date.month - 2) % 12 + 1, - year=date.year - 1 if date.month == 1 else date.year, - ), - datetime.time(0), - ).tz_localize('US/Eastern') - - class BlazeLoader(dict): """A PipelineLoader for datasets constructed with ``from_blaze``. @@ -945,6 +912,12 @@ class BlazeLoader(dict): return self raise KeyError(column) + def __repr__(self): + return '<%s: %s>' % ( + type(self).__name__, + super(BlazeLoader, self).__repr__(), + ) + def load_adjusted_array(self, columns, dates, assets, mask): return dict( concat(map( diff --git a/zipline/testing/predicates.py b/zipline/testing/predicates.py index 3bc35346..ed1c472c 100644 --- a/zipline/testing/predicates.py +++ b/zipline/testing/predicates.py @@ -380,6 +380,12 @@ def assert_timestamp_and_datetime_equal(result, ) +def assert_isidentical(result, expected, msg=''): + assert result.isidentical(expected), ( + '%s%s is not identical to %s' % (_fmt_msg(msg), result, expected) + ) + + try: # pull the dshape cases in from datashape.util.testing import assert_dshape_equal