-
Notifications
You must be signed in to change notification settings - Fork 0
/
train_and_process.py
157 lines (129 loc) · 6.41 KB
/
train_and_process.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
# -*- coding: utf-8 -*-
import os
from sys import platform
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from tensorflow.keras.models import Sequential, load_model
from tensorflow.keras.layers import LSTM, Dense, Bidirectional
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.utils import to_categorical
def train_and_process(iteration, ResulNPA_np, ResulNPB_np, ResulNPC_np, ResulNPD_np):
def sliding_window(data, window_size, step_size=1):
sequences = []
for i in range(0, len(data) - window_size + 1, step_size):
sequences.append(data[i:i + window_size])
return sequences
from statsmodels.tsa.stattools import acf
def autocorrelation(sequence, max_lag):
autocorr = acf(sequence, nlags=max_lag, fft=True, adjusted=False)
return autocorr[1:]
# 假设 data_A 和 data_B 是您的A和B数据集
window_size = 10000
step_size = 1
max_lag = 20
# 将数据集分割成子序列
print("==>数据集分割成子序列")
subsequences_A = sliding_window(ResulNPA_np, window_size, step_size)
subsequences_B = sliding_window(ResulNPB_np, window_size, step_size)
subsequences_C = sliding_window(ResulNPC_np, window_size, step_size)
subsequences_D = sliding_window(ResulNPD_np, window_size, step_size)
print("==>A列长度", len(subsequences_A))
print("==>B列长度", len(subsequences_B))
print("==>C列长度", len(subsequences_C))
print("==>D列长度", len(subsequences_D))
# 对每个子序列计算自相关特征
print("==>计算自相关特征")
from tqdm.auto import tqdm
import concurrent.futures
from functools import partial
import threading
from tqdm.auto import tqdm
import concurrent.futures
from functools import partial
def calculate_features(subsequences, max_lag, position):
# 定义一个用于计算自相关的函数，将其作为线程的任务
def autocorrelation_task(seq, max_lag):
return autocorrelation(seq, max_lag)
features = []
# 创建一个线程池，并指定线程数，这里我们将线程数设置为系统的CPU核心数
with concurrent.futures.ThreadPoolExecutor() as executor:
# 使用partial创建一个新的函数，将max_lag参数固定
autocorrelation_with_lag = partial(autocorrelation_task, max_lag=max_lag)
# 将子序列分配给线程池中的线程，并收集结果
for feature in tqdm(executor.map(autocorrelation_with_lag, subsequences), total=len(subsequences),
desc="自相关处理进度", ncols=100, leave=True, position=0):
features.append(feature)
return features
features_A = calculate_features(subsequences_A, max_lag, 0)
features_B = calculate_features(subsequences_B, max_lag, 0)
features_C = calculate_features(subsequences_C, max_lag, 0)
features_D = calculate_features(subsequences_D, max_lag, 0)
import tensorflow as tf
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.layers import Bidirectional
# TPU detection
try:
tpu = tf.distribute.cluster_resolver.TPUClusterResolver() # TPU detection
print('Running on TPU ', tpu.cluster_spec().as_dict()['worker'])
except ValueError:
tpu = None
if tpu:
tf.config.experimental_connect_to_cluster(tpu)
tf.tpu.experimental.initialize_tpu_system(tpu)
strategy = tf.distribute.experimental.TPUStrategy(tpu)
else:
strategy = tf.distribute.get_strategy()
print("REPLICAS: ", strategy.num_replicas_in_sync)
# features_A, features_B, features_C,features_D 是提取的自相关特征，没有进行标>准化
labels_A = np.zeros(len(features_A))
labels_B = np.ones(len(features_B))
labels_C = np.full(len(features_C), 2)
labels_D = np.full(len(features_D), 3)
# 将特征和标签组合成训练集
X = np.concatenate((features_A, features_B, features_C, features_D), axis=0)
y = np.concatenate((labels_A, labels_B, labels_C, labels_D), axis=0)
# 分出训练集和测试集
X_train_val, X_test, y_train_val, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 在训练集上进行数据处理
#scaler = StandardScaler()
#X_train_val = scaler.fit_transform(X_train_val)
# 将处理方法应用到测试集
#X_test = scaler.transform(X_test)
# 将类别标签转换为 one-hot 编码
y_train_val = to_categorical(y_train_val)
y_test = to_categorical(y_test)
# 将训练数据集分为训练集和验证集
X_train, X_val, y_train, y_val = train_test_split(X_train_val, y_train_val, test_size=0.25, random_state=42)
# 为LSTM模型准备数据（将数据调整为3D格式：[samples, timesteps, features]）
X_train = X_train.reshape((X_train.shape[0], 1, X_train.shape[1]))
X_val = X_val.reshape((X_val.shape[0], 1, X_val.shape[1]))
X_test = X_test.reshape((X_test.shape[0], 1, X_test.shape[1]))
# 如果提供了先前模型的路径，加载该模型；否则，创建一个新模型
if iteration > 0:
model = load_model("my_lstm_model_iteration_{}".format(iteration-1))
else:
# 创建一个新模型
model = Sequential()
model.add(Bidirectional(LSTM(64, return_sequences=True), input_shape=(X_train.shape[1], X_train.shape[2])))
model.add(Bidirectional(LSTM(32)))
model.add(Dense(y_train_val.shape[1], activation='softmax'))
# 编译模型
optimizer = Adam(learning_rate=0.001)
model.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])
# 训练模型
history = model.fit(X_train, y_train, epochs=5, batch_size=512, validation_data=(X_val, y_val), verbose=1, shuffle=False)
# 评估模型
test_loss, test_accuracy = model.evaluate(X_test, y_test, verbose=1)
print("Test loss:", test_loss)
print("Test accuracy:", test_accuracy)
# 保存模型
model_path = "my_lstm_model_iteration_{}".format(iteration)
model.save(model_path)
return model_path