Overview

• The current setup is a simple module of various methods.
• Methods creates a list of stock pairs and executes a test on them which matches a user defined requirement by way of parallel processing across multiple data ranges. Example given with a cointegration filter.

Architecture

Primary Variables

• stock_pair_list: list of chosen stock pairs
• shift_parameter: an integer which uniquely characterizes the data range being selected. It is essentially the starting point of the data list, meaning how much the data has been rolled forward from its initial starting point, which varies by frequency of the data. It is measured units as that of the freq
  • Note that this is a wrapper around yfinance which extracts ‘close price’ from Yahoo finance

Methods

• cointegration_filter(one_pair): executes the ADF test for a single pair of stocks
  • runner_multiple(): takes a list of stock pairs and list of shift parameters and returns the list of pairs which pass the ADF test across all given data ranges
• runner():
  • stock_pair_list,shift_parameter -> list_of_pairs_with_required_p_value,shift_parameter
  • takes a list of stock pairs as tuples and a shift parameter as an integer and returns a list of stock pairs which pass the ADF test
  • handles the parallel processing
    
• runner_multiple():
  • stock_pair_list, list_of_shift_parameters -> list_of_pairs_with_required_p_values_final
    • handles the iterations over various data sets and returns all the pairs of stocks which pass the ADF test
• The executed flow is runner_multiple() > runner() > cointegration_filter()
• Note I use a custom overarching method run() which is the same as __main__ for all intents and purposes

Code

# Filter to determine if pair of assets pass the ADF test with a p value requirement of .05

def cointegration_filter(cur_stock):  
  
    cur_pair=get_time_period(list(cur_stock[0:2]), custom_data=True, num_data_points=500,shift=cur_stock[2],freq='5m').dropna()  
    model = m.OLS(cur_pair[cur_stock[0]], m.add_constant(cur_pair[cur_stock[1]])).fit()  

    resudials = model.resid
    adf_result = adfuller(resudials)  
    if adf_result[1] < .0501:  
        return model.params.iloc[1]  
    else:  
        return -1000000  
  

# Executes efficient parallel processing where the joblib class, specifically pool, is used. 15 concurrent processes are being run in this situation. Note also there's a splitting based on the time period being run on by the data, which is characterized by the shift parameter
## args is an optional argument which stores time period related information. An example would be the beta values which can be used in the beta based weighing of stocks when executing a portfolio backtest

def runner(stock_pair: pd.Series, shift_parameter: int,filter_func,args= None):  
  
    if args is not None:  
	    p_list = [list(x) + [shift_parameter] + [float(y)] for x,y in zip(stock_pair.index,args)]  
	else:  
        p_list = [list(x) + [shift_parameter] for x in stock_pair.index]  
  
    with Pool(processes=15) as pool:  
        filter_results = pool.map(filter_func, p_list)  
    series_coint = pd.Series(index=stock_pair.index,data= filter_results)  
  
    filt = series_coint[[x for x in series_coint.index if -1000000 not in  series_coint[x]]]  
  
    return pd.Series(data=[[x] for x in filt],index=filt.index)  


# Handles the iteration of each of the stock pairs by time period.

def runner_multiple(stock_pair_list: pd.Series, shift_parameter_list: list,filter_func,args=None):  
	
    if all([len(x) == 1 for x in stock_pair_list]):
        args = pd.DataFrame([x[:len(x)-1] for x in args],index=stock_pair_list.index)  
        args = [list(args.iloc[:,x]) for x in range(len(args.columns))]  
  
    stock_pairs_final = runner(stock_pair_list,shift_parameter_list[0],filter_func,args[0])  
  
    if len(shift_parameter_list) == 1:  
        return (stock_pairs_final + stock_pair_list).dropna()  
    return runner_multiple((stock_pairs_final + stock_pair_list).dropna(),shift_parameter_list[1:],  
                           filter_func,args[1:])

def run():
	
    # Example pair of stock pairs and shift parameters
	stock_pair_list = pd.Series([('AAPL','AMAC'),('FICO','GE')])
	shift_parameter_list = [0,50]
	
	runner_multiple(stock_pair_list,shift_parameter_list,cointegration_filter)