代码地址
Pytorch代码实现
1:目录结构
基于图神经网络实现的交通流量预测，主要包括：GCN、GAR、ChebNet算法。
2:数据集信息
数据来自美国的加利福尼亚州的洛杉矶市，CSV文件是关于节点的表示情况，一共有307个节点，npz文件是交通流量的文件，每5分钟输出节点数据信息。
数据集信息：
- PEMS-04
- 数据时间范围：2018.1.1—2018.2.28
- 节点信息：共307个检测器，即Nodes数量为307
- 3个特征：流量值，占有率，速度值。
3:数据探索
dataView.py 文件为数据可视化探索分析，可直接运行，查看3个特征的数据分布。
可以看到每个节点有三个特征，但有两个节点基本是平稳不变的，所以只取第一维特征（即流量特征）。
4:数据处理
traffic_dataset.py文件为数据处理模块，主要包含：
- 读取邻接矩阵函数：get_adjacent_matrix
- 读取数据集函数：get_flow_data
- 加载数据类：class LoadData
其中，LoadData 主要包含数据读取，数据切分和数据归一化处理等。
可单独运行该模块查看结果。
class LoadData(Dataset): # 这个就是把读入的数据处理成模型需要的训练数据和测试数据，一个一个样本能读取出来
def __init__(self, data_path, num_nodes, divide_days, time_interval, history_length, train_mode):
"""
:param data_path: list, ["graph file name" , "flow data file name"], path to save the data file names.
:param num_nodes: int, number of nodes.
:param divide_days: list, [ days of train data, days of test data], list to divide the original data.
:param time_interval: int, time interval between two traffic data records (mins).---5 mins
:param history_length: int, length of history data to be used.
:param train_mode: list, ["train", "test"].
"""
self.data_path = data_path
self.num_nodes = num_nodes
self.train_mode = train_mode
self.train_days = divide_days[0] # 59-14 = 45, train_data
self.test_days = divide_days[1] # 7*2 = 14 ,test_data
self.history_length = history_length # 30/5 = 6, 历史长度为6
self.time_interval = time_interval # 5 min
self.one_day_length = int(24 * 60 / self.time_interval) # 一整天的数据量
self.graph = get_adjacent_matrix(distance_file=data_path[0], num_nodes=num_nodes)
self.flow_norm, self.flow_data = self.pre_process_data(data=get_flow_data(data_path[1]), norm_dim=1) # self.flow_norm为归一化的基
def __len__(self): # 表示数据集的长度
"""
:return: length of dataset (number of samples).
"""
if self.train_mode == "train":
return self.train_days * self.one_day_length - self.history_length # 训练的样本数 ＝ 训练集总长度 － 历史数据长度
elif self.train_mode == "test":
return self.test_days * self.one_day_length # 每个样本都能测试，测试样本数 ＝ 测试总长度
else:
raise ValueError("train mode: [{}] is not defined".format(self.train_mode))
def __getitem__(self, index): # 功能是如何取每一个样本 (x, y), index = [0, L1 - 1]这个是根据数据集的长度确定的
"""
:param index: int, range between [0, length - 1].
:return:
graph: torch.tensor, [N, N].
data_x: torch.tensor, [N, H, D].
data_y: torch.tensor, [N, 1, D].
"""
if self.train_mode == "train":
index = index#训练集的数据是从时间０开始的，这个是每一个流量数据，要和样本（ｘ,y）区别
elif self.train_mode == "test":
index += self.train_days * self.one_day_length#有一个偏移量
else:
raise ValueError("train mode: [{}] is not defined".format(self.train_mode))
data_x, data_y = LoadData.slice_data(self.flow_data, self.history_length, index, self.train_mode)#这个就是样本（ｘ,y）
data_x = LoadData.to_tensor(data_x) # [N, H, D] # 转换成张量
data_y = LoadData.to_tensor(data_y).unsqueeze(1) # [N, 1, D] # 转换成张量，在时间维度上扩维
return {"graph": LoadData.to_tensor(self.graph), "flow_x": data_x, "flow_y": data_y} #组成词典返回
@staticmethod
def slice_data(data, history_length, index, train_mode): #根据历史长度,下标来划分数据样本
"""
:param data: np.array, normalized traffic data.
:param history_length: int, length of history data to be used.
:param index: int, index on temporal axis.
:param train_mode: str, ["train", "test"].
:return:
data_x: np.array, [N, H, D].
data_y: np.array [N, D].
"""
if train_mode == "train":
start_index = index #开始下标就是时间下标本身，这个是闭区间
end_index = index + history_length #结束下标,这个是开区间
elif train_mode == "test":
start_index = index - history_length # 开始下标，这个最后面贴图了，可以帮助理解
end_index = index # 结束下标
else:
raise ValueError("train model {} is not defined".format(train_mode))
data_x = data[:, start_index: end_index] # 在切第二维，不包括end_index
data_y = data[:, end_index] # 把上面的end_index取上
return data_x, data_y
@staticmethod
def pre_process_data(data, norm_dim): # 预处理,归一化
"""
:param data: np.array,原始的交通流量数据
:param norm_dim: int,归一化的维度，就是说在哪个维度上归一化,这里是在dim=1时间维度上
:return:
norm_base: list, [max_data, min_data], 这个是归一化的基.
norm_data: np.array, normalized traffic data.
"""
norm_base = LoadData.normalize_base(data, norm_dim) # 计算 normalize base
norm_data = LoadData.normalize_data(norm_base[0], norm_base[1], data) # 归一化后的流量数据
return norm_base, norm_data # 返回基是为了恢复数据做准备的
@staticmethod
def normalize_base(data, norm_dim):#计算归一化的基
"""
:param data: np.array, 原始的交通流量数据
:param norm_dim: int, normalization dimension.归一化的维度，就是说在哪个维度上归一化,这里是在dim=1时间维度上
:return:
max_data: np.array
min_data: np.array
"""
max_data = np.max(data, norm_dim, keepdims=True) # [N, T, D] , norm_dim=1, [N, 1, D], keepdims=True就保持了纬度一致
min_data = np.min(data, norm_dim, keepdims=True)
return max_data, min_data # 返回最大值和最小值
@staticmethod
def normalize_data(max_data, min_data, data):#计算归一化的流量数据，用的是最大值最小值归一化法
"""
:param max_data: np.array, max data.
:param min_data: np.array, min data.
:param data: np.array, original traffic data without normalization.
:return:
np.array, normalized traffic data.
"""
mid = min_data
base = max_data - min_data
normalized_data = (data - mid) / base
return normalized_data
@staticmethod
def recover_data(max_data, min_data, data): # 恢复数据时使用的，为可视化比较做准备的
"""
:param max_data: np.array, max data.
:param min_data: np.array, min data.
:param data: np.array, normalized data.
:return:
recovered_data: np.array, recovered data.
"""
mid = min_data
base = max_data - min_data
recovered_data = data * base + mid
return recovered_data #这个就是原始的数据
@staticmethod
def to_tensor(data):
return torch.tensor(data, dtype=torch.float)
if __name__ == '__main__':
train_data = LoadData(data_path=["PeMS_04/PeMS04.csv", "PeMS_04/PeMS04.npz"], num_nodes=307, divide_days=[45, 14],
time_interval=5, history_length=6,
train_mode="train")
print(len(train_data))
print(train_data[0]["flow_x"].size())
print(train_data[0]["flow_y"].size())
这里保留１的维度，一是为了保持样本和标签一致，二是便于在别的数据上通用．
5:模型训练和验证
搭建网络框架需要四步：
（１）准备数据
（２）定义模型（模型的定义，以及输入、隐藏、输出层数）
（３）定义损失函数和优化器
（４）训练+测试（把数据灌入里面）
这里，先把整体的网络框架搭建好，之后再去实现具体的模型，这样做的好处是在后面更换模型的时候只需要改一两行代码即可。下面来看网络框架的搭建。
traffic_prediction.py文件中存放整个网络框架的入口函数。
- main函数中首先调用LoadData，从而获取训练集和验证集，然后定义训练轮数、损失函数、优化器、模型选取（GCN、GAT、ChebNet）。
- compute_performance函数用来评估模型性能，通过调用utils写好的封装类进行比较。
def main():
os.environ["CUDA_VISIBLE_DEVICES"] = "0" # 配置GPU,因为可能有多个GPU，这里用了第0号GPU
# 第一步：准备数据（上一节已经准备好了，这里只是调用而已，链接在最开头）
train_data = LoadData(data_path=["PeMS_04/PeMS04.csv", "PeMS_04/PeMS04.npz"], num_nodes=307, divide_days=[45, 14],
time_interval=5, history_length=6,
train_mode="train")
train_loader = DataLoader(train_data, batch_size=64, shuffle=True, num_workers=32) # num_workers是加载数据（batch）的线程数目
test_data = LoadData(data_path=["PeMS_04/PeMS04.csv", "PeMS_04/PeMS04.npz"], num_nodes=307, divide_days=[45, 14],
time_interval=5, history_length=6,
train_mode="test")
test_loader = DataLoader(test_data, batch_size=64, shuffle=False, num_workers=32)
# 第二步：定义模型（这里其实只是加载模型，关于模型的定义在下面单独写了，先假设已经写好）
# my_net = GCN(in_c=6, hid_c=6, out_c=1) # 加载GCN模型
# my_net = ChebNet(in_c=6, hid_c=6, out_c=1, K=2) # 加载ChebNet模型
my_net = GATNet(in_c=6 * 1, hid_c=6, out_c=1, n_heads=2) # 加载GAT模型
device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # 定义设备
my_net = my_net.to(device) # 模型送入设备
# 第三步：定义损失函数和优化器
criterion = nn.MSELoss() # 均方损失函数
optimizer = optim.Adam(params=my_net.parameters())# 没写学习率，表示使用的是默认的，也就是lr=1e-3
# 第四步：训练+测试
# Train model
Epoch = 2 # 训练的次数
my_net.train() # 打开训练模式
for epoch in range(Epoch):
epoch_loss = 0.0
count = 0
start_time = time.time()
for data in train_loader: # ["graph": [B, N, N] , "flow_x": [B, N, H, D], "flow_y": [B, N, 1, D]],一次把一个batch的训练数据取出来
my_net.zero_grad() # 梯度清零
count +=1
predict_value = my_net(data, device).to(torch.device("cpu")) # [B, N, 1, D],由于标签flow_y在cpu中，所以最后的预测值要放回到cpu中
loss = criterion(predict_value, data["flow_y"]) # 计算损失，切记这个loss不是标量
epoch_loss += loss.item() # 这里是把一个epoch的损失都加起来，最后再除训练数据长度，用平均loss来表示
loss.backward() # 反向传播
optimizer.step() # 更新参数
end_time = time.time()
print("Epoch: {:04d}, Loss: {:02.4f}, Time: {:02.2f} mins".format(epoch, 1000 * epoch_loss / len(train_data),
(end_time - start_time) / 60))
# Test Model
# 对于测试:
# 第一、除了计算loss之外，还需要可视化一下预测的结果（定性分析）
# 第二、对于预测的结果这里我使用了 MAE, MAPE, and RMSE 这三种评价标准来评估（定量分析）
my_net.eval() # 打开测试模式
with torch.no_grad(): # 关闭梯度
MAE, MAPE, RMSE = [], [], []# 定义三种指标的列表
Target = np.zeros([307, 1, 1]) # [N, T, D],T=1 ＃ 目标数据的维度，用０填充
Predict = np.zeros_like(Target) # [N, T, D],T=1 # 预测数据的维度
total_loss = 0.0
for data in test_loader: # 一次把一个batch的测试数据取出来
# 下面得到的预测结果实际上是归一化的结果，有一个问题是我们这里使用的三种评价标准以及可视化结果要用的是逆归一化的数据
predict_value = my_net(data, device).to(torch.device("cpu")) # [B, N, 1, D]，B是batch_size, N是节点数量,1是时间T=1, D是节点的流量特征
loss = criterion(predict_value, data["flow_y"]) # 使用MSE计算loss
total_loss += loss.item() # 所有的batch的loss累加
# 下面实际上是把预测值和目标值的batch放到第二维的时间维度，这是因为在测试数据的时候对样本没有shuffle，
# 所以每一个batch取出来的数据就是按时间顺序来的，因此放到第二维来表示时间是合理的.
predict_value = predict_value.transpose(0, 2).squeeze(0) # [1, N, B(T), D] -> [N, B(T), D] -> [N, T, D]
target_value = data["flow_y"].transpose(0, 2).squeeze(0) # [1, N, B(T), D] -> [N, B(T), D] -> [N, T, D]
performance, data_to_save = compute_performance(predict_value, target_value, test_loader) # 计算模型的性能，返回评价结果和恢复好的数据
# 下面这个是每一个batch取出的数据，按batch这个维度进行串联，最后就得到了整个时间的数据，也就是
# [N, T, D] = [N, T1+T2+..., D]
Predict = np.concatenate([Predict, data_to_save[0]], axis=1)
Target = np.concatenate([Target, data_to_save[1]], axis=1)
MAE.append(performance[0])
MAPE.append(performance[1])
RMSE.append(performance[2])
print("Test Loss: {:02.4f}".format(1000 * total_loss / len(test_data)))
# 三种指标取平均
print("Performance: MAE {:2.2f} {:2.2f}% {:2.2f}".format(np.mean(MAE), np.mean(MAPE * 100), np.mean(RMSE)))
Predict = np.delete(Predict, 0, axis=1) # 将第0行的0删除，因为开始定义的时候用0填充，但是时间是从1开始的
Target = np.delete(Target, 0, axis=1)
result_file = "GAT_result.h5"
file_obj = h5py.File(result_file, "w") # 将预测值和目标值保存到文件中，因为要多次可视化看看结果
file_obj["predict"] = Predict # [N, T, D]
file_obj["target"] = Target # [N, T, D]
def compute_performance(prediction, target, data): # 计算模型性能
# 下面的try和except实际上在做这样一件事：当训练+测试模型的时候，数据肯定是经过dataloader的，所以直接赋值就可以了
# 但是如果将训练好的模型保存下来，然后测试，那么数据就没有经过dataloader，是dataloader型的，需要转换成dataset型。
try:
dataset = data.dataset # 数据为dataloader型，通过它下面的属性.dataset类变成dataset型数据
except:
dataset = data # 数据为dataset型，直接赋值
# 下面就是对预测和目标数据进行逆归一化，recover_data()函数在上一小节的数据处理中
# flow_norm为归一化的基，flow_norm[0]为最大值，flow_norm[1]为最小值
# prediction.numpy()和target.numpy()是需要逆归一化的数据，转换成numpy型是因为 recover_data()函数中的数据都是numpy型，保持一致
prediction = LoadData.recover_data(dataset.flow_norm[0], dataset.flow_norm[1], prediction.numpy())
target = LoadData.recover_data(dataset.flow_norm[0], dataset.flow_norm[1], target.numpy())
# 对三种评价指标写了一个类，这个类封装在另一个文件中，在后面
mae, mape, rmse = Evaluation.total(target.reshape(-1), prediction.reshape(-1)) # 变成常向量才能计算这三种指标
performance = [mae, mape, rmse]
recovered_data = [prediction, target]
return performance, recovered_data # 返回评价结果，以及恢复好的数据（为可视化准备的）
if __name__ == '__main__':
main()
# 可视化，在下面的 Evaluation()类中，这里是对应的GAT算法运行的结果，进行可视化
# 如果要对GCN或者chebnet进行可视化，只需要在第45行，注释修改下对应的算法即可
visualize_result(h5_file="GAT_result.h5",
nodes_id = 120, time_se = [0, 24 * 12 * 2], # 是节点的时间范围
visualize_file = "gat_node_120")
备注：如果需要查看不同算法运行结果的比较，可以通过注释修改对应的算法模块即可。
# my_net = GCN(in_c=6, hid_c=6, out_c=1) # 加载GCN模型
# my_net = ChebNet(in_c=6, hid_c=6, out_c=1, K=2) # 加载ChebNet模型
my_net = GATNet(in_c=6 * 1, hid_c=6, out_c=1, n_heads=2) # 加载GAT模型
6:查看结果
以上使用了GCN, ChebNet, GAT三种图卷积来预测交通流量，只考虑了空间上的影响，没有考虑时序上的影响，所以效果有进一步提升空间。这里只是为了实现基于上述三种图卷积预测交通流量。
可以借鉴更为精准的图时空神经网络：如STGCN, ASTGCN, DCRNN等。
可参考：基于Pytorch实现的STGCN在交通流量中的预测