專案背景：
一個非法的位元幣交易地址會進行很多次非法的交易，將一個非法的地址的交易行為可以刻畫為一個有向或者無向的網路，那么對這個網路的特性，結構進行識別，來判別這個交易地址是不是一個非法的交易地址，
模型圖：
基于有向圖和無向圖進行隨機游走操作，然后根據路徑的資訊進行特征添加，進行多分支卷積，
根據原始資料簡歷有向圖和無向圖，撲捉到圖中的資訊流向和圖整個的結構分布，體現圖的結構和資訊流向特征，
主要代碼：
import pandas as pd
import numpy as np
import random
import re,os
from collections import Counter
import networkx as nx
from keras_preprocessing import sequence
edges=pd.read_csv('train_data/train_data/address5/edges.csv')
nodes_addr=pd.read_csv('train_data/train_data/address5/nodes_addr.csv')
nodes_tx=pd.read_csv('train_data/train_data/address5/nodes_tx.csv')
# in_tx 處理 成 01 的方式
type_string_01=[]
for i in edges["type:string"]:
if i=="in_tx":
type_string_01.append(0)
else:
type_string_01.append(1)
edges["type:string"]=type_string_01
# 為屬性值建立陣列用于邊權重索引屬性值索引：
attr_list=edges[["amount:double","usd:double","p:double","idx:int"]].values
# edges 圖中最主要的節點
Main_node=edges.values[0][0]
print(Main_node)
# 為所有出現的節點建立一個索引值 用序號為其做標記
node_index={}
index_node={}
nodes_sum=[]
for i in edges["source:START_ID"]:
nodes_sum.append(i)
for i in edges["target:END_ID"]:
nodes_sum.append(i)
nodes_sum=list(set(nodes_sum))
for index,i in enumerate(nodes_sum):
node_index[i]=''.join(re.findall("\d+",i))
index_node[''.join(re.findall("\d+",i))]=i
# print(node_index)
# print(index_node)
print(edges.values.shape)根據原始資料構建有向圖：
# 基于 有向圖
EDGES=[]
for idx,i in enumerate(edges.values):
hang=[]
# 添加節點 兩個 是有向的
hang.append(i[0])
hang.append(i[1])
# 添加邊的資訊
weight={}
weight["weight"]=int(idx)
hang.append(weight)
EDGES.append(hang)
# print(EDGES)
g=nx.DiGraph()
g.add_edges_from(EDGES)
print("邊的數量",len(g.edges()))
print("節點個數",len(g.nodes()))
# print(g.edges.data())
# print(g.edges[1,2]["weight"])
# print(g[1])
# print(list(g.neighbors(2))) # 鄰居也是有向的
# print("節點4和節點1的共同鄰居：",list(nx.common_neighbors(g,2,3)))
# - `node_size`: 指定節點的尺寸大小(默認是300)
# - `node_color`: 指定節點的顏色 (默認是紅色，可以用字串簡單標識顏色，例如'r'為紅色，'b'為綠色等)
# - `node_shape`: 節點的形狀（默認是圓形，用字串'o'標識）
# - `alpha`: 透明度 (默認是1.0，不透明，0為完全透明)
# - `width`: 邊的寬度 (默認為1.0)
# - `edge_color`: 邊的顏色(默認為黑色)
# - `style`: 邊的樣式(默認為實作，可選： solid|dashed|dotted,dashdot)
# - `with_labels`: 節點是否帶標簽（默認為True）
# - `font_size`: 節點標簽字體大小 (默認為12)
# - `font_color`: 節點標簽字體顏色（默認為黑色）
nx.draw(g,with_labels=True,node_shape="*",font_size=3,node_size=5)
# nx.draw(g,node_shape="*",font_size=1,node_size=2)根據有向圖進行游走：
# 基于有向圖的游走 強特征提取
random.seed(1)# 設定隨機種子用于后邊的結果的復現
np.random.seed(0)
num_walks=10
walk_length=10
walks=[]
for i in range(num_walks):
for node in g.nodes():
walk=[]
walk.append(node)
while(len(walk)<walk_length):
node_list=list(g.neighbors(node))
if len(node_list)==0: # 假設在有向圖中這個節點沒有有向的鄰居節點
break
# print("node_list",node_list)
node=np.random.choice(node_list,1).item()
# print(node)
# print("*"*100)
walk.append(node)
if len(walk)==walk_length:
walks.append(walk)
print(len(walks))
# print(walks)
# for i in walks:
# print(len(i))
print(np.array(walks).shape)
# 用節點的 邊的資訊來把 節點序列替換掉： edges中 存在個節點之間存在兩條邊的情況
walks_data=[]
for walk in walks:
walk_data=[]
for idx,node in enumerate(walk):
if node in list(nodes_addr["account:ID"]):
node_data=nodes_addr.loc[nodes_addr["account:ID"]==node,:].values[0][1]
walk_data.append(node_data)
if node in list(nodes_tx["tx_hash:ID"]):
node_data=nodes_tx.loc[nodes_tx["tx_hash:ID"]==node,:].values[0][2:10]
for i in node_data:
walk_data.append(i)
if idx != len(walk)-1:
currt_node=node
next_node=walk[idx+1]
edge_data=edges.loc[(edges["source:START_ID"]==currt_node)&(edges["target:END_ID"]==next_node),:].values[:,3:6].reshape(1,-1)[0]
# 存在個兩個節點之間存在兩條邊的情況 所以不能直接進行相加處理 要充分的展示出來
for i in edge_data:
walk_data.append(i)
walks_data.append(walk_data)
# break
# 對 walks_data中的資料做補齊操作
MaxLen=max(len(i)for i in walks_data)
walks_data=sequence.pad_sequences(walks_data,maxlen=MaxLen,value=0,padding='post')
# 用 0填充
walks_data=np.array(walks_data)
print(walks_data.shape)
根據原始資料構建無向圖：
EDGES=[]
for idx,i in enumerate(edges.values):
hang=[]
# 添加節點 兩個 是有向的
hang.append(i[0])
hang.append(i[1])
# 添加邊的資訊
weight={}
weight["weight"]=int(idx)
hang.append(weight)
EDGES.append(hang)
# print(EDGES)
g=nx.Graph()
g.add_edges_from(EDGES)
print("邊的數量",len(g.edges()))
print("節點個數",len(g.nodes()))
# print(g.edges.data())
# print(g.edges[1,2]["weight"])
# print(g[1])
# print(list(g.neighbors(2))) # 鄰居也是有向的
# print("節點4和節點1的共同鄰居：",list(nx.common_neighbors(g,2,3)))
# - `node_size`: 指定節點的尺寸大小(默認是300)
# - `node_color`: 指定節點的顏色 (默認是紅色，可以用字串簡單標識顏色，例如'r'為紅色，'b'為綠色等)
# - `node_shape`: 節點的形狀（默認是圓形，用字串'o'標識）
# - `alpha`: 透明度 (默認是1.0，不透明，0為完全透明)
# - `width`: 邊的寬度 (默認為1.0)
# - `edge_color`: 邊的顏色(默認為黑色)
# - `style`: 邊的樣式(默認為實作，可選： solid|dashed|dotted,dashdot)
# - `with_labels`: 節點是否帶標簽（默認為True）
# - `font_size`: 節點標簽字體大小 (默認為12)
# - `font_color`: 節點標簽字體顏色（默認為黑色）
nx.draw(g,with_labels=True,node_shape="*",font_size=3,node_size=5)
# nx.draw(g,node_shape="*",font_size=1,node_size=2)根據無向圖進行游走：
random.seed(1)# 設定隨機種子用于后邊的結果的復現
np.random.seed(0)
num_walks=10
walk_length=10
walks=[]
for i in range(num_walks):
for node in g.nodes():
walk=[]
walk.append(node)
while(len(walk)<walk_length):
node_list=list(g.neighbors(node))
if len(node_list)==0: # 假設在有向圖中這個節點沒有有向的鄰居節點
break
# print("node_list",node_list)
node=np.random.choice(node_list,1).item()
# print(node)
# print("*"*100)
walk.append(node)
if len(walk)==walk_length:
walks.append(walk)
print(len(walks))
# print(walks)
# for i in walks:
# print(len(i))
print(np.array(walks).shape)
# 用節點的 邊的資訊來把 節點序列替換掉： edges中 存在個節點之間存在兩條邊的情況
walks_data=[]
for walk in walks:
walk_data=[]
for idx,node in enumerate(walk):
if node in list(nodes_addr["account:ID"]):
node_data=nodes_addr.loc[nodes_addr["account:ID"]==node,:].values[0][1]
walk_data.append(node_data)
if node in list(nodes_tx["tx_hash:ID"]):
node_data=nodes_tx.loc[nodes_tx["tx_hash:ID"]==node,:].values[0][2:10]
for i in node_data:
walk_data.append(i)
if idx != len(walk)-1:
currt_node=node
next_node=walk[idx+1]
edge_data=edges.loc[(edges["source:START_ID"]==currt_node)&(edges["target:END_ID"]==next_node),:].values[:,3:6].reshape(1,-1)[0]
# 存在個兩個節點之間存在兩條邊的情況 所以不能直接進行相加處理 要充分的展示出來
for i in edge_data:
walk_data.append(i)
walks_data.append(walk_data)
# break
# 對 walks_data中的資料做補齊操作
MaxLen=max(len(i)for i in walks_data)
walks_data=sequence.pad_sequences(walks_data,maxlen=MaxLen,value=0,padding='post')
# 用 0填充
walks_data=np.array(walks_data)
print(walks_data.shape)
建模：pytorch 框架
匯入庫：
import pandas as pd
from collections import Counter
import torch
from torch import nn
from torch import optim
import math,os
import tensorflow as tf
import keras
import numpy as np
import pandas as pd
from sklearn.metrics import f1_score
import warnings
import re
import jieba
warnings.filterwarnings('ignore')
from tqdm import tqdm
from sklearn.model_selection import train_test_split
from collections import Counter
import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"] = ['Simhei']
plt.rcParams["axes.unicode_minus"] = False
import re
from pylab import *
import numpy as np
from sklearn.model_selection import train_test_split
import torch
import torch.nn as nn
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
import pandas as pd
import numpy as np
import random
import re,os
from collections import Counter
import networkx as nx
from keras_preprocessing import sequence進行有向圖和無向圖的特征建模個游走
# 有向圖特征提取
def directed_graph_walk(edges,nodes_addr,nodes_tx,Main_node,num_walks,walk_length): # 無有向圖讀取檔案
# random.seed(1)# 設定隨機種子用于后邊的結果的復現
# np.random.seed(0)
Main_node=edges.values[0][0] # 最主要的節點地址
EDGES=[]
for idx,i in enumerate(edges.values):
hang=[]
hang.append(i[0])
hang.append(i[1])
weight={}
weight["weight"]=int(idx)
hang.append(weight)
EDGES.append(hang)
g=nx.DiGraph()
g.add_edges_from(EDGES)
walks=[]
for i in range(num_walks):
for node in g.nodes():
walk=[]
walk.append(node)
while(len(walk)<walk_length):
node_list=list(g.neighbors(node))
if len(node_list)==0: # 假設在有向圖中這個節點沒有有向的鄰居節點
break
node=np.random.choice(node_list,1).item()
walk.append(node)
if len(walk)==walk_length:
if Main_node in walk:
walks.append(walk)
# 用節點的 邊的資訊來把 節點序列替換掉： edges中 存在個節點之間存在兩條邊的情況
walks_data=[]
for walk in walks:
walk_data=[]
for idx,node in enumerate(walk):
if node in list(nodes_addr["account:ID"]):
node_data=nodes_addr.loc[nodes_addr["account:ID"]==node,:].values[0][1]
walk_data.append(node_data)
if node in list(nodes_tx["tx_hash:ID"]):
node_data=nodes_tx.loc[nodes_tx["tx_hash:ID"]==node,:].values[0][2:10]
for i in node_data:
walk_data.append(i)
if idx != len(walk)-1:
currt_node=node
next_node=walk[idx+1]
edge_data=edges.loc[(edges["source:START_ID"]==currt_node)&(edges["target:END_ID"]==next_node),:].values[:,3:6].reshape(1,-1)[0]
# 存在個兩個節點之間存在兩條邊的情況 所以不能直接進行相加處理 要充分的展示出來
for i in edge_data:
walk_data.append(i)
if len(walk_data)> 120:
walk_data=walk_data[0:120]
walks_data.append(walk_data)
# 對 walks_data中的資料做補齊操作
if np.array(walks_data).shape[0]>20: # 資料特征太少的舍棄掉
MaxLen=120
walks_data=sequence.pad_sequences(walks_data,maxlen=MaxLen,value=0,padding='post')
walks_data=np.array(walks_data)
if int(walks_data.shape[0])<200:
walks_data = np.pad(walks_data,((0,int(200-walks_data.shape[0])),(0,0)), 'constant',constant_values=(0,0))# constant連續一樣的值填充
else:
walks_data=walks_data[0:200]
return walks_data
else:
return walks_data
# 無向圖特征提取
def Undirected_graph_walk(edges,nodes_addr,nodes_tx,Main_node,num_walks,walk_length): # 無有向圖讀取檔案
# random.seed(1)# 設定隨機種子用于后邊的結果的復現
# np.random.seed(0)
EDGES=[]
for idx,i in enumerate(edges.values):
# edges 圖中最主要的節點
hang=[]
# 添加節點 兩個 是有向的
hang.append(i[0])
hang.append(i[1])
# 添加邊的資訊
weight={}
weight["weight"]=int(idx)
hang.append(weight)
EDGES.append(hang)
g=nx.Graph()
g.add_edges_from(EDGES)
walks=[]
for i in range(num_walks):
for node in g.nodes():
walk=[]
walk.append(node)
while(len(walk)<walk_length):
node_list=list(g.neighbors(node))
if len(node_list)==0: # 假設在有向圖中這個節點沒有有向的鄰居節點
break
node=np.random.choice(node_list,1).item()
walk.append(node)
if len(walk)==walk_length:
if Main_node in walk:
walks.append(walk)
# 用節點的 邊的資訊來把 節點序列替換掉： edges中 存在個節點之間存在兩條邊的情況
walks_data=[]
for walk in walks:
walk_data=[]
for idx,node in enumerate(walk):
if node in list(nodes_addr["account:ID"]):
node_data=nodes_addr.loc[nodes_addr["account:ID"]==node,:].values[0][1]
walk_data.append(node_data)
if node in list(nodes_tx["tx_hash:ID"]):
node_data=nodes_tx.loc[nodes_tx["tx_hash:ID"]==node,:].values[0][2:10]
for i in node_data:
walk_data.append(i)
if idx != len(walk)-1:
currt_node=node
next_node=walk[idx+1]
edge_data=edges.loc[(edges["source:START_ID"]==currt_node)&(edges["target:END_ID"]==next_node),:].values[:,3:6].reshape(1,-1)[0]
# 存在個兩個節點之間存在兩條邊的情況 所以不能直接進行相加處理 要充分的展示出來
for i in edge_data:
walk_data.append(i)
if len(walk_data)> 120:
walk_data=walk_data[0:120]
walks_data.append(walk_data)
# print("無向圖",np.array(walks_data).shape)
# 對 walks_data中的資料做補齊操作
MaxLen=120
walks_data=sequence.pad_sequences(walks_data,maxlen=MaxLen,value=0,padding='post')
walks_data=np.array(walks_data)
if int(walks_data.shape[0])<200:
walks_data = np.pad(walks_data,((0,int(200-walks_data.shape[0])),(0,0)), 'constant',constant_values=(0,0))# constant連續一樣的值填充
else:
walks_data=walks_data[0:200]
return walks_data
# 正樣本函式加載
def load_g(tree_catalogue_idx):
edges=pd.read_csv("train_data/train_data/address"+str(tree_catalogue_idx)+"/edges.csv")
edges=edges.drop(['amount_and_percentage:string'],axis=1)# 洗掉一列無關緊要的
nodes_addr=pd.read_csv("train_data/train_data/address"+str(tree_catalogue_idx)+'/nodes_addr.csv')
nodes_tx=pd.read_csv("train_data/train_data/address"+str(tree_catalogue_idx)+'/nodes_tx.csv')
# in_tx 處理 成 01 的方式
type_string_01=[]
for i in edges["type:string"]:
if i=="in_tx":
type_string_01.append(0)
else:
type_string_01.append(1)
edges["type:string"]=type_string_01
# 為屬性值建立陣列用于邊權重索引屬性值索引：
attr_list=edges[["amount:double","usd:double","p:double","idx:int"]].values
# edges 圖中最主要的節點
Main_node=edges.values[0][0]
return edges,nodes_addr,nodes_tx,Main_node
# 負樣本函式加載
def load_g_0(tree_catalogue_idx):
edges=pd.read_csv("train_data3/train_data3/address"+str(tree_catalogue_idx)+"/edges.csv")
edges=edges.drop(['amount_and_percentage:string'],axis=1)# 洗掉一列無關緊要的
nodes_addr=pd.read_csv("train_data3/train_data3/address"+str(tree_catalogue_idx)+'/nodes_addr.csv')
nodes_tx=pd.read_csv("train_data3/train_data3/address"+str(tree_catalogue_idx)+'/nodes_tx.csv')
# in_tx 處理 成 01 的方式
type_string_01=[]
for i in edges["type:string"]:
if i=="in_tx":
type_string_01.append(0)
else:
type_string_01.append(1)
edges["type:string"]=type_string_01
# 為屬性值建立陣列用于邊權重索引屬性值索引：
attr_list=edges[["amount:double","usd:double","p:double","idx:int"]].values
# edges 圖中最主要的節點
Main_node=edges.values[0][0]
return edges,nodes_addr,nodes_tx,Main_node
# 正樣本讀取
# 讀取目錄下所有的檔案
data_catalogue= os.listdir("train_data/train_data")
train_x_class_1=[]
print("開始取正樣本")
for catalogue in data_catalogue:#先取幾個圖的資料做測驗
train_one_g=[]
tree_catalogue_idx=str(re.findall("\d+",catalogue)[0])
e=pd.read_csv("train_data/train_data/address"+str(tree_catalogue_idx)+"/edges.csv")
if len(e.values)>5:
edges,nodes_addr,nodes_tx,Main_node=load_g(tree_catalogue_idx)
num_walks=10
walk_length=5
directed_walks_data=directed_graph_walk(edges,nodes_addr,nodes_tx,Main_node,num_walks,walk_length)
num_walks=10
walk_length=5
Undirected_walks_data=Undirected_graph_walk(edges,nodes_addr,nodes_tx,Main_node,num_walks,walk_length)
if len(directed_walks_data)>20 :
print("第幾個圖：",catalogue)
print("有向圖",np.array(directed_walks_data).shape)
print("有向圖",np.array(Undirected_walks_data).shape)
print("----------------------------------")
train_one_g.append(directed_walks_data)
train_one_g.append(Undirected_walks_data)
train_x_class_1.append(train_one_g)
if len(train_x_class_1)==100:
break
# 負樣本讀取
class_0_data_catalogue= os.listdir("train_data3/train_data3")
train_x_class_0=[]
print("開始取負樣本")
for catalogue in class_0_data_catalogue:#先取幾個圖的資料做測驗
train_one_g=[]
tree_catalogue_idx=str(re.findall("\d+",catalogue)[0])
# 測驗是否是空 空則跳過
e=pd.read_csv("train_data3/train_data3/address"+str(tree_catalogue_idx)+"/edges.csv")
if len(e.values)>5:
edges,nodes_addr,nodes_tx,Main_node=load_g_0(tree_catalogue_idx)
num_walks=10
walk_length=5
directed_walks_data=directed_graph_walk(edges,nodes_addr,nodes_tx,Main_node,num_walks,walk_length)
# print("有向圖",np.array(directed_walks_data).shape)
num_walks=10
walk_length=5
Undirected_walks_data=Undirected_graph_walk(edges,nodes_addr,nodes_tx,Main_node,num_walks,walk_length)
if len(directed_walks_data)>20 :
print("第幾個圖：",catalogue)
print("有向圖",np.array(directed_walks_data).shape)
print("有向圖",np.array(Undirected_walks_data).shape)
print("----------------------------------")
train_one_g.append(directed_walks_data)
train_one_g.append(Undirected_walks_data)
train_x_class_0.append(train_one_g)
if len(train_x_class_0)==100:
break
構建資料加載類：
# 資料類加載
from sklearn.model_selection import train_test_split
from torch.utils.data import random_split
from torch.utils.data import Dataset, DataLoader
train_x=train_x_class_1 + train_x_class_0
tranin_y_0=[ 0 for i in range(len(train_x_class_0))]
tranin_y_1=[ 1 for i in range(len(train_x_class_1))]
train_y=tranin_y_1 + tranin_y_0
# X_train, X_test, y_train, y_test = train_test_split(train_x,train_y, test_size=0.1, random_state=1)
class mydataset(Dataset):
def __init__(self): # 讀取加載資料
self._x=torch.FloatTensor(np.array(train_x).astype(float))
self._y=torch.FloatTensor(np.array(train_y).astype(float))
self._len=len(train_x)
def __getitem__(self,item):
return self._x[item],self._y[item]
def __len__(self):# 回傳整個資料的長度
return self._len
data=mydataset()
print(data._len)
# 劃分 訓練集 測驗集
torch.manual_seed(0)
train_data,test_data=random_split(data,[round(0.9*data._len),round(0.1*data._len)])#這個引數有的版本沒有 generator=torch.Generator().manual_seed(0)
# 隨機混亂順序劃分的 四舍五入
# 訓練 loader
train_loader =DataLoader(train_data, batch_size = 10, shuffle = True, num_workers = 0 , drop_last=False)
# 測驗 loader
test_loader =DataLoader(test_data, batch_size =10, shuffle = True, num_workers = 0 , drop_last=False)
# dorp_last 是說最后一組資料不足一個batch的時候 能繼續用還是舍棄， # num_workers 多少個行程載入資料
# # 訓練
# for step,(train_x,train_y) in enumerate(train_loader):
# print(step,':',(train_x.shape,train_y.shape),(train_x,train_y))
# print("------------------------------------------------------------------------------------------------------")
# # 測驗
# for step,(test_x,test_y) in enumerate(test_loader):
# print(step,':',(test_x.shape,test_y.shape),(test_x,test_y))定義模型訓練預測：
# 模型定義
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
class conv_LSTM(nn.Module): # 注意Module首字母需要大寫
def __init__(self, ):
super().__init__()
input_size = 106
hidden_size = 106
input_size = 106
self.conv1_1 = nn.Conv2d(in_channels=1,out_channels=1,kernel_size=(2,15))
self.conv1_2 = nn.Conv2d(in_channels=1,out_channels=1,kernel_size=(10,15))
self.conv2_1 = nn.Conv2d(in_channels=1,out_channels=1,kernel_size=(2,15))
self.conv2_2 = nn.Conv2d(in_channels=1,out_channels=1,kernel_size=(10,15))
# input_size：輸入lstm單元向量的長度 ，hidden_size輸出lstm單元向量的長度，也是輸入、輸出隱藏層向量的長度
self.lstm = nn.LSTM(input_size, hidden_size, num_layers=1) # ,batch_first=True
# --------------------------------------------------------------------------
self.multihead_Linear_k = nn.Linear(hidden_size, hidden_size)
self.multihead_Linear_q = nn.Linear(hidden_size, hidden_size)
self.multihead_Linear_v = nn.Linear(hidden_size, hidden_size)
self.multihead_attn = nn.MultiheadAttention(embed_dim=hidden_size, num_heads=2)
# 因此模型維度 hidden_size 必須可以被頭部數量整除
# --------------------------------------------------------------------------
self.lstm_2 = nn.LSTM(hidden_size, hidden_size, num_layers=1)
# --------------------------------------------------------------------------
self.linear_1 = nn.Linear(hidden_size, 1)
self.ReLU = nn.ReLU()
self.linear_2 = nn.Linear(780,2)
self.softmax=nn.Softmax(dim=1)
def forward(self,x,batch_size):
# x [10, 2, 500, 120]
x=x.transpose(1,0)
x1=x[0]
x2=x[1]
# 上邊這三行這樣取可能有錯誤
x1=x1.to(device)
x2=x2.to(device)
x1=x1.unsqueeze(1)
x2=x2.unsqueeze(1)
# self.conv1(x1)輸入：x[ batch_size, channels, height_1, width_1 ]
x1_1 =self.conv1_1(x1)
x1_2 =self.conv1_2(x1)
x2_1 =self.conv2_1(x2)
x2_2 =self.conv2_2(x2)
x=torch.cat((x1_1, x1_2,x2_1,x2_2), 2)
x=x.squeeze(1)
x = x.transpose(1,0)
x=self.ReLU(x)
# 輸入 lstm的矩陣形狀是：[序列長度，batch_size,每個向量的維度] [序列長度,batch, 64]
lstm_out, h_n = self.lstm(x, None)
# print(lstm_out.shape) #[序列長度,batch_size, 64]
# query，key，value的輸入形狀一定是 [sequence_size, batch_size, emb_size] 比如：value.shape torch.Size( [序列長度,batch_size, 64])
query = self.multihead_Linear_q(lstm_out)
key = self.multihead_Linear_k(lstm_out)
value = self.multihead_Linear_v(lstm_out)
# multihead_attention 輸入矩陣計算 ：
attn_output, attn_output_weights = self.multihead_attn(query, key, value)
# 輸出 attn_output.shape torch.Size([序列長度,batch_size, 64])
lstm_out_2, h_n_2 = self.lstm_2(attn_output, h_n)
lstm_out_2 = lstm_out_2.transpose(0, 1)
# print("lstm_out_2.shape",lstm_out_2.shape)# lstm_out_2.shape torch.Size([20, 600, 128])
# [序列長度,batch_size, 64]
# prediction=lstm_out_2[-1].to(device)
# print(prediction.shape)# torch.Size([batch_size, 64])
# 使用卷積
# 兩個全連接+激活函式
prediction = self.linear_1(lstm_out_2)
prediction = prediction.squeeze(2)
prediction = self.ReLU(prediction)
prediction = self.linear_2(prediction)
prediction = prediction.squeeze(1)
prediction=self.softmax(prediction)
return prediction
model = conv_LSTM().to(device)
loss_function = torch.nn.CrossEntropyLoss().to(device)# 損失函式的計算 交叉熵損失函式計算
optimizer = torch.optim.Adam(model.parameters(), lr=0.001) # 建立優化器實體
batch_size=64
train_loader =DataLoader(train_data, batch_size =batch_size, shuffle = True, num_workers = 0 , drop_last=False)
criterion = torch.nn.CrossEntropyLoss() # 損失函式的計算 交叉熵損失函式計算
sum_train_epoch_loss=[] # 存盤每個epoch 下 訓練train資料的loss
sum_test_epoch_loss=[] # 存盤每個epoch 下 測驗 test資料的loss
# 這個函式是測驗用來測驗x_test y_test 資料 函式
def eval_test(model): # 回傳的是這10個 測驗資料的平均loss
test_epoch_loss = []
with torch.no_grad():
optimizer.zero_grad()
for step, (test_x, test_y) in enumerate(test_loader):
y_pre = model(test_x, batch_size).to(device)
test_y = test_y.to(device)
test_y =test_y .long()
test_loss = loss_function(y_pre, test_y)
test_epoch_loss.append(test_loss.item())
return np.mean(test_epoch_loss)
best_test_loss=10000
epochs=100
# 開始訓練
for epoch in range(epochs):
epoch_loss=[]
for step,(train_x,train_y) in enumerate(train_loader):
y_pred = model(train_x,batch_size).to(device)
train_y=train_y.to(device)
single_loss = loss_function(y_pred,train_y.long())
del train_y
del train_x
del y_pred
epoch_loss.append(single_loss.item())
single_loss.backward()#呼叫backward()自動生成梯度
optimizer.step()#使用optimizer.step()執行優化器，把梯度傳播回每個網路
train_epoch_loss=np.mean(epoch_loss)
test_epoch_loss=eval_test(model)#測驗資料的平均loss
sum_train_epoch_loss.append(train_epoch_loss)
sum_test_epoch_loss.append(test_epoch_loss)
if test_epoch_loss<best_test_loss:
best_test_loss=test_epoch_loss
print("best_test_loss",best_test_loss)
best_model=model
print("epoch:" + str(epoch) + " train_epoch_loss： " + str(train_epoch_loss) + " test_epoch_loss: " + str(test_epoch_loss))
torch.save(best_model, 'best_model.pth')
模型結果預測：
from sklearn.metrics import accuracy_score
#模型加載： 進行預測資料預測
test_loader =DataLoader(test_data, batch_size =20, shuffle = True, num_workers = 0 , drop_last=False)
model.load_state_dict(torch.load('75best_model.pth').cpu().state_dict())
model.eval()
with torch.no_grad():
optimizer.zero_grad()
for step, (test_x, test_y) in enumerate(test_loader):
y_pre = model(test_x,20).cpu()
y_pre=torch.argmax(y_pre,dim=1)
print(y_pre)
print(test_y)
accuracy=accuracy_score(test_y, y_pre)
print(accuracy)0.75準確率
loss圖：
# 畫圖 loss損失函式
fig = plt.figure(facecolor='white', figsize=(10,7 ))
plt.xlabel('第幾個epoch')
plt.ylabel('loss值')
plt.xlim(xmax=len(sum_train_epoch_loss),xmin=0)
plt.ylim(ymax=max(sum_train_epoch_loss),ymin=0)
#畫兩條（0-9）的坐標軸并設定軸標簽x，y
x1 =[i for i in range(0,len(sum_train_epoch_loss),1)] # 隨機產生300個平均值為2，方差為1.2的浮點數，即第一簇點的x軸坐標
y1 = sum_train_epoch_loss # 隨機產生300個平均值為2，方差為1.2的浮點數，即第一簇點的y軸坐標
x2 = [i for i in range(0,len(sum_test_epoch_loss),1)]
y2 = sum_test_epoch_loss
colors1 = '#00CED4' #點的顏色
colors2 = '#DC143C'
area = np.pi * 4**1 # 點面積
# 畫散點圖
plt.scatter(x1, y1, s=area, c=colors1, alpha=0.4, label='train_loss')
plt.scatter(x2, y2, s=area, c=colors2, alpha=0.4, label='val_loss')
# plt.plot([0,9.5],[9.5,0],linewidth = '0.5',color='#000000')
plt.legend()
plt.savefig('loss曲線.png', dpi=300)
plt.show()
請給個贊+收藏、分享不易！
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/404029.html
標籤：區塊鏈
上一篇：基金知識學習