Neural Network

# Convert data to PyTorch tensors
train_X_ts = torch.Tensor(train_X.values)
train_y_ts = torch.Tensor(train_y).view(-1, 1)
test_X_ts = torch.Tensor(test_X.values)
test_y_ts = torch.Tensor(test_y).view(-1, 1)

mport torch.nn as nn

# Define the neural network model
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(train_X.shape[1], 64)
        self.fc2 = nn.Linear(64, 32)
        self.fc3 = nn.Linear(32, 1)
        self.relu = nn.ReLU()

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        x = self.relu(x)
        x = self.fc3(x)
        return x

from torchsummary import summary

# Create an instance of the neural network model
nn_model = Net()
# print the summary of the customized neural network
summary(nn_model, input_size=(1, train_X.shape[1]))

# Define the loss function and optimizer
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(nn_model.parameters(), lr=0.001)

# Train the model
for epoch in range(100):
    optimizer.zero_grad()
    outputs = nn_model(train_X_ts)
    loss = criterion(outputs, train_y_ts)
    loss.backward()
    optimizer.step()

    # Print the loss for every 10 epochs
    if epoch % 10 == 0:
        print("Epoch {}, Loss: {:.4f}".format(epoch, loss.item()))

# evaluate the model on the training and testing set
train_pred = nn_model(train_X_ts).detach().numpy()
print("training rmse: ", np.sqrt(mean_squared_error(train_y_ts, train_pred)))
test_pred = nn_model(test_X_ts).detach().numpy()
print("test rmse: ", np.sqrt(mean_squared_error(test_y_ts, test_pred)))

from torchsummary import summary

# Neural Network Model
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(train_X.shape[1], 64)
        self.fc2 = nn.Linear(64, 32)
        self.fc3 = nn.Linear(32, 1)
        self.relu = nn.ReLU()

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        x = self.relu(x)
        x = self.fc3(x)
        return x

# Initialize neural network model
nn_model = Net()

# Define loss function and optimizer
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(nn_model.parameters(), lr=0.001)

# Function to calculate RMSE and Annualized Return
def compute_rmse_and_annualized_return(nn_model, train_X_ts, train_y_ts, test_X_ts, test_y_ts, spread):
    # Evaluate the model on training and test data
    train_pred = nn_model(train_X_ts).detach().numpy()
    test_pred = nn_model(test_X_ts).detach().numpy()

    # Calculate RMSE
    train_rmse = np.sqrt(mean_squared_error(train_y_ts, train_pred))
    test_rmse = np.sqrt(mean_squared_error(test_y_ts, test_pred))

    print(f"Training RMSE: {train_rmse:.4f}")
    print(f"Test RMSE: {test_rmse:.4f}")

    # Calculate cumulative returns and annualized return
    # Using strategy function similar to Random Forest strategy
    zscore = (spread - test_pred.mean()) / test_pred.std()

    stock1_position = pd.Series(data=0, index=zscore.index)
    stock2_position = pd.Series(data=0, index=zscore.index)

    entry_threshold = 1.0
    exit_threshold = 0.5

    for i in range(1, len(zscore)):
        if zscore.iloc[i] < -entry_threshold and stock1_position.iloc[i-1] == 0:
            stock1_position.iloc[i] = 1
            stock2_position.iloc[i] = -1
        elif zscore.iloc[i] > entry_threshold and stock2_position.iloc[i-1] == 0:
            stock1_position.iloc[i] = -1
            stock2_position.iloc[i] = 1
        elif abs(zscore.iloc[i]) < exit_threshold:
            stock1_position.iloc[i] = 0
            stock2_position.iloc[i] = 0
        else:
            stock1_position.iloc[i] = stock1_position.iloc[i-1]
            stock2_position.iloc[i] = stock2_position.iloc[i-1]

    # Assuming 'spread' contains the returns for both stocks in the pair
    # Convert tensor to NumPy array
    asset1_returns = test_X_ts[:, 0].detach().numpy()  # Assuming asset 1 returns are in the first column
    asset2_returns = test_X_ts[:, 1].detach().numpy()  # Assuming asset 2 returns are in the second column

    # Calculate stock returns based on positions
    stock1_returns = (np.exp(asset1_returns) * stock1_position.shift(1)).fillna(0)
    stock2_returns = (np.exp(asset2_returns) * stock2_position.shift(1)).fillna(0)

    total_returns = stock1_returns + stock2_returns
    cumulative_returns = (1 + total_returns).cumprod()

    final_value = cumulative_returns.iloc[-1]
    n_days = len(cumulative_returns)
    annualized_return = (final_value ** (252 / n_days)) - 1
    annualized_return_percent = annualized_return * 100

    return train_rmse, test_rmse, annualized_return_percent

# Initialize dictionary to store results
nn_strategy_returns = {}

# Iterate over top_10_pairs to calculate RMSE and annualized returns
for pair in top_10_pairs:
    stocks = pair[0]
    data = train_test_split_dict[stocks]

    train_X = data['train_X']
    train_y = data['train_y']
    test_X = data['test_X']
    test_y = data['test_y']
    spread = test_y  # This is the spread (target variable)

    # Convert data to PyTorch tensors
    train_X_ts = torch.Tensor(train_X.values)
    train_y_ts = torch.Tensor(train_y).view(-1, 1)
    test_X_ts = torch.Tensor(test_X.values)
    test_y_ts = torch.Tensor(test_y).view(-1, 1)

    # Train the neural network model
    for epoch in range(100):
        optimizer.zero_grad()
        outputs = nn_model(train_X_ts)
        loss = criterion(outputs, train_y_ts)
        loss.backward()
        optimizer.step()

        if epoch % 10 == 0:
            print(f"Epoch {epoch}, Loss: {loss.item():.4f}")

    # Compute RMSE and annualized return
    train_rmse, test_rmse, annualized_return_percent = compute_rmse_and_annualized_return(
        nn_model, train_X_ts, train_y_ts, test_X_ts, test_y_ts, spread
    )

    # Store results in dictionary
    nn_strategy_returns[stocks] = annualized_return_percent
    print(f"{stocks}: Annualized Return: {annualized_return_percent:.2f}%\n")

# Output results
print("Neural Network Strategy Returns:")
for stock, annualized_return in nn_strategy_returns.items():
    print(f"{stock}: {annualized_return:.2f}%")

pairs = list(nn_strategy_returns.keys())
returns = list(nn_strategy_returns.values())
pair_labels = [f"{p[0]}-{p[1]}" if isinstance(p, (list, tuple)) else str(p) for p in pairs]

mean_return = np.mean(returns)

#Visualization - Neural Network
plt.figure(figsize=(12, 6))
bars = plt.bar(pair_labels, returns, color='lightblue', edgecolor='black')
plt.axhline(mean_return, color='red', linestyle='--', label=f'Mean = {mean_return:.2f}%')

for bar in bars:
    height = bar.get_height()
    plt.text(bar.get_x() + bar.get_width()/2, height + 0.5, f'{height:.2f}%', ha='center', va='bottom')

plt.title('Annualized Return of Neural Network Strategy for Top 10 Pairs')
plt.ylabel('Annualized Return (%)')
plt.xticks(rotation=45)
plt.legend()
plt.tight_layout()
plt.show()