#!/usr/bin/env python
# coding: utf-8
# In[1]:
import random # 导入 random 模块，用于生成随机数
import numpy as np # 导入 NumPy 库，用于数值计算和数组操作
import math # 导入 math 模块，提供数学运算函数
import sys # 导入 sys 模块，用于访问系统相关的功能
from numpy import linalg as LA # 导入 NumPy 的 linalg 模块，用于线性代数运算
import os # 导入 os 模块，提供与操作系统交互的功能
import matplotlib.pyplot as plt # 导入 Matplotlib 库的 pyplot 模块，用于绘图
import pandas as pd # 导入 Pandas 库，用于数据处理和分析
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score # 导入 sklearn 库的 metrics 模块，用于评估模型性能的指标
from sklearn.preprocessing import MinMaxScaler # 导入 sklearn 库的 preprocessing 模块，用于数据预处理
from tensorflow.python.keras.callbacks import EarlyStopping # 导入 TensorFlow 的 keras 模块的 callbacks 模块，用于定义回调函数
from tensorflow.keras.layers import Dense, LSTM, Dropout, GRU, Bidirectional # 导入keras中的Dense、LSTM、Dropout、GRU和Bidirectional层
from tensorflow.python.keras.layers.core import Activation # 导入 TensorFlow 的 keras 模块的 layers 模块的 Activation 类，用于定义激活函数
from tensorflow.python.keras.models import Sequential # 导入 TensorFlow 的 keras 模块的 models 模块，用于定义模型
import tensorflow as tf # 导入 TensorFlow 库，用于构建和训练深度学习模型
# In[2]:
plt.rcParams['font.sans-serif'] = ['SimHei'] # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False # 用来正常显示负号
# In[ ]:
#更多模型搜索机器学习之心，支持模型定制
# In[3]:
# 调用GPU加速
gpus = tf.config.experimental.list_physical_devices(device_type='GPU')
for gpu in gpus:
tf.config.experimental.set_memory_growth(gpu, True)
# In[4]:
'''
定义一个函数用于创建数据集，输入参数包括:
dataset: 数据集
look_back: 回溯长度，即用多少个时间步作为输入预测下一个时间步
'''
def creat_dataset(dataset, look_back):
dataX, dataY = [], []
for i in range(len(dataset) - look_back - 1):# 循环遍历数据集
a = dataset[i: (i + look_back)]# 提取回溯长度的数据作为输入
dataX.append(a)# 将该输入数据添加到输入数据集
dataY.append(dataset[i + look_back])# 将下一个时间步的数据添加到输出数据集
return np.array(dataX), np.array(dataY)# 将输入输出数据集转换为numpy数组并返回
# In[5]:
dataframe = pd.read_csv('焦作.csv',header=0, parse_dates=[0],index_col=0, usecols=[0, 1])# 读取CSV文件，并设置第一列为日期，作为索引
dataset = dataframe.values# 获取数据集的值
dataframe.head(10)# 查看前10行数据
# In[6]:
scaler = MinMaxScaler(feature_range=(0, 1))# 使用 MinMaxScaler 进行数据归一化
dataset = scaler.fit_transform(dataset.reshape(-1, 1))# 将数据变为二维数组并进行归一化
# In[7]:
train_size = int(len(dataset) * 0.8) # 计算训练集大小，将数据集长度的 80% 转换为整数
test_size = len(dataset) - train_size # 计算测试集大小，将剩余的数据作为测试集
train, test = dataset[0: train_size], dataset[train_size: len(dataset)] # 将数据集划分为训练集和测试集，使用切片操作实现
# In[8]:
look_back = 10# 定义时序窗口大小
trainX, trainY = creat_dataset(train, look_back)# 生成训练集
testX, testY = creat_dataset(test, look_back)# 生成测试集
# In[9]:
def build_model(neurons1, neurons2, dropout):#定义一个函数，用于构建 LSTM 模型
# 将训练集和测试集的输入和输出数据分别赋值给 X_train、y_train、X_test 和 y_test
X_train, y_train = trainX, trainY
X_test, y_test = testX, testY
model = Sequential()# 创建 Sequential 模型
model.add(Bidirectional(GRU(units=neurons1, return_sequences=True, input_shape=(look_back, 1))))# 添加第一个 BiGRU 层，指定神经元个数和输入数据形状
model.add(Bidirectional(GRU(units=neurons2, return_sequences=False)))# 添加第二个 BiGRU 层，指定神经元个数
model.add(Dropout(dropout))# 添加 Dropout 层，指定失活率
model.add(Dense(units=1))# 添加全连接层，输出维度为 1
model.add(Activation('relu'))# 添加激活函数层，使用 ReLU 函数
model.compile(loss='mean_absolute_error', optimizer='Adam')# 编译模型，指定损失函数和优化器
return model, X_train, y_train, X_test, y_test# 返回构建好的模型以及训练集和测试集的输入和输出数据
# In[10]:
def training(X):
neurons1 = int(X[0]) # 从 X 中获取超参数 neurons1，并将其转换为整数
neurons2 = int(X[1]) # 从 X 中获取超参数 neurons2，并将其转换为整数
dropout = round(X[2], 6) # 从 X 中获取超参数 dropout，并将其保留 6 位小数
batch_size = int(X[3]) # 从 X 中获取超参数 batch_size，并将其转换为整数
print([neurons1, neurons2, dropout, batch_size]) # 打印超参数列表
model, X_train, y_train, X_test, y_test = build_model(neurons1, neurons2, dropout) # 构建模型并获取训练集、测试集
model.fit(
X_train,
y_train,
batch_size=batch_size,
epochs=100,
validation_split=0.1,
verbose=0,
callbacks=[EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)]) # 训练模型
pred = model.predict(X_test) # 在测试集上进行预测
temp_mse = mean_squared_error(y_test, pred) # 计算预测结果与真实值之间的均方误差
print(temp_mse) # 打印均方误差
return temp_mse # 返回均方误差作为适应度函数的结果
# In[11]:
def fun(X):
output = sum(np.square(X)) + random.random() # 计算输出值，将 X 的平方求和后加上一个随机数
return output
def initial(pop, dim, ub, lb):
X = np.zeros([pop, dim]) # 创建一个大小为 pop x dim 的全零矩阵作为初始种群
for i in range(pop):
for j in range(dim):
X[i, j] = random.random() * (ub[j] - lb[j]) + lb[j] # 随机生成每个个体的位置，位于上下界之间
return X
def CaculateFitness1(X, fun):
fitness = fun(X) # 计算每个个体的适应度值
return fitness
def SortFitness(Fit):
fitness = np.sort(Fit, axis=0) # 对适应度值进行排序
index = np.argsort(Fit, axis=0) # 获取排序后的索引
return fitness, index
def SortPosition(X, index):
Xnew = np.zeros(X.shape) # 创建一个与 X 大小相同的全零矩阵
for i in range(X.shape[0]):
Xnew[i, :] = X[index[i], :] # 根据排序后的索引对个体位置进行排序
return Xnew
def BorderCheck1(X, lb, ub, dim):
for j in range(dim):
if X[j] < lb[j]: # 如果个体位置低于下界，则将其位置设置为上界
X[j] = ub[j]
elif X[j] > ub[j]: # 如果个体位置高于上界，则将其位置设置为下界
X[j] = lb[j]
return X
# In[12]:
def rouletteWheelSelection(x):
CS = np.cumsum(x) # 计算累积和
Random_value = random.random() # 生成一个随机数
index = np.where(Random_value <= CS) # 找到随机数小于等于累积和的位置索引
index = sum(index) # 对索引进行求和，得到选择的个体索引
return index
# In[13]:
def random_select(Pbest_Vulture_1, Pbest_Vulture_2, alpha, betha):
probabilities = [alpha, betha] # 设置选择的概率
index = rouletteWheelSelection(probabilities) # 使用轮盘赌选择算法选择一个索引
if np.any(index) > 0: # 如果索引大于0
random_vulture_X = Pbest_Vulture_1 # 选择 Pbest_Vulture_1
else:
random_vulture_X = Pbest_Vulture_2 # 否则选择
