docstr: Tests for the built-in factors `RollingPearsonOfReturns` and
  `RollingSpearmanOfReturns`.
docstr_tok: ["Tests","for","the","built-in","factors","`","RollingPearsonOfReturns","`","and","`","RollingSpearmanOfReturns","`","."]
code: def test_correlation_factors(
      self, returns_length, correlation_length
  ):
      """
      Tests for the built-in factors `RollingPearsonOfReturns` and
      `RollingSpearmanOfReturns`.
      """
      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)
  
          # Run a separate pipeline that calculates returns starting
          # (correlation_length - 1) days prior to our start date. This is
          # because we need (correlation_length - 1) extra days of returns to
          # compute our expected correlations.
          results = run_pipeline(
              Pipeline(columns={"returns": returns}),
              dates[start_date_index - (correlation_length - 1)],
              dates[end_date_index],
          )
          returns_results = results["returns"].unstack()
  
          # On each day, calculate the expected correlation coefficients
          # between the asset we are interested in and each other asset. Each
          # correlation is calculated over `correlation_length` days.
          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
          )
  
imports: import numpy
  from numpy import arange, full, full_like, nan, where
  from pandas import DataFrame, date_range, Int64Index, Timestamp
  from pandas.util.testing import assert_frame_equal
  from scipy.stats import linregress, pearsonr, spearmanr
  from empyrical.stats import beta_aligned
  from zipline.assets import Equity, ExchangeInfo
  from zipline.errors import IncompatibleTerms, NonExistentAssetInTimeFrame
  from zipline.pipeline import CustomFactor, Pipeline
  from zipline.pipeline.data import USEquityPricing
  from zipline.pipeline.data.testing import TestingDataSet
  from zipline.pipeline.domain import US_EQUITIES
  from zipline.pipeline.engine import SimplePipelineEngine
  from zipline.pipeline.factors import Returns, RollingLinearRegressionOfReturns, RollingPearsonOfReturns, RollingSpearmanOfReturns, SimpleBeta
  from zipline.pipeline.factors.statistical import vectorized_beta, vectorized_pearson_r
  from zipline.pipeline.loaders.frame import DataFrameLoader
  from zipline.pipeline.sentinels import NotSpecified
  from zipline.testing import AssetID, AssetIDPlusDay, check_arrays, make_alternating_boolean_array, make_cascading_boolean_array, parameter_space
  import zipline.testing.fixtures
  from zipline.testing.predicates import assert_equal
  from zipline.utils.numpy_utils import as_column, bool_dtype, datetime64ns_dtype, float64_dtype
label_desc: Use image node_id 2 for calling the StatisticalBuiltInsTestCase obj's underlying member method code with example usage: obj.test_correlation_factors(returns_length, correlation_length) without return types