1蜓耻、摘要

本文主要講解：PSO粒子群優(yōu)化-LSTM-優(yōu)化神經(jīng)網(wǎng)絡(luò)神經(jīng)元個(gè)數(shù)dropout和batch_size，目標(biāo)為對(duì)滬深300價(jià)格進(jìn)行預(yù)測(cè)
主要思路：

1. PSO Parameters ：粒子數(shù)量雾袱、搜索維度恤筛、所有粒子的位置和速度、個(gè)體經(jīng)歷的最佳位置和全局最佳位置芹橡、每個(gè)個(gè)體的歷史最佳適應(yīng)值
2. LSTM Parameters 神經(jīng)網(wǎng)絡(luò)第一層神經(jīng)元個(gè)數(shù)毒坛、神經(jīng)網(wǎng)絡(luò)第二層神經(jīng)元個(gè)數(shù)、dropout比率僻族、batch_size
3. 開始搜索：初始粒子適應(yīng)度計(jì)算粘驰、計(jì)算初始全局最優(yōu)屡谐、計(jì)算適應(yīng)值述么、初始全局最優(yōu)參數(shù)、適應(yīng)度函數(shù)愕掏、更新個(gè)體最優(yōu)度秘、更新全局最優(yōu)、全局最優(yōu)參數(shù)
4. 訓(xùn)練模型，使用PSO找到的最好的全局最優(yōu)參數(shù)
5. plt.show()

2剑梳、數(shù)據(jù)介紹

['SP', 'High', 'Low', 'KP', 'QSP', 'ZDE', 'ZDF', 'CJL']

在這里插入圖片描述

需要數(shù)據(jù)的話去我其他文章找到我的聯(lián)系方式唆貌，有償

3、相關(guān)技術(shù)

PSO好的地方就是論文多垢乙，好寫引用文獻(xiàn)
不過說實(shí)話锨咙，算法優(yōu)化我并不推薦用PSO，雖然說PSO的論文多追逮，但是都被用爛了酪刀，AutoML-NNI，hyperopt钮孵，optuna骂倘，ray都是很好很先進(jìn)的優(yōu)化框架，里面集成了很多效果非常好的優(yōu)化算法巴席，推薦大家學(xué)習(xí)历涝。

4、完整代碼和步驟

代碼輸出如下：

請(qǐng)?zhí)砑訄D片描述

此程序運(yùn)行代碼依賴版本為：

tensorflow==2.5.0
numpy==1.19.5
keras==2.6.0
matplotlib==3.5.2

主運(yùn)行程序入口

import random
import time

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from keras.models import Sequential
from sklearn.metrics import r2_score  # R2
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler
from tensorflow.python.keras.models import Sequential

batch_size = 128
epochs = 2
steps = 10

def process_data():
    dataset = pd.read_csv("D:\項(xiàng)目\量化交易\滬深300/hs300.csv", engine='python', parse_dates=['date'], index_col=['date'])
    columns = ['SP', 'High', 'Low', 'KP', 'QSP', 'ZDE', 'ZDF', 'CJL']

    for col in columns:
        scaler = MinMaxScaler()
        dataset[col] = scaler.fit_transform(dataset[col].values.reshape(-1, 1))
    X = dataset.drop(columns=['SP'], axis=1)
    y = dataset['SP']
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.24, shuffle=False, random_state=666)

    return X_train, y_train, X_test, y_test


def create_dataset(X, y, seq_len=10):
    features = []
    targets = []  # 標(biāo)簽

    for i in range(0, len(X) - seq_len, 1):  # 此處的1表示步長(zhǎng)漾唉，每隔一步滑一下
        data = X.iloc[i:i + seq_len]  # 序列數(shù)據(jù)荧库；前閉后開
        label = y.iloc[i + seq_len]  # 標(biāo)簽數(shù)據(jù)
        # 保存到features和labels
        features.append(data)
        targets.append(label)

    return np.array(features), np.array(targets)


X_train, y_train, X_test, y_test = process_data()
train_dataset, train_labels = create_dataset(X_train, y_train, seq_len=10)
X_test, y_test = create_dataset(X_test, y_test, seq_len=10)

from tensorflow.keras import Sequential, layers

def build_model(neurons, dropout):
    model = Sequential([
        layers.LSTM(units=neurons, input_shape=train_dataset.shape[-2:], return_sequences=True),
        # units=256表示有256個(gè)神經(jīng)元；return_sequences=True表示將結(jié)果傳到下一步
        layers.Dropout(dropout),  # 表示刪除一些神經(jīng)元
        layers.LSTM(units=256, return_sequences=True),
        layers.Dropout(dropout),
        layers.LSTM(units=128, return_sequences=True),
        layers.LSTM(units=32),
        layers.Dense(1)  # 因?yàn)橹挥幸粋€(gè)特征值的輸出
    ])
    return model


def training(X):
    neurons = int(X[0])
    dropout = round(X[1], 6)
    batch_size = int(X[2])
    model = build_model(neurons, dropout)
    model.compile(optimizer='adam',
                  loss='mse')
    model.fit(
        train_dataset,
        train_labels,
        batch_size=batch_size,
        epochs=1,
        validation_data=(X_test, y_test),
        verbose=1)
    model.save(
        'neurons' + str(int(X[0])) + '_dropout' + str(dropout) + '_batch_size' + str(batch_size) + '.h5')
    # 訓(xùn)練完成后可直接加載模型
    # model_lstm = load_model('LSTM_bus_' + str(X[0]) + '_' + str(X[1]) + '_' + str(X[2]) + '_' + '.h5')
    pred = model.predict(X_test)
    le = len(pred)
    y_t = y_test.reshape(-1, 1)
    return pred, le, y_t


