test_filter.py

"""
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,
    array,
    eye,
    float64,
    full,
    inf,
    isfinite,
    nan,
    nanpercentile,
    ones,
    ones_like,
    putmask,
    rot90,
    sum as np_sum
)
from numpy.random import randn, seed as random_seed
import pandas as pd

from catalyst.errors import BadPercentileBounds
from catalyst.pipeline import Filter, Factor, Pipeline
from catalyst.pipeline.classifiers import Classifier
from catalyst.pipeline.factors import CustomFactor
from catalyst.pipeline.filters import (
    All,
    Any,
    AtLeastN,
    StaticAssets,
    StaticSids,
)
from catalyst.testing import parameter_space, permute_rows, CatalystTestCase
from catalyst.testing.fixtures import WithSeededRandomPipelineEngine
from catalyst.testing.predicates import assert_equal
from catalyst.utils.numpy_utils import float64_dtype, int64_dtype
from .base import BasePipelineTestCase, with_default_shape


def rowwise_rank(array, mask=None):
    """
    Take a 2D array and return the 0-indexed sorted position of each element in
    the array for each row.

    Examples
    --------
    In [5]: data
    Out[5]:
    array([[-0.141, -1.103, -1.0171,  0.7812,  0.07  ],
           [ 0.926,  0.235, -0.7698,  1.4552,  0.2061],
           [ 1.579,  0.929, -0.557 ,  0.7896, -1.6279],
           [-1.362, -2.411, -1.4604,  1.4468, -0.1885],
           [ 1.272,  1.199, -3.2312, -0.5511, -1.9794]])

    In [7]: argsort(argsort(data))
    Out[7]:
    array([[2, 0, 1, 4, 3],
           [3, 2, 0, 4, 1],
           [4, 3, 1, 2, 0],
           [2, 0, 1, 4, 3],
           [4, 3, 0, 2, 1]])
    """
    # note that unlike scipy.stats.rankdata, the output here is 0-indexed, not
    # 1-indexed.
    return argsort(argsort(array))


class SomeFactor(Factor):
    dtype = float64_dtype
    inputs = ()
    window_length = 0


class SomeOtherFactor(Factor):
    dtype = float64_dtype
    inputs = ()
    window_length = 0


class SomeClassifier(Classifier):
    dtype = int64_dtype
    inputs = ()
    window_length = 0
    missing_value = -1


class Mask(Filter):
    inputs = ()
    window_length = 0


class FilterTestCase(BasePipelineTestCase):

    def init_instance_fixtures(self):
        super(FilterTestCase, self).init_instance_fixtures()
        self.f = SomeFactor()
        self.g = SomeOtherFactor()
        self.c = SomeClassifier()

    @with_default_shape
    def randn_data(self, seed, shape):
        """
        Build a block of testing data from numpy.random.randn.
        """
        random_seed(seed)
        return randn(*shape)

    def test_bad_percentiles(self):
        f = self.f

        bad_percentiles = [
            (-.1, 10),
            (10, 100.1),
            (20, 10),
            (50, 50),
        ]
        for min_, max_ in bad_percentiles:
            with self.assertRaises(BadPercentileBounds):
                f.percentile_between(min_, max_)

    def _test_top_and_bottom(self):
        data = self.randn_data(seed=5)  # Fix a seed for determinism.

        mask_data = ones_like(data, dtype=bool)
        mask_data[:, 0] = False

        nan_data = data.copy()
        nan_data[:, 0] = nan

        mask = Mask()

        methods = ['top', 'bottom']
        counts = 2, 3, 10
        term_combos = list(product(methods, counts, [True, False]))

        def termname(method, count, masked):
            return '_'.join([method, str(count), 'mask' if masked else ''])

        def expected_result(method, count, masked):
            # Ranking with a mask is equivalent to ranking with nans applied on
            # the masked values.
            to_rank = nan_data if masked else data

            if method == 'top':
                return rowwise_rank(-to_rank) < count
            elif method == 'bottom':
                return rowwise_rank(to_rank) < count

        # Add a term for each permutation of top/bottom, count, and
        # mask/no_mask.
        terms = {}
        expected = {}
        for method, count, masked in term_combos:
            kwargs = {'N': count}
            if masked:
                kwargs['mask'] = mask
            term = getattr(self.f, method)(**kwargs)
            name = termname(method, count, masked)
            terms[name] = term
            expected[name] = expected_result(method, count, masked)

        self.check_terms(
            terms,
            expected,
            initial_workspace={self.f: data, mask: mask_data},
            mask=self.build_mask(self.ones_mask()),
        )

    def _test_percentile_between(self):

        quintiles = range(5)
        filter_names = ['pct_' + str(q) for q in quintiles]
        iter_quintiles = list(zip(filter_names, quintiles))
        terms = {
            name: self.f.percentile_between(q * 20.0, (q + 1) * 20.0)
            for name, q in iter_quintiles
        }

        # Test with 5 columns and no NaNs.
        eye5 = eye(5, dtype=float64)
        expected = {}
        for name, quintile in iter_quintiles:
            if quintile < 4:
                # There are four 0s and one 1 in each row, so the first 4
                # quintiles should be all the locations with zeros in the input
                # array.
                expected[name] = ~eye5.astype(bool)
            else:
                # The top quintile should match the sole 1 in each row.
                expected[name] = eye5.astype(bool)

        self.check_terms(
            terms=terms,
            expected=expected,
            initial_workspace={self.f: eye5},
            mask=self.build_mask(ones((5, 5))),
        )

        # Test with 6 columns, no NaNs, and one masked entry per day.
        eye6 = eye(6, dtype=float64)
        mask = array([[1, 1, 1, 1, 1, 0],
                      [0, 1, 1, 1, 1, 1],
                      [1, 0, 1, 1, 1, 1],
                      [1, 1, 0, 1, 1, 1],
                      [1, 1, 1, 0, 1, 1],
                      [1, 1, 1, 1, 0, 1]], dtype=bool)
        expected = {}
        for name, quintile in iter_quintiles:
            if quintile < 4:
                # Should keep all values that were 0 in the base data and were
                # 1 in the mask.
                expected[name] = mask & ~eye6.astype(bool)
            else:
                # The top quintile should match the sole 1 in each row.
                expected[name] = eye6.astype(bool)

        self.check_terms(
            terms=terms,
            expected=expected,
            initial_workspace={self.f: eye6},
            mask=self.build_mask(mask),
        )

        # Test with 6 columns, no mask, and one NaN per day.  Should have the
        # same outcome as if we had masked the NaNs.
        # In particular, the NaNs should never pass any filters.
        eye6_withnans = eye6.copy()
        putmask(eye6_withnans, ~mask, nan)
        expected = {}
        for name, quintile in iter_quintiles:
            if quintile < 4:
                # Should keep all values that were 0 in the base data and were
                # 1 in the mask.
                expected[name] = mask & (~eye6.astype(bool))
            else:
                # Should keep all the 1s in the base data.
                expected[name] = eye6.astype(bool)

        self.check_terms(
            terms,
            expected,
            initial_workspace={self.f: eye6},
            mask=self.build_mask(mask),
        )

    def _test_percentile_nasty_partitions(self):
        # Test percentile with nasty partitions: divide up 5 assets into
        # quartiles.
        # There isn't a nice mathematical definition of correct behavior here,
        # so for now we guarantee the behavior of numpy.nanpercentile.  This is
        # mostly for regression testing in case we write our own specialized
        # percentile calculation at some point in the future.

        data = arange(25, dtype=float).reshape(5, 5) % 4
        quartiles = range(4)
        filter_names = ['pct_' + str(q) for q in quartiles]

        terms = {
            name: self.f.percentile_between(q * 25.0, (q + 1) * 25.0)
            for name, q in zip(filter_names, quartiles)
        }

        expected = {}
        for name, quartile in zip(filter_names, quartiles):
            lower = quartile * 25.0
            upper = (quartile + 1) * 25.0
            expected[name] = and_(
                nanpercentile(data, lower, axis=1, keepdims=True) <= data,
                data <= nanpercentile(data, upper, axis=1, keepdims=True),
            )

        self.check_terms(
            terms,
            expected,
            initial_workspace={self.f: data},
            mask=self.build_mask(ones((5, 5))),
        )

    def _test_percentile_after_mask(self):
        f_input = eye(5)
        g_input = arange(25, dtype=float).reshape(5, 5)
        initial_mask = self.build_mask(ones((5, 5)))

        custom_mask = self.f < 1
        without_mask = self.g.percentile_between(80, 100)
        with_mask = self.g.percentile_between(80, 100, mask=custom_mask)

        terms = {
            'mask': custom_mask,
            'without_mask': without_mask,
            'with_mask': with_mask,
        }
        expected = {
            # Mask that accepts everything except the diagonal.
            'mask': ~eye(5, dtype=bool),
            # Second should pass the largest value each day.  Each row is
            # strictly increasing, so we always select the last value.
            'without_mask': array(
                [[0, 0, 0, 0, 1],
                 [0, 0, 0, 0, 1],
                 [0, 0, 0, 0, 1],
                 [0, 0, 0, 0, 1],
                 [0, 0, 0, 0, 1]],
                dtype=bool,
            ),
            # With a mask, we should remove the diagonal as an option before
            # computing percentiles.  On the last day, we should get the
            # second-largest value, rather than the largest.
            'with_mask': array(
                [[0, 0, 0, 0, 1],
                 [0, 0, 0, 0, 1],
                 [0, 0, 0, 0, 1],
                 [0, 0, 0, 0, 1],
                 [0, 0, 0, 1, 0]],  # Different from with!
                dtype=bool,
            ),
        }

        self.check_terms(
            terms,
            expected,
            initial_workspace={self.f: f_input, self.g: g_input},
            mask=initial_mask,
        )

    def _test_isnan(self):
        data = self.randn_data(seed=10)
        diag = eye(*data.shape, dtype=bool)
        data[diag] = nan

        self.check_terms(
            terms={
                'isnan': self.f.isnan(),
                'isnull': self.f.isnull(),
            },
            expected={
                'isnan': diag,
                'isnull': diag,
            },
            initial_workspace={self.f: data},
            mask=self.build_mask(self.ones_mask()),
        )

    def _test_notnan(self):
        data = self.randn_data(seed=10)
        diag = eye(*data.shape, dtype=bool)
        data[diag] = nan

        self.check_terms(
            terms={
                'notnan': self.f.notnan(),
                'notnull': self.f.notnull(),
            },
            expected={
                'notnan': ~diag,
                'notnull': ~diag,
            },
            initial_workspace={self.f: data},
            mask=self.build_mask(self.ones_mask()),
        )

    def _test_isfinite(self):
        data = self.randn_data(seed=10)
        data[:, 0] = nan
        data[:, 2] = inf
        data[:, 4] = -inf

        self.check_terms(
            terms={'isfinite': self.f.isfinite()},
            expected={'isfinite': isfinite(data)},
            initial_workspace={self.f: data},
            mask=self.build_mask(self.ones_mask()),
        )

    def _test_all(self):

        data = array([[1, 1, 1, 1, 1, 1],
                      [0, 1, 1, 1, 1, 1],
                      [1, 0, 1, 1, 1, 1],
                      [1, 1, 0, 1, 1, 1],
                      [1, 1, 1, 0, 1, 1],
                      [1, 1, 1, 1, 0, 1],
                      [1, 1, 1, 1, 1, 0]], dtype=bool)

        # With a window_length of N, 0's should be "sticky" for the (N - 1)
        # days after the 0 in the base data.

        # Note that, the way ``self.run_graph`` works, we compute the same
        # number of output rows for all inputs, so we only get the last 4
        # outputs for expected_3 even though we have enought input data to
        # compute 5 rows.
        expected_3 = array([[0, 0, 0, 1, 1, 1],
                            [1, 0, 0, 0, 1, 1],
                            [1, 1, 0, 0, 0, 1],
                            [1, 1, 1, 0, 0, 0]], dtype=bool)

        expected_4 = array([[0, 0, 0, 1, 1, 1],
                            [0, 0, 0, 0, 1, 1],
                            [1, 0, 0, 0, 0, 1],
                            [1, 1, 0, 0, 0, 0]], dtype=bool)

        class Input(Filter):
            inputs = ()
            window_length = 0

        self.check_terms(
            terms={
                '3': All(inputs=[Input()], window_length=3),
                '4': All(inputs=[Input()], window_length=4),
            },
            expected={
                '3': expected_3,
                '4': expected_4,
            },
            initial_workspace={Input(): data},
            mask=self.build_mask(ones(shape=data.shape)),
        )

    def _test_any(self):

        # FUN FACT: The inputs and outputs here are exactly the negation of
        # the inputs and outputs for test_all above. This isn't a coincidence.
        #
        # By de Morgan's Laws, we have::
        #
        #     ~(a & b) == (~a | ~b)
        #
        # negating both sides, we have::
        #
        #      (a & b) == ~(a | ~b)
        #
        # Since all(a, b) is isomorphic to (a & b), and any(a, b) is isomorphic
        # to (a | b), we have::
        #
        #     all(a, b) == ~(any(~a, ~b))
        #
        data = array([[0, 0, 0, 0, 0, 0],
                      [1, 0, 0, 0, 0, 0],
                      [0, 1, 0, 0, 0, 0],
                      [0, 0, 1, 0, 0, 0],
                      [0, 0, 0, 1, 0, 0],
                      [0, 0, 0, 0, 1, 0],
                      [0, 0, 0, 0, 0, 1]], dtype=bool)

        # With a window_length of N, 1's should be "sticky" for the (N - 1)
        # days after the 1 in the base data.

        # Note that, the way ``self.run_graph`` works, we compute the same
        # number of output rows for all inputs, so we only get the last 4
        # outputs for expected_3 even though we have enought input data to
        # compute 5 rows.
        expected_3 = array([[1, 1, 1, 0, 0, 0],
                            [0, 1, 1, 1, 0, 0],
                            [0, 0, 1, 1, 1, 0],
                            [0, 0, 0, 1, 1, 1]], dtype=bool)

        expected_4 = array([[1, 1, 1, 0, 0, 0],
                            [1, 1, 1, 1, 0, 0],
                            [0, 1, 1, 1, 1, 0],
                            [0, 0, 1, 1, 1, 1]], dtype=bool)

        class Input(Filter):
            inputs = ()
            window_length = 0

        self.check_terms(
            terms={
                '3': Any(inputs=[Input()], window_length=3),
                '4': Any(inputs=[Input()], window_length=4),
            },
            expected={
                '3': expected_3,
                '4': expected_4,
            },
            initial_workspace={Input(): data},
            mask=self.build_mask(ones(shape=data.shape)),
        )

    def _test_at_least_N(self):

        # With a window_length of K, AtLeastN should return 1
        # if N or more 1's exist in the lookback window

        # This smoothing filter gives customizable "stickiness"

        data = array([[1, 1, 1, 1, 1, 1],
                      [1, 1, 1, 1, 1, 1],
                      [1, 1, 1, 1, 1, 0],
                      [1, 1, 1, 1, 0, 0],
                      [1, 1, 1, 0, 0, 0],
                      [1, 1, 0, 0, 0, 0],
                      [1, 0, 0, 0, 0, 0]], dtype=bool)

        expected_1 = array([[1, 1, 1, 1, 1, 1],
                            [1, 1, 1, 1, 1, 1],
                            [1, 1, 1, 1, 1, 0],
                            [1, 1, 1, 1, 0, 0]], dtype=bool)

        expected_2 = array([[1, 1, 1, 1, 1, 1],
                            [1, 1, 1, 1, 1, 0],
                            [1, 1, 1, 1, 0, 0],
                            [1, 1, 1, 0, 0, 0]], dtype=bool)

        expected_3 = array([[1, 1, 1, 1, 1, 0],
                            [1, 1, 1, 1, 0, 0],
                            [1, 1, 1, 0, 0, 0],
                            [1, 1, 0, 0, 0, 0]], dtype=bool)

        expected_4 = array([[1, 1, 1, 1, 0, 0],
                            [1, 1, 1, 0, 0, 0],
                            [1, 1, 0, 0, 0, 0],
                            [1, 0, 0, 0, 0, 0]], dtype=bool)

        class Input(Filter):
            inputs = ()
            window_length = 0

        all_but_one = AtLeastN(inputs=[Input()],
                               window_length=4,
                               N=3)

        all_but_two = AtLeastN(inputs=[Input()],
                               window_length=4,
                               N=2)

        any_equiv = AtLeastN(inputs=[Input()],
                             window_length=4,
                             N=1)

        all_equiv = AtLeastN(inputs=[Input()],
                             window_length=4,
                             N=4)

        self.check_terms(
            terms={
                'AllButOne': all_but_one,
                'AllButTwo': all_but_two,
                'AnyEquiv': any_equiv,
                'AllEquiv': all_equiv,
                'Any': Any(inputs=[Input()], window_length=4),
                'All': All(inputs=[Input()], window_length=4)
            },
            expected={
                'Any': expected_1,
                'AnyEquiv': expected_1,
                'AllButTwo': expected_2,
                'AllButOne': expected_3,
                'All': expected_4,
                'AllEquiv': expected_4,
            },
            initial_workspace={Input(): data},
            mask=self.build_mask(ones(shape=data.shape)),
        )

    @parameter_space(factor_len=[2, 3, 4])
    def _test_window_safe(self, factor_len):
        # all true data set of (days, securities)
        data = full(self.default_shape, True, dtype=bool)

        class InputFilter(Filter):
            inputs = ()
            window_length = 0

        class TestFactor(CustomFactor):
            dtype = float64_dtype
            inputs = (InputFilter(), )
            window_length = factor_len

            def compute(self, today, assets, out, filter_):
                # sum for each column
                out[:] = np_sum(filter_, axis=0)

        n = self.default_shape[0]
        output_shape = ((n - factor_len + 1), self.default_shape[1])
        full(output_shape, factor_len, dtype=float64)

        self.check_terms(
            terms={
                'windowsafe': TestFactor(),
            },
            expected={
                'windowsafe': full(output_shape, factor_len, dtype=float64),
            },
            initial_workspace={InputFilter(): data},
            mask=self.build_mask(self.ones_mask()),
        )

    @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)))),
        )


class SidFactor(CustomFactor):
    """A factor that just returns each asset's sid."""
    inputs = ()
    window_length = 1

    def compute(self, today, sids, out):
        out[:] = sids


