docstr: Tests for `Factor.linear_regression` when passed another 2D factor
  instead of a Slice.
docstr_tok: ["Tests","for","`","Factor.linear_regression","`","when","passed","another","2D","factor","instead","of","a","Slice","."]
code: def _test_factor_regression_method_two_factors(
      self, regression_length
  ):
      """
      Tests for `Factor.linear_regression` when passed another 2D factor
      instead of a Slice.
      """
      assets = self.assets
      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
  
      # The order of these is meant to align with the output of `linregress`.
      outputs = ["beta", "alpha", "r_value", "p_value", "stderr"]
  
      # Ensure that the `linear_regression` method cannot be called with two
      # 2D factors which have different masks.
      returns_masked_1 = Returns(
          window_length=5,
          inputs=[self.col],
          mask=AssetID().eq(1),
      )
      returns_masked_2 = Returns(
          window_length=5,
          inputs=[self.col],
          mask=AssetID().eq(2),
      )
      with self.assertRaises(IncompatibleTerms):
          returns_masked_1.linear_regression(
              target=returns_masked_2,
              regression_length=regression_length,
          )
  
      returns_5 = Returns(window_length=5, inputs=[self.col])
      returns_10 = Returns(window_length=10, inputs=[self.col])
  
      regression_factor = returns_5.linear_regression(
          target=returns_10,
          regression_length=regression_length,
      )
  
      columns = {
          output: getattr(regression_factor, output)
          for output in outputs
      }
      pipeline = Pipeline(columns=columns)
  
      results = run_pipeline(pipeline, start_date, end_date)
  
      output_results = {}
      expected_output_results = {}
      for output in outputs:
          output_results[output] = results[output].unstack()
          expected_output_results[output] = full_like(
              output_results[output],
              nan,
          )
  
      # Run a separate pipeline that calculates returns starting
      # (regression_length - 1) days prior to our start date. This is because
      # we need (regression_length - 1) extra days of returns to compute our
      # expected regressions.
      columns = {"returns_5": returns_5, "returns_10": returns_10}
      results = run_pipeline(
          Pipeline(columns=columns),
          dates[start_date_index - (regression_length - 1)],
          dates[end_date_index],
      )
      returns_5_results = results["returns_5"].unstack()
      returns_10_results = results["returns_10"].unstack()
  
      # On each day, for each asset, calculate the expected regression
      # results of Y ~ X where Y is the asset's rolling 5 day returns and X
      # is the asset's rolling 10 day returns. Each regression is calculated
      # over `regression_length` days of data.
      for day in range(num_days):
          todays_returns_5 = returns_5_results.iloc[
              day : day + regression_length
          ]
          todays_returns_10 = returns_10_results.iloc[
              day : day + regression_length
          ]
          for asset, asset_returns_5 in todays_returns_5.iteritems():
              asset_column = int(asset) - 1
              asset_returns_10 = todays_returns_10[asset]
              expected_regression_results = linregress(
                  y=asset_returns_5,
                  x=asset_returns_10,
              )
              for i, output in enumerate(outputs):
                  expected_output_results[output][
                      day, asset_column
                  ] = expected_regression_results[i]
  
      for output in outputs:
          output_result = output_results[output]
          expected_output_result = DataFrame(
              expected_output_results[output],
              index=dates[start_date_index : end_date_index + 1],
              columns=assets,
          )
          assert_frame_equal(output_result, expected_output_result)
  
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 7 for calling the StatisticalMethodsTestCase obj's underlying member method code with example usage: obj._test_factor_regression_method_two_factors(regression_length) without return types