Overview

• Cointegration is tested on a majority of stock pairs from the SP 500 from daily time series data over a 5-year time period. The data is extracted from Yahoo Finance and called with the help of custom module market_analy.py.
• The data ranges from TimeStamp(2020-09-11 00:00:00) to TimeStamp(2025-09-12 00:00:00) and is composed of 1257 data points with a total of 490 stocks. Assets which have > 10 NA values are dropped totally. It is a DataFrame object of shape (1257,490). All NA values of left over assets are dropped.
• Time periods are defined by shift parameters, defined as the number of points shifted from the final index in the DataFrame, 2025-09-12 00:00:00. The set of 5 shift parameters which indicate each roll-forward test are 700, 600,500,400,300. The first 500 points will be the training set and the roll forward length will be 100 points. Validation data set is not defined at present. There will be two types of held out data sets of different lengths. The first set will be those of length 300, meaning the time periods as defined by shift parameters 400,300,200,100,0 respectively. The second set will be data sets of length 500, which can only be defined for the first 3 time periods, with shift param 200,100,0 respectively.
• There is also a custom run method which generates all the stock pairs as tuples and stores them as the indices of a DataFrame along with the parameters of the time periods where the backtesting occurs. All of these objects are taken as inputs, by the custom function runner_multiple(), improved compared to previous versions.
• Filters can be defined for various uses in different parts of the experimental pipeline. Since the goal of this strategy is cointegration, the ADF test will be executed on all filtered stock pairs across each of the 5 testing time periods. The stock pair will then be simulated for pairs based trading if that pair passes the ADF test across all 5 testing periods with a p-value of less than .05. This is strictly a biased test and its main purpose is to determine if pairs can be classified by their joint behavior under pair wise cointegration.

Code

   
# This is the method which sets the framework to execute the backtests.
  
def run():
   
    # The DataFrame as defined above is extracted after being stored as a parquet file.
  
    data = pd.read_parquet('Close_Data_approx_5y_d.parquet')
    b_test_size = 500
    initial_shift = 700
    full_stock = pd.read_parquet('Close_Data_approx_5y_d.parquet').columns
   
    # This generates the list of stock pairs to be tested as tuples. Note that symmetric pairs are not
    included.
  
    stock_pair = [(x,full_stock[y]) for x in full_stock for y in range(list(full_stock).index(x) + 1,len(full_stock) )]
   
    # This generates the shift parameters.
  
    parameters = [initial_shift - x *100 for x in list(range(3))]

    arr = pd.DataFrame(index = stock_pair)
   
    # This is the custom method which takes an empty DataFrame with the indices being the stock pairs
    and executes the ADF test on each of them in each of the time periods. It returns a DataFrame with the results for
    each stock pair with the columns being the metric and the time period it was found. For example, in this situation,
    the p-values are stored for each time period for each stock pair and also the beta values of the proposed linear
    relationship between each of the stocks in each stock pair which can be obtained from the ADF test. This is in order
    to avoid having to recalculate these quantities for beta defined weights, used in the backtesting periods of the
    portfolio simulations later.
  
    runner_multiple(arr, parameters, cointegration_filter).to_parquet('cointegrated_stocks_br_redone.parquet')
    print(pd.read_parquet('cointegrated_stocks_br_redone.parquet'))

   
# This is the custom cointegration filter method which takes as a parameter a list which includes
the stock pair being tested as the first two entries and the shift parameter of the time period this pair is being
tested on for cointegration in the last entry.
  
def cointegration_filter(cur_stock):
    
    # This is a custom method which returns a DataFrame of the closing prices of each of the stocks.
    
    cur_pair=get_time_period(list(cur_stock[0:2]), custom_data=True, num_data_points=500,shift=int(cur_stock[2])+1)
    
    # This executes the ADF test with the use of the statmodels module.
    
    model = m.OLS(cur_pair[cur_stock[0]], m.add_constant(cur_pair[cur_stock[1]])).fit()
    resudials = model.resid
    adf_result = adfuller(resudials)
    arr = np.zeros(2)
    if adf_result[1] < .050000 + .0000000001:
      arr = np.array([model.params.iloc[1],adf_result[1]])
    
      # This returns the filtered pair of stocks, meaning it returns as an np.array the beta value of the proposed lin
      relationship between the stocks as obtained from the ADF test and the p-value for reference if the p-value is less
      than .05. Otherwise, it returns an empty 2 element np.array.
      
      return arr


# This is the custom data extraction method which returns the data of the stocks passed in based on the shift para of the data set,
meaning how much the end of the data set is shifted from the last possible index in the entire
DataFrame. Note the shift is set equal to 1 by default just to avoid the endpoints. The custom_data
parameter is defined to determine if the data set should be obtained from this DataFrame or based on the latest
data as defined by the alpaca live trading api. This is because I use this method for backtesting and live trading.
The frequency is also defined as parameter to indicate what frequency of data to obtain.
## Note that only the necessary part is shown.


def get_time_period( args, custom_data = False, num_data_points = 100,freq='d',details=False,shift =1):
if custom_data:
    if freq == 'd':
        data = pd.read_parquet('Close_Data_approx_5y_d.parquet').iloc[-num_data_points - shift:-shift][args]
...

return data

# The results are stored in a DataFrame of the stock pairs which passed the ADF test in all 5 time periods.