class SpecificAssetsTestCase(WithSeededRandomPipelineEngine,
                             CatalystTestCase):

    ASSET_FINDER_EQUITY_SIDS = tuple(range(10))

    def _check_filters(self, evens, odds, first_five, last_three):
        pipe = Pipeline(
            columns={
                'sid': SidFactor(),
                'evens': evens,
                'odds': odds,
                'first_five': first_five,
                'last_three': last_three,
            },
        )

        start, end = self.trading_days[[-10, -1]]
        results = self.run_pipeline(pipe, start, end).unstack()

        sids = results.sid.astype(int64_dtype)

        assert_equal(results.evens, ~(sids % 2).astype(bool))
        assert_equal(results.odds, (sids % 2).astype(bool))
        assert_equal(results.first_five, sids < 5)
        assert_equal(results.last_three, sids >= 7)

    def _test_specific_assets(self):
        assets = self.asset_finder.retrieve_all(self.ASSET_FINDER_EQUITY_SIDS)

        self._check_filters(
            evens=StaticAssets(assets[::2]),
            odds=StaticAssets(assets[1::2]),
            first_five=StaticAssets(assets[:5]),
            last_three=StaticAssets(assets[-3:]),
        )

    def _test_specific_sids(self):
        sids = self.ASSET_FINDER_EQUITY_SIDS

        self._check_filters(
            evens=StaticSids(sids[::2]),
            odds=StaticSids(sids[1::2]),
            first_five=StaticSids(sids[:5]),
            last_three=StaticSids(sids[-3:]),
        )


class TestPostProcessAndToWorkSpaceValue(CatalystTestCase):
    def test_reversability(self):
        class F(Filter):
            inputs = ()
            window_length = 0
            missing_value = False

        f = F()
        column_data = array(
            [[True, f.missing_value],
             [True, f.missing_value],
             [True, True]],
            dtype=bool,
        )

        assert_equal(f.postprocess(column_data.ravel()), column_data.ravel())

        # only include the non-missing data
        pipeline_output = pd.Series(
            data=True,
            index=pd.MultiIndex.from_arrays([
                [pd.Timestamp('2014-01-01'),
                 pd.Timestamp('2014-01-02'),
                 pd.Timestamp('2014-01-03'),
                 pd.Timestamp('2014-01-03')],
                [0, 0, 0, 1],
            ]),
        )

        assert_equal(
            f.to_workspace_value(pipeline_output, pd.Index([0, 1])),
            column_data,
        )