pso-lstm python代码

以下是一个基于PSO优化的LSTM模型的Python代码示例：

import numpy as np
import pandas as pd
import math
import random
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error

# 定义LSTM模型
class LSTM:
    def __init__(self, input_dim, hidden_dim, output_dim):
        # 初始化权重
        self.input_dim = input_dim
        self.hidden_dim = hidden_dim
        self.output_dim = output_dim
        self.Wf = np.random.randn(input_dim, hidden_dim) / np.sqrt(input_dim)
        self.Wi = np.random.randn(input_dim, hidden_dim) / np.sqrt(input_dim)
        self.Wc = np.random.randn(input_dim, hidden_dim) / np.sqrt(input_dim)
        self.Wo = np.random.randn(input_dim, hidden_dim) / np.sqrt(input_dim)
        self.Uf = np.random.randn(hidden_dim, hidden_dim) / np.sqrt(hidden_dim)
        self.Ui = np.random.randn(hidden_dim, hidden_dim) / np.sqrt(hidden_dim)
        self.Uc = np.random.randn(hidden_dim, hidden_dim) / np.sqrt(hidden_dim)
        self.Uo = np.random.randn(hidden_dim, hidden_dim) / np.sqrt(hidden_dim)
        self.V = np.random.randn(hidden_dim, output_dim) / np.sqrt(hidden_dim)
        self.bf = np.zeros((1, hidden_dim))
        self.bi = np.zeros((1, hidden_dim))
        self.bc = np.zeros((1, hidden_dim))
        self.bo = np.zeros((1, hidden_dim))
        self.by = np.zeros((1, output_dim))
        
    def sigmoid(self, x):
        return 1 / (1 + np.exp(-x))
    
    def forward(self, X):
        self.T = len(X)
        self.h = np.zeros((self.T + 1, self.hidden_dim))
        self.c = np.zeros((self.T + 1, self.hidden_dim))
        self.f = np.zeros((self.T, self.hidden_dim))
        self.i = np.zeros((self.T, self.hidden_dim))
        self.o = np.zeros((self.T, self.hidden_dim))
        self.ct = np.zeros((self.T, self.hidden_dim))
        self.y = np.zeros((self.T, self.output_dim))
        for t in range(self.T):
            self.f[t] = self.sigmoid(np.dot(X[t], self.Wf) + np.dot(self.h[t-1], self.Uf) + self.bf)
            self.i[t] = self.sigmoid(np.dot(X[t], self.Wi) + np.dot(self.h[t-1], self.Ui) + self.bi)
            self.ct[t] = np.tanh(np.dot(X[t], self.Wc) + np.dot(self.h[t-1], self.Uc) + self.bc)
            self.c[t] = self.f[t] * self.c[t-1] + self.i[t] * self.ct[t]
            self.o[t] = self.sigmoid(np.dot(X[t], self.Wo) + np.dot(self.h[t-1], self.Uo) + self.bo)
            self.h[t] = self.o[t] * np.tanh(self.c[t])
            self.y[t] = np.dot(self.h[t], self.V) + self.by
        return self.y
    
    def predict(self, X):
        y_pred = self.forward(X)
        return y_pred[-1]
    
    def get_weights(self):
        weights = np.concatenate((self.Wf.flatten(), self.Wi.flatten(), self.Wc.flatten(), self.Wo.flatten(),
                                   self.Uf.flatten(), self.Ui.flatten(), self.Uc.flatten(), self.Uo.flatten(),
                                   self.V.flatten(), self.bf.flatten(), self.bi.flatten(), self.bc.flatten(),
                                   self.bo.flatten(), self.by.flatten()))
        return weights
    
    def set_weights(self, weights):
        start = 0
        end = self.input_dim * self.hidden_dim
        self.Wf = np.reshape(weights[start:end], (self.input_dim, self.hidden_dim))
        start = end
        end += self.input_dim * self.hidden_dim
        self.Wi = np.reshape(weights[start:end], (self.input_dim, self.hidden_dim))
        start = end
        end += self.input_dim * self.hidden_dim
        self.Wc = np.reshape(weights[start:end], (self.input_dim, self.hidden_dim))
        start = end
        end += self.input_dim * self.hidden_dim
        self.Wo = np.reshape(weights[start:end], (self.input_dim, self.hidden_dim))
        start = end
        end += self.hidden_dim * self.hidden_dim
        self.Uf = np.reshape(weights[start:end], (self.hidden_dim, self.hidden_dim))
        start = end
        end += self.hidden_dim * self.hidden_dim
        self.Ui = np.reshape(weights[start:end], (self.hidden_dim, self.hidden_dim))
        start = end
        end += self.hidden_dim * self.hidden_dim
        self.Uc = np.reshape(weights[start:end], (self.hidden_dim, self.hidden_dim))
        start = end
        end += self.hidden_dim * self.hidden_dim
        self.Uo = np.reshape(weights[start:end], (self.hidden_dim, self.hidden_dim))
        start = end
        end += self.hidden_dim * self.output_dim
        self.V = np.reshape(weights[start:end], (self.hidden_dim, self.output_dim))
        start = end
        end += self.hidden_dim
        self.bf = np.reshape(weights[start:end], (1, self.hidden_dim))
        start = end
        end += self.hidden_dim
        self.bi = np.reshape(weights[start:end], (1, self.hidden_dim))
        start = end
        end += self.hidden_dim
        self.bc = np.reshape(weights[start:end], (1, self.hidden_dim))
        start = end
        end += self.hidden_dim
        self.bo = np.reshape(weights[start:end], (1, self.hidden_dim))
        start = end
        end += self.output_dim
        self.by = np.reshape(weights[start:end], (1, self.output_dim))
        
