From 49bb8264dce54c92f06df991c227d3b6f8d49b18 Mon Sep 17 00:00:00 2001 From: Scott Sanderson Date: Tue, 26 Jul 2016 01:56:50 -0400 Subject: [PATCH] ENH: Finish adding groupby to rank/top/bottom. - Added test coverage for grouped and masked top/bottom. - Added test coverage for grouped rank on datetime factors. - Fixed an issue where grouped rank would fail on datetime inputs because unary-negative isn't defined for datetimes. We now instead directly invoke a function from rank.pyx that does the normalizations as neeeded. - Fixed an issue where GroupedRowTransform assumed that it produced the same dtype as its input. This isn't true for rank() of a datetime-dtype factor. GroupedRowTransform now takes a required dtype parameter. - Similarly, fixed an issue where GroupedRowTransform assumed that its missing_value was the same as its parent's, which isn't true for rank() of a datetime-dtype factor. GroupedRowTransform now takes a required dtype parameter. - Fixed an issue where Factor.demean() and Factor.zscore() weren't properly cached because their static_identity included a closure that was dynamically generated on each invocation. They both now always use a function defined at module scope. --- tests/pipeline/test_factor.py | 11 +- tests/pipeline/test_filter.py | 254 ++++++++++++++++++++++++++++- tests/pipeline/test_term.py | 11 ++ zipline/lib/normalize.py | 10 +- zipline/lib/rank.pyx | 7 + zipline/pipeline/factors/factor.py | 83 ++++++---- 6 files changed, 332 insertions(+), 44 deletions(-) diff --git a/tests/pipeline/test_factor.py b/tests/pipeline/test_factor.py index 5de7a6be..4c63429d 100644 --- a/tests/pipeline/test_factor.py +++ b/tests/pipeline/test_factor.py @@ -336,7 +336,8 @@ class FactorTestCase(BasePipelineTestCase): for method in results: check_arrays(expected[method], results[method]) - def test_grouped_rank_ascending(self, factor_dtype=float64_dtype): + @for_each_factor_dtype + def test_grouped_rank_ascending(self, name, factor_dtype=float64_dtype): f = F(dtype=factor_dtype) c = C() @@ -439,7 +440,8 @@ class FactorTestCase(BasePipelineTestCase): check({'ordinal': f.rank(groupby=c, ascending=True)}) check({'ordinal': f.rank(groupby=str_c, ascending=True)}) - def test_grouped_rank_descending(self, factor_dtype=float64_dtype): + @for_each_factor_dtype + def test_grouped_rank_descending(self, name, factor_dtype): f = F(dtype=factor_dtype) c = C() @@ -532,11 +534,6 @@ class FactorTestCase(BasePipelineTestCase): check({'ordinal': f.rank(groupby=c, ascending=False)}) check({'ordinal': f.rank(groupby=str_c, ascending=False)}) - # TODO finish this - # @for_each_factor_dtype - # def test_grouped_rank_after_mask(self, name, factor_dtype): - # pass - @parameterized.expand([ # Test cases computed by doing: # from numpy.random import seed, randn diff --git a/tests/pipeline/test_filter.py b/tests/pipeline/test_filter.py index c403e909..ccf6211f 100644 --- a/tests/pipeline/test_filter.py +++ b/tests/pipeline/test_filter.py @@ -1,9 +1,11 @@ """ Tests for filter terms. """ +from functools import partial from itertools import product from operator import and_ +from toolz import compose from numpy import ( arange, argsort, @@ -19,15 +21,17 @@ from numpy import ( ones, ones_like, putmask, + rot90, sum as np_sum ) from numpy.random import randn, seed as random_seed from zipline.errors import BadPercentileBounds from zipline.pipeline import Filter, Factor, TermGraph +from zipline.pipeline.classifiers import Classifier from zipline.pipeline.factors import CustomFactor -from zipline.testing import check_arrays, parameter_space -from zipline.utils.numpy_utils import float64_dtype +from zipline.testing import check_arrays, parameter_space, permute_rows +from zipline.utils.numpy_utils import float64_dtype, int64_dtype from .base import BasePipelineTestCase, with_default_shape @@ -71,6 +75,13 @@ class SomeOtherFactor(Factor): window_length = 0 +class SomeClassifier(Classifier): + dtype = int64_dtype + inputs = () + window_length = 0 + missing_value = -1 + + class Mask(Filter): inputs = () window_length = 0 @@ -82,6 +93,7 @@ class FilterTestCase(BasePipelineTestCase): super(FilterTestCase, self).init_instance_fixtures() self.f = SomeFactor() self.g = SomeOtherFactor() + self.c = SomeClassifier() @with_default_shape def randn_data(self, seed, shape): @@ -415,3 +427,241 @@ class FilterTestCase(BasePipelineTestCase): results['windowsafe'], full(output_shape, factor_len, dtype=float64) ) + + @parameter_space( + dtype=('float64', 'datetime64[ns]'), + seed=(1, 2, 3), + __fail_fast=True + ) + def test_top_with_groupby(self, dtype, seed): + permute = partial(permute_rows, seed) + permuted_array = compose(permute, partial(array, dtype=int64_dtype)) + + shape = (8, 8) + + # Shuffle the input rows to verify that we correctly pick out the top + # values independently of order. + factor_data = permute(arange(0, 64, dtype=dtype).reshape(shape)) + + classifier_data = permuted_array([[0, 0, 1, 1, 2, 2, 0, 0], + [0, 0, 1, 1, 2, 2, 0, 0], + [0, 1, 2, 3, 0, 1, 2, 3], + [0, 1, 2, 3, 0, 1, 2, 3], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 0]]) + f = self.f + c = self.c + self.check_terms( + terms={ + '1': f.top(1, groupby=c), + '2': f.top(2, groupby=c), + '3': f.top(3, groupby=c), + }, + initial_workspace={ + f: factor_data, + c: classifier_data, + }, + expected={ + # Should be the rightmost location of each entry in + # classifier_data. + '1': permuted_array([[0, 0, 0, 1, 0, 1, 0, 1], + [0, 0, 0, 1, 0, 1, 0, 1], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 0, 1], + [0, 0, 0, 1, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 1]], dtype=bool), + # Should be the first and second-rightmost location of each + # entry in classifier_data. + '2': permuted_array([[0, 0, 1, 1, 1, 1, 1, 1], + [0, 0, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 1, 1, 0, 0, 1, 1], + [0, 0, 1, 1, 0, 0, 1, 1], + [0, 0, 0, 0, 0, 0, 1, 1], + [0, 0, 0, 0, 0, 0, 1, 1]], dtype=bool), + # Should be the first, second, and third-rightmost location of + # each entry in classifier_data. + '3': permuted_array([[0, 1, 1, 1, 1, 1, 1, 1], + [0, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [0, 1, 1, 1, 0, 1, 1, 1], + [0, 1, 1, 1, 0, 1, 1, 1], + [0, 0, 0, 0, 0, 1, 1, 1], + [0, 0, 0, 0, 0, 1, 1, 1]], dtype=bool), + }, + mask=self.build_mask(self.ones_mask(shape=shape)), + ) + + @parameter_space( + dtype=('float64', 'datetime64[ns]'), + seed=(1, 2, 3), + __fail_fast=True + ) + def test_top_and_bottom_with_groupby(self, dtype, seed): + permute = partial(permute_rows, seed) + permuted_array = compose(permute, partial(array, dtype=int64_dtype)) + + shape = (8, 8) + + # Shuffle the input rows to verify that we correctly pick out the top + # values independently of order. + factor_data = permute(arange(0, 64, dtype=dtype).reshape(shape)) + classifier_data = permuted_array([[0, 0, 1, 1, 2, 2, 0, 0], + [0, 0, 1, 1, 2, 2, 0, 0], + [0, 1, 2, 3, 0, 1, 2, 3], + [0, 1, 2, 3, 0, 1, 2, 3], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 0]]) + + f = self.f + c = self.c + + self.check_terms( + terms={ + 'top1': f.top(1, groupby=c), + 'top2': f.top(2, groupby=c), + 'top3': f.top(3, groupby=c), + 'bottom1': f.bottom(1, groupby=c), + 'bottom2': f.bottom(2, groupby=c), + 'bottom3': f.bottom(3, groupby=c), + }, + initial_workspace={ + f: factor_data, + c: classifier_data, + }, + expected={ + # Should be the rightmost location of each entry in + # classifier_data. + 'top1': permuted_array([[0, 0, 0, 1, 0, 1, 0, 1], + [0, 0, 0, 1, 0, 1, 0, 1], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 0, 1], + [0, 0, 0, 1, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 1]], dtype=bool), + # Should be the leftmost location of each entry in + # classifier_data. + 'bottom1': permuted_array([[1, 0, 1, 0, 1, 0, 0, 0], + [1, 0, 1, 0, 1, 0, 0, 0], + [1, 1, 1, 1, 0, 0, 0, 0], + [1, 1, 1, 1, 0, 0, 0, 0], + [1, 0, 0, 0, 1, 0, 0, 0], + [1, 0, 0, 0, 1, 0, 0, 0], + [1, 0, 0, 0, 0, 0, 0, 0], + [1, 0, 0, 0, 0, 0, 0, 0]], + dtype=bool), + # Should be the first and second-rightmost location of each + # entry in classifier_data. + 'top2': permuted_array([[0, 0, 1, 1, 1, 1, 1, 1], + [0, 0, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 1, 1, 0, 0, 1, 1], + [0, 0, 1, 1, 0, 0, 1, 1], + [0, 0, 0, 0, 0, 0, 1, 1], + [0, 0, 0, 0, 0, 0, 1, 1]], dtype=bool), + # Should be the first and second leftmost location of each + # entry in classifier_data. + 'bottom2': permuted_array([[1, 1, 1, 1, 1, 1, 0, 0], + [1, 1, 1, 1, 1, 1, 0, 0], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 0, 0, 1, 1, 0, 0], + [1, 1, 0, 0, 1, 1, 0, 0], + [1, 1, 0, 0, 0, 0, 0, 0], + [1, 1, 0, 0, 0, 0, 0, 0]], + dtype=bool), + # Should be the first, second, and third-rightmost location of + # each entry in classifier_data. + 'top3': permuted_array([[0, 1, 1, 1, 1, 1, 1, 1], + [0, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [0, 1, 1, 1, 0, 1, 1, 1], + [0, 1, 1, 1, 0, 1, 1, 1], + [0, 0, 0, 0, 0, 1, 1, 1], + [0, 0, 0, 0, 0, 1, 1, 1]], dtype=bool), + # Should be the first, second, and third-leftmost location of + # each entry in classifier_data. + 'bottom3': permuted_array([[1, 1, 1, 1, 1, 1, 1, 0], + [1, 1, 1, 1, 1, 1, 1, 0], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [1, 1, 1, 0, 1, 1, 1, 0], + [1, 1, 1, 0, 1, 1, 1, 0], + [1, 1, 1, 0, 0, 0, 0, 0], + [1, 1, 1, 0, 0, 0, 0, 0]], + dtype=bool), + }, + mask=self.build_mask(self.ones_mask(shape=shape)), + ) + + @parameter_space( + dtype=('float64', 'datetime64[ns]'), + seed=(1, 2, 3), + __fail_fast=True, + ) + def test_top_and_bottom_with_groupby_and_mask(self, dtype, seed): + permute = partial(permute_rows, seed) + permuted_array = compose(permute, partial(array, dtype=int64_dtype)) + + shape = (8, 8) + + # Shuffle the input rows to verify that we correctly pick out the top + # values independently of order. + factor_data = permute(arange(0, 64, dtype=dtype).reshape(shape)) + classifier_data = permuted_array([[0, 0, 1, 1, 2, 2, 0, 0], + [0, 0, 1, 1, 2, 2, 0, 0], + [0, 1, 2, 3, 0, 1, 2, 3], + [0, 1, 2, 3, 0, 1, 2, 3], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 1, 1, 1, 1], + [0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 0]]) + + f = self.f + c = self.c + + self.check_terms( + terms={ + 'top2': f.top(2, groupby=c), + 'bottom2': f.bottom(2, groupby=c), + }, + initial_workspace={ + f: factor_data, + c: classifier_data, + }, + expected={ + # Should be the rightmost two entries in classifier_data, + # ignoring the off-diagonal. + 'top2': permuted_array([[0, 1, 1, 1, 1, 1, 1, 0], + [0, 1, 1, 1, 1, 1, 0, 1], + [1, 1, 1, 1, 1, 0, 1, 1], + [1, 1, 1, 1, 0, 1, 1, 1], + [0, 1, 1, 0, 0, 0, 1, 1], + [0, 1, 0, 1, 0, 0, 1, 1], + [0, 0, 0, 0, 0, 0, 1, 1], + [0, 0, 0, 0, 0, 0, 1, 1]], dtype=bool), + # Should be the rightmost two entries in classifier_data, + # ignoring the off-diagonal. + 'bottom2': permuted_array([[1, 1, 1, 1, 1, 1, 0, 0], + [1, 1, 1, 1, 1, 1, 0, 0], + [1, 1, 1, 1, 1, 0, 1, 1], + [1, 1, 1, 1, 0, 1, 1, 1], + [1, 1, 0, 0, 1, 1, 0, 0], + [1, 1, 0, 0, 1, 1, 0, 0], + [1, 0, 1, 0, 0, 0, 0, 0], + [0, 1, 1, 0, 0, 0, 0, 0]], + dtype=bool), + }, + mask=self.build_mask(permute(rot90(self.eye_mask(shape=shape)))), + ) diff --git a/tests/pipeline/test_term.py b/tests/pipeline/test_term.py index 0cda474d..386f14e4 100644 --- a/tests/pipeline/test_term.py +++ b/tests/pipeline/test_term.py @@ -400,6 +400,17 @@ class ObjectIdentityTestCase(TestCase): method = getattr(f, funcname) self.assertIs(method(), method()) + def test_instance_caching_grouped_transforms(self): + f = SomeFactor() + c = GenericClassifier() + m = GenericFilter() + + for meth in f.demean, f.zscore, f.rank: + self.assertIs(meth(), meth()) + self.assertIs(meth(groupby=c), meth(groupby=c)) + self.assertIs(meth(mask=m), meth(mask=m)) + self.assertIs(meth(groupby=c, mask=m), meth(groupby=c, mask=m)) + class SomeFactorParameterized(SomeFactor): params = ('a', 'b') diff --git a/zipline/lib/normalize.py b/zipline/lib/normalize.py index 72ac7ec6..c4d351bc 100644 --- a/zipline/lib/normalize.py +++ b/zipline/lib/normalize.py @@ -1,7 +1,11 @@ import numpy as np -def naive_grouped_rowwise_apply(data, group_labels, func, out=None): +def naive_grouped_rowwise_apply(data, + group_labels, + func, + func_args=(), + out=None): """ Simple implementation of grouped row-wise function application. @@ -14,6 +18,8 @@ def naive_grouped_rowwise_apply(data, group_labels, func, out=None): Should be the same shape as array. func : function[ndarray[ndim=1]] -> function[ndarray[ndim=1]] Function to apply to pieces of each row in array. + func_args : tuple + Additional positional arguments to provide to each row in array. out : ndarray, optional Array into which to write output. If not supplied, a new array of the same shape as ``data`` is allocated and returned. @@ -41,5 +47,5 @@ def naive_grouped_rowwise_apply(data, group_labels, func, out=None): for (row, label_row, out_row) in zip(data, group_labels, out): for label in np.unique(label_row): locs = (label_row == label) - out_row[locs] = func(row[locs]) + out_row[locs] = func(row[locs], *func_args) return out diff --git a/zipline/lib/rank.pyx b/zipline/lib/rank.pyx index 3f529f4d..a23fa231 100644 --- a/zipline/lib/rank.pyx +++ b/zipline/lib/rank.pyx @@ -37,6 +37,13 @@ cpdef is_missing(ndarray data, object missing_value): return (data == missing_value) +def rankdata_1d_descending(ndarray data, str method): + """ + 1D descending version of scipy.stats.rankdata. + """ + return rankdata(-(data.view(float64)), method=method) + + def masked_rankdata_2d(ndarray data, ndarray mask, object missing_value, diff --git a/zipline/pipeline/factors/factor.py b/zipline/pipeline/factors/factor.py index cfcc0a5d..c466577d 100644 --- a/zipline/pipeline/factors/factor.py +++ b/zipline/pipeline/factors/factor.py @@ -5,12 +5,12 @@ from functools import wraps from operator import attrgetter from numbers import Number -from numpy import inf, where +from numpy import empty_like, inf, nan, where from scipy.stats import rankdata from zipline.errors import UnknownRankMethod from zipline.lib.normalize import naive_grouped_rowwise_apply -from zipline.lib.rank import masked_rankdata_2d +from zipline.lib.rank import masked_rankdata_2d, rankdata_1d_descending from zipline.pipeline.api_utils import restrict_to_dtype from zipline.pipeline.classifiers import Classifier, Everything, Quantiles from zipline.pipeline.expression import ( @@ -314,7 +314,6 @@ float64_only = restrict_to_dtype( ) ) - FACTOR_DTYPES = frozenset([datetime64ns_dtype, float64_dtype, int64_dtype]) @@ -501,16 +500,14 @@ class Factor(RestrictedDTypeMixin, ComputableTerm): -------- :meth:`pandas.DataFrame.groupby` """ - # This is a named function so that it has a __name__ for use in the - # graph repr of GroupedRowTransform. - def demean(row): - return row - nanmean(row) - return GroupedRowTransform( transform=demean, + transform_args=(), factor=self, - mask=mask, groupby=groupby, + dtype=self.dtype, + missing_value=self.missing_value, + mask=mask, ) @expect_types( @@ -569,17 +566,14 @@ class Factor(RestrictedDTypeMixin, ComputableTerm): -------- :meth:`pandas.DataFrame.groupby` """ - # This is a named function so that it has a __name__ for use in the - # graph repr of GroupedRowTransform. - def zscore(row): - return (row - nanmean(row)) / nanstd(row) - return GroupedRowTransform( transform=zscore, + transform_args=(), factor=self, - mask=mask, groupby=groupby, - window_safe=True, + dtype=self.dtype, + missing_value=self.missing_value, + mask=mask, ) def rank(self, @@ -631,17 +625,16 @@ class Factor(RestrictedDTypeMixin, ComputableTerm): if groupby is NotSpecified: return Rank(self, method=method, ascending=ascending, mask=mask) - else: - def rank(row): - return rankdata(row if ascending else -row, method=method) - - return GroupedRowTransform( - transform=rank, - factor=self, - mask=mask, - groupby=groupby, - window_safe=True, - ) + return GroupedRowTransform( + transform=rankdata if ascending else rankdata_1d_descending, + transform_args=(method,), + factor=self, + groupby=groupby, + dtype=float64_dtype, + missing_value=nan, + mask=mask, + window_safe=True, + ) @expect_types( target=Term, correlation_length=int, mask=(Filter, NotSpecifiedType), @@ -1113,6 +1106,8 @@ class GroupedRowTransform(Factor): groupby : zipline.pipeline.Classifier Classifier partitioning ``factor`` into groups to use when calculating means. + transform_args : tuple[hashable] + Additional positional arguments to forward to ``transform``. Notes ----- @@ -1128,7 +1123,15 @@ class GroupedRowTransform(Factor): """ window_length = 0 - def __new__(cls, transform, factor, mask, groupby, **kwargs): + def __new__(cls, + transform, + transform_args, + factor, + groupby, + dtype, + missing_value, + mask, + **kwargs): if mask is NotSpecified: mask = factor.mask @@ -1141,22 +1144,25 @@ class GroupedRowTransform(Factor): return super(GroupedRowTransform, cls).__new__( GroupedRowTransform, transform=transform, + transform_args=transform_args, inputs=(factor, groupby), - missing_value=factor.missing_value, + missing_value=missing_value, mask=mask, - dtype=factor.dtype, + dtype=dtype, **kwargs ) - def _init(self, transform, *args, **kwargs): + def _init(self, transform, transform_args, *args, **kwargs): self._transform = transform + self._transform_args = transform_args return super(GroupedRowTransform, self)._init(*args, **kwargs) @classmethod - def _static_identity(cls, transform, *args, **kwargs): + def _static_identity(cls, transform, transform_args, *args, **kwargs): return ( super(GroupedRowTransform, cls)._static_identity(*args, **kwargs), transform, + transform_args, ) def _compute(self, arrays, dates, assets, mask): @@ -1178,13 +1184,14 @@ class GroupedRowTransform(Factor): # Make a copy with the null code written to masked locations. group_labels = where(mask, group_labels, null_label) - return where( group_labels != null_label, naive_grouped_rowwise_apply( data=data, group_labels=group_labels, func=self._transform, + func_args=self._transform_args, + out=empty_like(data, dtype=self.dtype), ), self.missing_value, ) @@ -1492,3 +1499,13 @@ class Latest(LatestMixin, CustomFactor): def compute(self, today, assets, out, data): out[:] = data[-1] + + +# Functions to be passed to GroupedRowTransform. These aren't defined inline +# because the transformation function is part of the instance hash key. +def demean(row): + return row - nanmean(row) + + +def zscore(row): + return (row - nanmean(row)) / nanstd(row)