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: 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 catalyst.assets import Equity
  from catalyst.errors import IncompatibleTerms, NonExistentAssetInTimeFrame
  from catalyst.pipeline import CustomFactor, Pipeline
  from catalyst.pipeline.data import USEquityPricing
  from catalyst.pipeline.data.testing import TestingDataSet
  from catalyst.pipeline.engine import SimplePipelineEngine
  from catalyst.pipeline.factors.equity import Returns, RollingLinearRegressionOfReturns, RollingPearsonOfReturns, RollingSpearmanOfReturns
  from catalyst.pipeline.loaders.frame import DataFrameLoader
  from catalyst.pipeline.sentinels import NotSpecified
  from catalyst.testing import AssetID, AssetIDPlusDay, check_arrays, make_alternating_boolean_array, make_cascading_boolean_array, parameter_space
  from catalyst.testing.fixtures import WithSeededRandomPipelineEngine, WithTradingEnvironment, CatalystTestCase
  from catalyst.utils.numpy_utils import 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