wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-13 12:00:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1276 from quantopian/blaze-loader-checkpoints

Browse Source 
ENH: add ffill checkpointing to blaze core loader
This commit is contained in:
Joe Jevnik
2016-06-21 16:48:08 -04:00
committed by GitHub
parent 65bf854897 b210acb121
commit d608e0af4f
5 changed files with 701 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/pipeline/test_blaze.py
+481 -60
View File
@@ -21,16 +21,21 @@ 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,
BlazeLoader,
NoDeltasWarning,
NoMetaDataWarning,
)
from zipline.pipeline.loaders.blaze.core import (
ExprData,
NonPipelineField,
no_deltas_rules,
)
from zipline.testing import (
ZiplineTestCase,
parameter_space,
tmp_asset_finder,
)
from zipline.testing.fixtures import WithAssetFinder
from zipline.utils.numpy_utils import (
@@ -38,7 +43,7 @@ from zipline.utils.numpy_utils import (
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')
@@ -54,6 +59,7 @@ asset_infos = (
pd.Timestamp('2015'),
),),
)
simple_asset_info = asset_infos[0][0]
with_extra_sid = parameterized.expand(asset_infos)
with_ignore_sid = parameterized.expand(
product(chain.from_iterable(asset_infos), [True, False])
@@ -106,13 +112,21 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
cls.garbage_loader = BlazeLoader()
cls.missing_values = {'int_value': 0}
cls.value_dshape = dshape("""var * {
sid: ?int64,
value: float64,
asof_date: datetime,
timestamp: datetime,
}""")
def test_tabular(self):
name = 'expr'
expr = bz.data(self.df, name=name, dshape=self.dshape)
ds = from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
self.assertEqual(ds.__name__, name)
@@ -129,7 +143,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
),
ds,
@@ -141,7 +156,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
value = from_blaze(
expr.value,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
self.assertEqual(value.name, 'value')
@@ -153,7 +169,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr.value,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
),
value,
@@ -162,7 +179,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
).value,
value,
@@ -173,7 +191,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
),
value.dataset,
@@ -184,9 +203,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
expr = bz.data(
self.df.loc[:, ['sid', 'value', 'timestamp']],
name='expr',
dshape="""
var * {
sid: ?int64,
dshape="""var * {
sid: int64,
value: float64,
timestamp: datetime,
}""",
@@ -196,32 +214,150 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
)
self.assertIn("'asof_date'", str(e.exception))
self.assertIn(repr(str(expr.dshape.measure)), str(e.exception))
def test_auto_deltas(self):
def test_missing_timestamp(self):
expr = bz.data(
{'ds': self.df,
'ds_deltas': pd.DataFrame(columns=self.df.columns)},
dshape=var * Record((
('ds', self.dshape.measure),
('ds_deltas', self.dshape.measure),
)),
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'))
m = {'ds': self.df}
if deltas:
m['ds_deltas'] = pd.DataFrame(columns=self.df.columns),
if checkpoints:
m['ds_checkpoints'] = pd.DataFrame(columns=self.df.columns),
expr = bz.data(
m,
dshape=var * Record((k, self.dshape.measure) for k in m),
)
loader = BlazeLoader()
ds = from_blaze(
expr.ds,
loader=loader,
missing_values=self.missing_values,
no_deltas_rule=select_level(deltas),
no_checkpoints_rule=select_level(checkpoints),
)
self.assertEqual(len(loader), 1)
exprdata = loader[ds]
self.assertTrue(exprdata.expr.isidentical(expr.ds))
self.assertTrue(exprdata.deltas.isidentical(expr.ds_deltas))
if deltas:
self.assertTrue(exprdata.deltas.isidentical(expr.ds_deltas))
else:
self.assertIsNone(exprdata.deltas)
if checkpoints:
self.assertTrue(
exprdata.checkpoints.isidentical(expr.ds_checkpoints),
)
else:
self.assertIsNone(exprdata.checkpoints)
def test_auto_deltas_fail_warn(self):
@parameter_space(deltas={True, False}, checkpoints={True, False})
def test_auto_metadata_fail_warn(self, deltas, checkpoints):
select_level = op.getitem(('ignore', 'warn'))
with warnings.catch_warnings(record=True) as ws:
warnings.simplefilter('always')
loader = BlazeLoader()
@@ -229,22 +365,31 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.warn,
no_deltas_rule=select_level(deltas),
no_checkpoints_rule=select_level(checkpoints),
missing_values=self.missing_values,
)
self.assertEqual(len(ws), 1)
w = ws[0].message
self.assertIsInstance(w, NoDeltasWarning)
self.assertIn(str(expr), str(w))
self.assertEqual(len(ws), deltas + checkpoints)
def test_auto_deltas_fail_raise(self):
for w in ws:
w = w.message
self.assertIsInstance(w, NoMetaDataWarning)
self.assertIn(str(expr), str(w))
@parameter_space(deltas={True, False}, checkpoints={True, False})
def test_auto_metadata_fail_raise(self, deltas, checkpoints):
if not (deltas or checkpoints):
# not a real case
return
select_level = op.getitem(('ignore', 'raise'))
loader = BlazeLoader()
expr = bz.data(self.df, dshape=self.dshape)
with self.assertRaises(ValueError) as e:
from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.raise_,
no_deltas_rule=select_level(deltas),
no_checkpoints_rule=select_level(checkpoints),
)
self.assertIn(str(expr), str(e.exception))
@@ -261,7 +406,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
)
with self.assertRaises(AttributeError):
ds.a
@@ -540,45 +686,71 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
)
def test_complex_expr(self):
expr = bz.data(self.df, dshape=self.dshape)
expr = bz.data(self.df, dshape=self.dshape, name='expr')
# put an Add in the table
expr_with_add = bz.transform(expr, value=expr.value + 1)
# Test that we can have complex expressions with no deltas
# test that we can have complex expressions with no metadata
from_blaze(
expr_with_add,
deltas=None,
checkpoints=None,
loader=self.garbage_loader,
missing_values=self.missing_values,
no_checkpoints_rule='ignore',
)
with self.assertRaises(TypeError):
with self.assertRaises(TypeError) as e:
# test that we cannot create a single column from a non field
from_blaze(
expr.value + 1, # put an Add in the column
deltas=None,
checkpoints=None,
loader=self.garbage_loader,
missing_values=self.missing_values,
no_checkpoints_rule='ignore',
)
assert_equal(
str(e.exception),
"expression 'expr.value + 1' was array-like but not a simple field"
" of some larger table",
)
deltas = bz.data(
pd.DataFrame(columns=self.df.columns),
dshape=self.dshape,
name='deltas',
)
checkpoints = bz.data(
pd.DataFrame(columns=self.df.columns),
dshape=self.dshape,
name='checkpoints',
)
with self.assertRaises(TypeError):
from_blaze(
expr_with_add,
deltas=deltas,
loader=self.garbage_loader,
missing_values=self.missing_values,
)
with self.assertRaises(TypeError):
# test that we can have complex expressions with explicit metadata
from_blaze(
expr_with_add,
deltas=deltas,
checkpoints=checkpoints,
loader=self.garbage_loader,
missing_values=self.missing_values,
)
with self.assertRaises(TypeError) as e:
# test that we cannot create a single column from a non field
# even with explicit metadata
from_blaze(
expr.value + 1,
deltas=deltas,
checkpoints=checkpoints,
loader=self.garbage_loader,
missing_values=self.missing_values,
)
assert_equal(
str(e.exception),
"expression 'expr.value + 1' was array-like but not a simple field"
" of some larger table",
)
def _test_id(self, df, dshape, expected, finder, add):
expr = bz.data(df, name='expr', dshape=dshape)
@@ -586,7 +758,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
p = Pipeline()
@@ -617,7 +790,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
p = Pipeline()
@@ -809,13 +983,12 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
Equity(66 [B]) 2
Equity(67 [C]) 2
"""
asset_info = asset_infos[0][0]
nassets = len(asset_info)
nassets = len(simple_asset_info)
expected = pd.DataFrame(
list(concatv([0] * nassets, [1] * nassets, [2] * nassets)),
index=pd.MultiIndex.from_product((
self.macro_df.timestamp,
self.asset_finder.retrieve_all(asset_info.index),
self.asset_finder.retrieve_all(simple_asset_info.index),
)),
columns=('value',),
)
@@ -907,15 +1080,14 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
fields = OrderedDict(self.macro_dshape.measure.fields)
fields['other'] = fields['value']
asset_info = asset_infos[0][0]
with tmp_asset_finder(equities=asset_info) as finder:
with tmp_asset_finder(equities=simple_asset_info) as finder:
expected = pd.DataFrame(
np.array([[0, 1],
[1, 2],
[2, 3]]).repeat(3, axis=0),
index=pd.MultiIndex.from_product((
df.timestamp,
finder.retrieve_all(asset_info.index),
finder.retrieve_all(simple_asset_info.index),
)),
columns=('value', 'other'),
).sort_index(axis=1)
@@ -1044,6 +1216,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
def _run_pipeline(self,
expr,
deltas,
checkpoints,
expected_views,
expected_output,
finder,
@@ -1056,8 +1229,10 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
deltas,
checkpoints,
loader=loader,
no_deltas_rule=no_deltas_rules.raise_,
no_deltas_rule='raise',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
p = Pipeline()
@@ -1070,7 +1245,11 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
window_length = window_length_
def compute(self, today, assets, out, data):
assert_array_almost_equal(data, expected_views[today])
assert_array_almost_equal(
data,
expected_views[today],
err_msg=str(today),
)
out[:] = compute_fn(data)
p.add(TestFactor(), 'value')
@@ -1142,6 +1321,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1194,6 +1374,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1205,7 +1386,6 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
)
def test_deltas_macro(self):
asset_info = asset_infos[0][0]
expr = bz.data(self.macro_df, name='expr', dshape=self.macro_dshape)
deltas = bz.data(
self.macro_df.iloc[:-1],
@@ -1218,18 +1398,18 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
timestamp=deltas.timestamp + timedelta(days=1),
)
nassets = len(asset_info)
nassets = len(simple_asset_info)
expected_views = keymap(pd.Timestamp, {
'2014-01-02': repeat_last_axis(np.array([10.0, 1.0]), nassets),
'2014-01-03': repeat_last_axis(np.array([11.0, 2.0]), nassets),
})
with tmp_asset_finder(equities=asset_info) as finder:
with tmp_asset_finder(equities=simple_asset_info) as finder:
expected_output = pd.DataFrame(
list(concatv([10] * nassets, [11] * nassets)),
index=pd.MultiIndex.from_product((
sorted(expected_views.keys()),
finder.retrieve_all(asset_info.index),
finder.retrieve_all(simple_asset_info.index),
)),
columns=('value',),
)
@@ -1237,6 +1417,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1311,6 +1492,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1322,7 +1504,6 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
)
def test_novel_deltas_macro(self):
asset_info = asset_infos[0][0]
base_dates = pd.DatetimeIndex([
pd.Timestamp('2014-01-01'),
pd.Timestamp('2014-01-04')
@@ -1340,7 +1521,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
timestamp=deltas.timestamp + timedelta(days=1),
)
nassets = len(asset_info)
nassets = len(simple_asset_info)
expected_views = keymap(pd.Timestamp, {
'2014-01-03': repeat_last_axis(
np.array([10.0, 10.0, 10.0]),
@@ -1359,18 +1540,19 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
# omitting the 4th and 5th to simulate a weekend
pd.Timestamp('2014-01-06'),
])
with tmp_asset_finder(equities=asset_info) as finder:
with tmp_asset_finder(equities=simple_asset_info) as finder:
expected_output = pd.DataFrame(
list(concatv([10] * nassets, [11] * nassets)),
index=pd.MultiIndex.from_product((
sorted(expected_views.keys()),
finder.retrieve_all(asset_info.index),
finder.retrieve_all(simple_asset_info.index),
)),
columns=('value',),
)
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1380,3 +1562,242 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
window_length=3,
compute_fn=op.itemgetter(-1),
)
def _test_checkpoints_macro(self, checkpoints, ffilled_value=-1.0):
"""Simple checkpoints test that accepts a checkpoints dataframe and
the expected value for 2014-01-03 for macro datasets.
The underlying data has value -1.0 on 2014-01-01 and 1.0 on 2014-01-04.
Parameters
----------
checkpoints : pd.DataFrame
The checkpoints data.
ffilled_value : float, optional
The value to be read on the third, if not provided, it will be the
value in the base data that will be naturally ffilled there.
"""
dates = pd.Timestamp('2014-01-01'), pd.Timestamp('2014-01-04')
baseline = pd.DataFrame({
'value': [-1.0, 1.0],
'asof_date': dates,
'timestamp': dates,
})
nassets = len(simple_asset_info)
expected_views = keymap(pd.Timestamp, {
'2014-01-03': repeat_last_axis(
np.array([ffilled_value]),
nassets,
),
'2014-01-04': repeat_last_axis(
np.array([1.0]),
nassets,
),
})
with tmp_asset_finder(equities=simple_asset_info) as finder:
expected_output = pd.DataFrame(
list(concatv([ffilled_value] * nassets, [1.0] * nassets)),
index=pd.MultiIndex.from_product((
sorted(expected_views.keys()),
finder.retrieve_all(simple_asset_info.index),
)),
columns=('value',),
)
self._run_pipeline(
bz.data(baseline, name='expr', dshape=self.macro_dshape),
None,
bz.data(
checkpoints,
name='expr_checkpoints',
dshape=self.macro_dshape,
),
expected_views,
expected_output,
finder,
calendar=pd.date_range('2014-01-01', '2014-01-04'),
start=pd.Timestamp('2014-01-03'),
end=dates[-1],
window_length=1,
compute_fn=op.itemgetter(-1),
)
def test_checkpoints_macro(self):
ffilled_value = 0.0
checkpoints_ts = pd.Timestamp('2014-01-02')
checkpoints = pd.DataFrame({
'value': [ffilled_value],
'asof_date': checkpoints_ts,
'timestamp': checkpoints_ts,
})
self._test_checkpoints_macro(checkpoints, ffilled_value)
def test_empty_checkpoints_macro(self):
empty_checkpoints = pd.DataFrame({
'value': [],
'asof_date': [],
'timestamp': [],
})
self._test_checkpoints_macro(empty_checkpoints)
def test_checkpoints_out_of_bounds_macro(self):
# provide two checkpoints, one before the data in the base table
# and one after, these should not affect the value on the third
dates = pd.to_datetime(['2013-12-31', '2014-01-05'])
checkpoints = pd.DataFrame({
'value': [-2, 2],
'asof_date': dates,
'timestamp': dates,
})
self._test_checkpoints_macro(checkpoints)
def _test_checkpoints(self, checkpoints, ffilled_values=None):
"""Simple checkpoints test that accepts a checkpoints dataframe and
the expected value for 2014-01-03.
The underlying data has value -1.0 on 2014-01-01 and 1.0 on 2014-01-04.
Parameters
----------
checkpoints : pd.DataFrame
The checkpoints data.
ffilled_value : float, optional
The value to be read on the third, if not provided, it will be the
value in the base data that will be naturally ffilled there.
"""
nassets = len(simple_asset_info)
dates = pd.to_datetime(['2014-01-01', '2014-01-04'])
dates_repeated = np.tile(dates, nassets)
values = np.arange(nassets) + 1
values = np.hstack((values[::-1], values))
baseline = pd.DataFrame({
'sid': np.tile(simple_asset_info.index, 2),
'value': values,
'asof_date': dates_repeated,
'timestamp': dates_repeated,
})
if ffilled_values is None:
ffilled_values = baseline.value.iloc[:nassets]
updated_values = baseline.value.iloc[nassets:]
expected_views = keymap(pd.Timestamp, {
'2014-01-03': [ffilled_values],
'2014-01-04': [updated_values],
})
with tmp_asset_finder(equities=simple_asset_info) as finder:
expected_output = pd.DataFrame(
list(concatv(ffilled_values, updated_values)),
index=pd.MultiIndex.from_product((
sorted(expected_views.keys()),
finder.retrieve_all(simple_asset_info.index),
)),
columns=('value',),
)
self._run_pipeline(
bz.data(baseline, name='expr', dshape=self.value_dshape),
None,
bz.data(
checkpoints,
name='expr_checkpoints',
dshape=self.value_dshape,
),
expected_views,
expected_output,
finder,
calendar=pd.date_range('2014-01-01', '2014-01-04'),
start=pd.Timestamp('2014-01-03'),
end=dates[-1],
window_length=1,
compute_fn=op.itemgetter(-1),
)
def test_checkpoints(self):
nassets = len(simple_asset_info)
ffilled_values = (np.arange(nassets, dtype=np.float64) + 1) * 10
dates = [pd.Timestamp('2014-01-02')] * nassets
checkpoints = pd.DataFrame({
'sid': simple_asset_info.index,
'value': ffilled_values,
'asof_date': dates,
'timestamp': dates,
})
self._test_checkpoints(checkpoints, ffilled_values)
def test_empty_checkpoints(self):
checkpoints = pd.DataFrame({
'sid': [],
'value': [],
'asof_date': [],
'timestamp': [],
})
self._test_checkpoints(checkpoints)
def test_checkpoints_out_of_bounds(self):
nassets = len(simple_asset_info)
# provide two sets of checkpoints, one before the data in the base
# table and one after, these should not affect the value on the third
dates = pd.to_datetime(['2013-12-31', '2014-01-05'])
dates_repeated = np.tile(dates, nassets)
ffilled_values = (np.arange(nassets) + 2) * 10
ffilled_values = np.hstack((ffilled_values[::-1], ffilled_values))
checkpoints = pd.DataFrame({
'sid': np.tile(simple_asset_info.index, 2),
'value': ffilled_values,
'asof_date': dates_repeated,
'timestamp': dates_repeated,
})
self._test_checkpoints(checkpoints)
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()), '<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')