<jupyter_start><jupyter_text>CS7641 Machine Learning*Application of Machine Learning in Pairs Trading*<jupyter_code>import pandas as pd
import numpy as np
import os
import datetime
import matplotlib.pyplot as plt<jupyter_output><empty_output><jupyter_text>Generate the input data Price historyPrice history will be about only one stock.<jupyter_code>total_data = pd.read_csv("training_data.csv")
total_data.head(3)
# Import training dataset
training_set = pd.read_csv("training_data.csv")

# Remove all the data except the pairs we choose
pairs_list = [[83186, 89003],
              [81294, 82581],
              [53640, 83597],
              [43350, 82651],
              [12781, 48531],
              [44644, 90458],
              [21742, 76639],
              [51633, 58819],
              [24969, 24985],
              [81294, 83186],
              [42585, 83621],
              [10395, 53640],
              [23931, 48531],
              [60186, 81095],
              [13856, 48531],
              [16548, 81577]
              ]

for i in range(len(pairs_list)):
    if i==0:
        pairs_training_set = \
            training_set.loc[training_set['PERMNO']==pairs_list[0][0]]
        pairs_training_set = pd.concat([pairs_training_set, 
                   training_set.loc[training_set['PERMNO']==pairs_list[0][1]]])
    else:
        pairs_training_set = pd.concat([pairs_training_set,
                   training_set.loc[training_set['PERMNO']==pairs_list[i][0]]])
        pairs_training_set = pd.concat([pairs_training_set,
                   training_set.loc[training_set['PERMNO']==pairs_list[i][1]]])

training_set = pairs_training_set
training_set.head(3)
# Filtering the table only for the price history
filter_col = ['PERMNO']
filter_col2 = [col for col in training_set if col.startswith('price_')]
filter_col.extend(filter_col2)

training_set_price = training_set[filter_col]
training_set_price = training_set_price.drop(columns=['PERMNO'])
training_set_price.head(3)
class Backtesting():
    """
    Run the backtesting and track the total property history
    """

    def __init__(self, z_score_list, stop_loss=3, buy_z=2, sell_z=-0.5, 
                 total_money=100000):
        self.pairs_num = z_score_list.shape[0]
        self.total_time = z_score_list.shape[1]
        self.stop_loss = stop_loss
        self.buy_z = buy_z
        self.sell_z = sell_z

        self.inverse_price = np.zeros((2 * self.pairs_num))
        self.total_stock = np.zeros((2 * self.pairs_num))
        self.total_inverse = np.zeros((2 * self.pairs_num))
        self.money_for_pair = int(total_money/self.pairs_num) * \
            np.ones((self.pairs_num))

    def trade_decision(self, z_score, pair_num):
        """
        Decide whether we will buy/sell the stock/inverse.
        It is decided by z-score.

        Args:
            z_score: scalar z_score value
            pair_num: pair's number
        Return:
            stock: buy(+1), sell(-1), stop_loss(-2)
            inverse: buy(+1), sell(-1), stop_loss(-2)
        """

        stop_loss = self.stop_loss
        buy_z = self.buy_z
        sell_z = self.sell_z
        total_stock = self.total_stock[2 * pair_num]
        total_inverse = self.total_inverse[2 * pair_num]
        stock, inverse = 0, 0

        # Need to sell everything to stop loss
        if z_score >= stop_loss or z_score <= -1*stop_loss:
            stock, inverse = -2, -2

        # If the z-score is over the buy_z, buy the stock a
        elif z_score >= buy_z and total_stock == 0:
            stock, inverse = 1, 0
        
        # If the z-score below sell_z, sell all the stocks to make profit 
        elif z_score <= sell_z and total_stock > 0:
            stock, inverse = -1, 0

        # If the z-score below -1 * buy_z, buy the inverse a
        elif z_score <= -1 * buy_z and total_inverse == 0:
            stock, inverse = 0, 1

        # If the z-score above -1 * sell_z, sell the inverse a to make profit
        elif z_score >= -1 * sell_z and total_inverse > 0:
            stock, inverse = 0, -1
        
        return stock, inverse


    def cal_trade_vol(self, stock_num, stock_signal, inv_signal, current_price):
        """
        Calculate the volume of trade we will trade

        Args:
            stock_num: Personal number to distinguish the stocks
            stock_signal: Signal to trade the stock (+: buy, -: sell)
            inv_signal: Signal to trade the inverse (+: buy, -: sell)
            current_price: current price to calculate how many stock we will 
                trade
        Return:
            trade_stock_vol: Volume of stock to buy(+) or sell(-)
            trade_inverse_vol: Volume of inverse to buy(+) or sell(-)
        """

        # Initialization
        trade_stock_vol = 0
        trade_inverse_vol = 0

        # Calculate how much money we will use for this stock
        money_for_stock = self.money_for_pair[int(stock_num/2)] / 2

        # Sell the stock if the stock_signal < 0
        if stock_signal < 0:
            trade_stock_vol = -1 * self.total_stock[stock_num]

        # Buy the stock if the stock_signal > 0
        elif stock_signal > 0:
            trade_stock_vol = int(money_for_stock/current_price)

        # Sell the inverse if the inverse_signal < 0
        if inv_signal < 0:
            trade_inverse_vol = -1 * self.total_inverse[stock_num]

        # Buy the stock if the inverse_signal > 0
        elif inv_signal > 0:
            trade_inv_vol = int(money_for_stock/current_price)

        return trade_stock_vol, trade_inverse_vol
        
    def cal_result(self, stock_num, stock_vol, inverse_vol, prev_price, current_price):
        """
        Calculate the each stock's trade results

        Args:
            stock_num: Personal number to distinguish the stocks
            stock_vol: Volume of stock to buy(+) or sell(-)
            inverse_vol: Volume of inverse to buy(+) or sell(-)
            prev_price: Price of the stock the day before.
            current_price: Price of the stock today.
        Return:
            total_property: Sum of all the assets today.
        """

        # Update the price of current inverse
        if self.total_inverse[stock_num] > 0:
            change_ratio = (current_price - prev_price) / prev_price
            self.inverse_price[stock_num] = (1 - change_ratio) * self.inverse_price[stock_num]
        
        # Buy the stock
        if stock_vol > 0:
            self.total_stock[stock_num] += stock_vol
            self.money_for_pair[int(stock_num/2)] -= stock_vol * current_price

        # Sell the stock
        elif stock_vol < 0:
            self.total_stock[stock_num] += stock_vol
            self.money_for_pair[int(stock_num/2)] -= stock_vol * current_price
        
        # Buy the inverse
        if inverse_vol > 0:
            self.total_inverse[stock_num] += inverse_vol
            self.money_for_pair[int(stock_num/2)] -= inverse_vol * current_price
            self.inverse_price[stock_num] = current_price
        
        # Sell the inverse
        if inverse_vol < 0:
            self.total_inverse[stock_num] += inverse_vol
            self.money_for_pair[int(stock_num/2)] -= inverse_vol * self.inverse_price[stock_num]
            self.inverse_price[stock_num] = 0

        total_property = (self.total_inverse[stock_num] * self.inverse_price[stock_num] + 
                          self.total_stock[stock_num] * current_price)

        return total_property


    def backtesting(self, z_score_list, price_list):
        """
        Args:
            z_score_list: list of z-score history (NxT).
            price_list: list of price history (2NxT).
        Return:
            final money list
        """

        total_property = np.zeros((self.pairs_num, self.total_time))

        for t in range(self.total_time):
            for pair_num in range(self.pairs_num):
                stock_a_num = 2*pair_num
                stock_b_num = 2*pair_num+1
                z_score = z_score_list[pair_num]
            
                # Decide whether buy or sell stock a.
                stock_a, inverse_a = self.trade_decision(
                    z_score[t], pair_num)
                stock_b = inverse_a
                inverse_b = stock_a

                # Calculate the buy/sell volume
                stock_a_vol, inverse_a_vol = self.cal_trade_vol(
                    stock_a_num, stock_a, inverse_a, price_list[stock_a_num][t])
                stock_b_vol, inverse_b_vol = self.cal_trade_vol(
                    stock_b_num, stock_b, inverse_b, price_list[stock_b_num][t])

                # Calculate the trade result
                total_property[pair_num][t] += self.cal_result(stock_a_num, stock_a_vol, 
                                                  inverse_a_vol, 
                                                  price_list[stock_a_num][t-1], 
                                                  price_list[stock_a_num][t])
                
                total_property[pair_num][t] += self.cal_result(stock_b_num, stock_b_vol,
                                                     inverse_b_vol, 
                                                     price_list[stock_b_num][t-1], 
                                                     price_list[stock_b_num][t])

                total_property[pair_num][t] += self.money_for_pair[int(pair_num)]

        return total_property<jupyter_output><empty_output><jupyter_text>Run the code with pairsNow, let's run the backtesting and get the money<jupyter_code># Read the z-score history list.