# 定义PSO算法
class Particle:
    def __init__(self, input_dim, hidden_dim, output_dim):
        self.position = np.random.randn(1, input_dim * hidden_dim * 9 + hidden_dim * output_dim * 2 + hidden_dim * 5 + output_dim)
        self.velocity = np.zeros_like(self.position)
        self.best_position = self.position
        self.best_error = float('inf')
        
    def update_velocity(self, global_best_position, omega, phi_p, phi_g):
        self.velocity = omega * self.velocity + phi_p * random.random() * (self.best_position - self.position) + phi_g * random.random() * (global_best_position - self.position)
        
    def update_position(self):
        self.position += self.velocity
        
    def get_error(self, X_train, y_train, X_test, y_test, input_dim, hidden_dim, output_dim):
        lstm = LSTM(input_dim, hidden_dim, output_dim)
        lstm.set_weights(self.position)
        y_pred_train = lstm.forward(X_train)
        y_pred_test = lstm.forward(X_test)
        error_train = mean_squared_error(y_train, y_pred_train)
        error_test = mean_squared_error(y_test, y_pred_test)
        if error_test < self.best_error:
            self.best_position = self.position
            self.best_error = error_test
        return error_train, error_test
    
class PSO:
    def __init__(self, n_particles, input_dim, hidden_dim, output_dim, X_train, y_train, X_test, y_test):
        self.n_particles = n_particles
        self.input_dim = input_dim
        self.hidden_dim = hidden_dim
        self.output_dim = output_dim
        self.X_train = X_train
        self.y_train = y_train
        self.X_test = X_test
        self.y_test = y_test
        self.particles = [Particle(input_dim, hidden_dim, output_dim) for _ in range(n_particles)]
        self.global_best_position = np.zeros((1, input_dim * hidden_dim * 9 + hidden_dim * output_dim * 2 + hidden_dim * 5 + output_dim))
        self.global_best_error = float('inf')
        
    def optimize(self, omega, phi_p, phi_g, n_iterations):
        for i in range(n_iterations):
            for particle in self.particles:
                error_train, error_test = particle.get_error(self.X_train, self.y_train, self.X_test, self.y_test, self.input_dim, self.hidden_dim, self.output_dim)
                if error_test < self.global_best_error:
                    self.global_best_position = particle.position
                    self.global_best_error = error_test
                particle.update_velocity(self.global_best_position, omega, phi_p, phi_g)
                particle.update_position()
            print('Iteration {}, Best Error: {}'.format(i + 1, self.global_best_error))
        lstm = LSTM(self.input_dim, self.hidden_dim, self.output_dim)
        lstm.set_weights(self.global_best_position)
        return lstm
    
# 读取数据
df = pd.read_csv('data.csv')
dataset = df['value'].values.reshape(-1, 1)
scaler = MinMaxScaler(feature_range=(0, 1))
dataset = scaler.fit_transform(dataset)

# 划分训练集和测试集
train_size = int(len(dataset) * 0.7)
test_size = len(dataset) - train_size
train, test = dataset[0:train_size, :], dataset[train_size:len(dataset), :]

# 构造时间序列数据
def create_dataset(dataset, look_back=1):
    X, y = [], []
    for i in range(len(dataset) - look_back):
        a = dataset[i:(i+look_back), 0]
        X.append(a)
        y.append(dataset[i + look_back, 0])
    return np.array(X), np.array(y)

look_back = 3
X_train, y_train = create_dataset(train, look_back)
X_test, y_test = create_dataset(test, look_back)

# 转换为LSTM模型的输入格式
X_train = np.reshape(X_train, (X_train.shape[0], X_train.shape[1], 1))
X_test = np.reshape(X_test, (X_test.shape[0], X_test.shape[1], 1))

# 定义PSO算法的参数
n_particles = 10
omega = 0.7
phi_p = 0.2
phi_g = 0.6
n_iterations = 50

# 运行PSO算法
pso = PSO(n_particles, look_back, 4, 1, X_train, y_train, X_test, y_test)
lstm = pso.optimize(omega, phi_p, phi_g, n_iterations)

# 预测
y_pred_train = []
for i in range(len(X_train)):
    y_pred_train.append(lstm.predict(X_train[i]))
y_pred_test = []
for i in range(len(X_test)):
    y_pred_test.append(lstm.predict(X_test[i]))
y_pred_train = scaler.inverse_transform(y_pred_train)
y_pred_test = scaler.inverse_transform(y_pred_test)
y_train = scaler.inverse_transform(y_train.reshape(-1, 1))
y_test = scaler.inverse_transform(y_test.reshape(-1, 1))

# 计算误差
train_error = math.sqrt(mean_squared_error(y_train, y_pred_train))
test_error = math.sqrt(mean_squared_error(y_test, y_pred_test))
print('Train RMSE: %.3f' % train_error)
print('Test RMSE: %.3f' % test_error)