def function(ps, test, le):
    ss = sum(((abs(test - ps)) / test) / le)
    return ss


# (1) PSO Parameters
MAX_EPISODES = 2
MAX_EP_STEPS = 2
c1 = 1
c2 = 1
w = 0.5
pN = 1  # 粒子數(shù)量

# (2) LSTM Parameters
dim = 3  # 搜索維度
X = np.zeros((pN, dim))  # 所有粒子的位置和速度
V = np.zeros((pN, dim))
pbest = np.zeros((pN, dim))  # 個(gè)體經(jīng)歷的最佳位置和全局最佳位置
gbest = np.zeros(dim)
p_fit = np.zeros(pN)  # 每個(gè)個(gè)體的歷史最佳適應(yīng)值
print(p_fit.shape)
print(p_fit.shape)
t1 = time.time()

'''
神經(jīng)網(wǎng)絡(luò)第一層神經(jīng)元個(gè)數(shù)： 256-259
dropout比率： 0.03-0.19
batch_size： 64-128
'''
UP = [259, 0.19, 128]
DOWN = [256, 0.03, 64]

# (4) 開始搜索
for i_episode in range(MAX_EPISODES):
    """初始化s"""
    random.seed(8)
    fit = -1e5  # 全局最佳適應(yīng)值
    # 初始粒子適應(yīng)度計(jì)算
    print("計(jì)算初始全局最優(yōu)")
    for i in range(pN):
        for j in range(dim):
            V[i][j] = random.uniform(0, 1)
            if j == 1:
                X[i][j] = random.uniform(DOWN[j], UP[j])
            else:
                X[i][j] = round(random.randint(DOWN[j], UP[j]), 0)
        pbest[i] = X[i]
        le, pred, y_t = training(X[i])
        NN = 1
        tmp = function(pred, y_t, le)
        p_fit[i] = tmp
        if tmp > fit:
            fit = tmp
            gbest = X[i]
    print("初始全局最優(yōu)參數(shù)：{:}".format(gbest))

    fitness = []  # 適應(yīng)度函數(shù)
    for j in range(MAX_EP_STEPS):
        fit2 = []
        plt.title("第{}次迭代".format(i_episode))
        for i in range(pN):
            le, pred, y_t = training(X[i])
            temp = function(pred, y_t, le)
            fit2.append(temp / 1000)
            if temp > p_fit[i]:  # 更新個(gè)體最優(yōu)
                p_fit[i] = temp
                pbest[i] = X[i]
                if p_fit[i] > fit:  # 更新全局最優(yōu)
                    gbest = X[i]
                    fit = p_fit[i]
        print("搜索步數(shù)：{:}".format(j))
        print("個(gè)體最優(yōu)參數(shù)：{:}".format(pbest))
        print("全局最優(yōu)參數(shù)：{:}".format(gbest))
        # [30.          0.14277071 95.        ]
        for i in range(pN):
            V[i] = w * V[i] + c1 * random.uniform(0, 1) * (pbest[i] - X[i]) + c2 * random.uniform(0, 1) * (gbest - X[i])
            ww = 1
            for k in range(dim):
                if DOWN[k] < X[i][k] + V[i][k] < UP[k]:
                    continue
                else:
                    ww = 0
            X[i] = X[i] + V[i] * ww
        fitness.append(fit)

print('Running time: ', time.time() - t1)

# 訓(xùn)練模型  使用PSO找到的最好的神經(jīng)元個(gè)數(shù)
neurons = int(gbest[0])
dropout = gbest[1]
batch_size = int(gbest[2])
model = build_model(neurons, dropout)
model.compile(optimizer='adam',
              loss='mse')
model.summary()
history = model.fit(train_dataset, train_labels, epochs=epochs, batch_size=batch_size, verbose=2)

# 模型預(yù)測(cè)數(shù)據(jù)
test_preds = model.predict(X_test)
test_preds = test_preds[:, 0] # 獲取數(shù)組中的第1列值


# 計(jì)算r2值
score = r2_score(y_test, test_preds)
print("r^2 值為： ", score)

# 繪制 預(yù)測(cè)與真值結(jié)果
plt.figure(figsize=(16,8))
plt.plot(y_test[:1149], label="True value")
plt.plot(test_preds[:1149], label="Pred value")#預(yù)測(cè)值
plt.legend(loc='best')
plt.show()
# 顯示訓(xùn)練結(jié)果

plt.figure(figsize=(16,8))
plt.plot(history.history['loss'], label='train loss')
plt.legend(loc='best')
plt.show()

from sklearn import metrics
#MSE
print(metrics.mean_squared_error(y_test,test_preds))
#RMSE
print(np.sqrt(metrics.mean_squared_error(y_test,test_preds)))
#MAE
print(metrics.mean_absolute_error(y_test,test_preds))

代碼比較復(fù)雜赵刑，如需幫忙請(qǐng)私聊

5电爹、學(xué)習(xí)鏈接

PSO粒子群優(yōu)化-LSTM-pyswarms框架-實(shí)現(xiàn)期貨價(jià)格預(yù)測(cè)

其他類似學(xué)習(xí)鏈接

https://pypi.org/project/pyswarms/
ljvmiranda921/pyswarms
PySwarms(Python粒子群優(yōu)化工具包)的使用:GlobalBestPSO例子解析