with open('z_score_history.npy', 'rb') as infile:
    z_score_list = np.load(infile)
    z_score_list = z_score_list[:len(pairs_list)]

# Remove the initial 700 data. We generate the z-score after 700 days.
price_list = training_set_price.to_numpy()[:, 700:]

# Initial money is $100000.
initial_money = 100000

# Run the backtesting code
BT = Backtesting(z_score_list, total_money = initial_money)
property_per_pair = BT.backtesting(z_score_list, price_list)

# Plot the graphs about results
x = np.arange(z_score_list.shape[1])

# Graph about each pair's z-score
plt.title('Each Pair\'s z-score')
plt.ylabel('z-score')
plt.xlabel('t (day)')
for i in range(len(pairs_list)):
    fig = plt.plot(x, z_score_list[i])
plt.savefig('each_pair_z_score.png')
plt.show()

# Graph about each pair's assets
plt.title('Each Pair\'s Assets')
plt.ylabel('Assets ($)')
plt.xlabel('t (day)')
for i in range(len(pairs_list)):
    plt.plot(x, property_per_pair[i])
plt.savefig('each_pair_assets.png')
plt.show()

# Graph about total property
plt.title('Total Assets')
plt.ylabel('Assets ($)')
plt.xlabel('t (day)')
total_property = np.sum(property_per_pair, axis=0)
plt.plot(x, total_property)
plt.savefig('total_assets.png')
plt.show()

total_earning_ratio = total_property[-1] / initial_money
print("Total earning ratio: " + str(total_earning_ratio))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09472531483847417, 'fraction_numerical': 0.025095820405183426, 'mean_word_length': 2.18760907504363, 'pattern_counts': {'":': 0, '<': 22, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '24896', 'license': 'Apache-2.0', 'n_tokens_mistral': 3654, 'n_tokens_neox': 3343, 'n_words': 814, 'path': 'Backtesting.ipynb', 'repo_name': 'nrahman35/CS_ML_Project', 'size': '10958'}