'Tests for the built-in factors `RollingPearsonOfReturns` and
`RollingSpearmanOfReturns`.'

@parameter_space(returns_length=[2, 3], correlation_length=[3, 4])
def test_correlation_factors(self, returns_length, correlation_length):

    assets = self.assets
    my_asset = self.my_asset
    my_asset_column = self.my_asset_column
    dates = self.dates
    start_date = self.pipeline_start_date
    end_date = self.pipeline_end_date
    start_date_index = self.start_date_index
    end_date_index = self.end_date_index
    num_days = self.num_days
    run_pipeline = self.run_pipeline
    returns = Returns(window_length=returns_length)
    masks = (self.cascading_mask, self.alternating_mask, NotSpecified)
    expected_mask_results = (self.expected_cascading_mask_result, self.expected_alternating_mask_result, self.expected_no_mask_result)
    for (mask, expected_mask) in zip(masks, expected_mask_results):
        pearson_factor = RollingPearsonOfReturns(target=my_asset, returns_length=returns_length, correlation_length=correlation_length, mask=mask)
        spearman_factor = RollingSpearmanOfReturns(target=my_asset, returns_length=returns_length, correlation_length=correlation_length, mask=mask)
        columns = {'pearson_factor': pearson_factor, 'spearman_factor': spearman_factor}
        pipeline = Pipeline(columns=columns)
        if (mask is not NotSpecified):
            pipeline.add(mask, 'mask')
        results = run_pipeline(pipeline, start_date, end_date)
        pearson_results = results['pearson_factor'].unstack()
        spearman_results = results['spearman_factor'].unstack()
        if (mask is not NotSpecified):
            mask_results = results['mask'].unstack()
            check_arrays(mask_results.values, expected_mask)
        results = run_pipeline(Pipeline(columns={'returns': returns}), dates[(start_date_index - (correlation_length - 1))], dates[end_date_index])
        returns_results = results['returns'].unstack()
        expected_pearson_results = full_like(pearson_results, nan)
        expected_spearman_results = full_like(spearman_results, nan)
        for day in range(num_days):
            todays_returns = returns_results.iloc[day:(day + correlation_length)]
            my_asset_returns = todays_returns.iloc[:, my_asset_column]
            for (asset, other_asset_returns) in todays_returns.iteritems():
                asset_column = (int(asset) - 1)
                expected_pearson_results[(day, asset_column)] = pearsonr(my_asset_returns, other_asset_returns)[0]
                expected_spearman_results[(day, asset_column)] = spearmanr(my_asset_returns, other_asset_returns)[0]
        expected_pearson_results = DataFrame(data=where(expected_mask, expected_pearson_results, nan), index=dates[start_date_index:(end_date_index + 1)], columns=assets)
        assert_frame_equal(pearson_results, expected_pearson_results)
        expected_spearman_results = DataFrame(data=where(expected_mask, expected_spearman_results, nan), index=dates[start_date_index:(end_date_index + 1)], columns=assets)
        assert_frame_equal(spearman_results, expected_spearman_results)
