import os import numpy as np from matplotlib import pyplot as plt from keras.models import Sequential from keras import layers from keras.optimizers import RMSprop # data ：浮點數數據組成的原始數組，標準化后 # lookback ：輸入數據應該包括過去多少個時間步。 # delay ：目標應該在未來多少個時間步之后。 # min_index和 max_index ：data數組中的索引，用于界定需要抽取哪些時間步。這有助于保存一部分數據用于驗證、另一部分用于測試。 # shuffle ：是打亂樣本，還是按順序抽取樣本。 # batch_size ：每個批量的樣本數。 # step ：數據采樣的周期（單位：時間步）。我們將其設為6，為的是每小時抽取一個數據點。 def generator(data, lookback, delay, min_index, max_index,shuffle=False, batch_size=128, step=6): if max_index is None: max_index = len(data) - delay - 1 i = min_index + lookback while 1: if shuffle: rows = np.random.randint(min_index + lookback, max_index, size=batch_size) else: if i + batch_size >= max_index: i = min_index + lookback rows = np.arange(i, min(i + batch_size, max_index)) i += len(rows) samples = np.zeros((len(rows), lookback // step, data.shape[-1])) targets = np.zeros((len(rows),)) for j, row in enumerate(rows): indices = range(rows[j] - lookback, rows[j], step) samples[j] = data[indices] targets[j] = data[rows[j] + delay][1] yield samples, targets def evaluate_naive_method(): batch_maes = [] for step in range(val_steps): samples, targets = next(val_gen) preds = samples[:, -1, 1] mae = np.mean(np.abs(preds - targets)) batch_maes.append(mae) print(np.mean(batch_maes)) data_dir = r'E:\GPT2\Dog_Cat' fname = os.path.join(data_dir, r'jena_climate_2009_2016.csv') f = open(fname) data = f.read() f.close() lines = data.split('\n') header = lines[0].split(',') lines = lines[1:] print(header) print(len(lines)) float_data = np.zeros((len(lines), len(header) - 1)) for i, line in enumerate(lines): values = [float(x) for x in line.split(',')[1:]] float_data[i, :] = values mean = float_data[:200000].mean(axis=0) float_data -= mean std = float_data[:200000].std(axis=0) float_data /= std lookback = 1440 step = 6 delay = 144 batch_size = 128 train_gen = generator(float_data, lookback=lookback, delay=delay, min_index=0, max_index=200000, shuffle=True, step=step, batch_size=batch_size) val_gen = generator(float_data, lookback=lookback, delay=delay, min_index=200001, max_index=300000, step=step, batch_size=batch_size) test_gen = generator(float_data, lookback=lookback, delay=delay, min_index=300001, max_index=None, step=step, batch_size=batch_size) val_steps = (300000 - 200001 - lookback) //batch_size #為了查看整個驗證集，需要從val_gen中抽取多少次 test_steps = (len(float_data) - 300001 - lookback) //batch_size evaluate_naive_method() celsius_mae = 0.29 * std[1] model = Sequential() model.add(layers.GRU(32, dropout=0.2, recurrent_dropout=0.2, input_shape=(None, float_data.shape[-1]))) model.add(layers.Dense(1)) model.compile(optimizer=RMSprop(), loss='mae') history = model.fit_generator(train_gen, steps_per_epoch=500, epochs=40, validation_data=val_gen, validation_steps=val_steps) loss = history.history['loss'] val_loss = history.history['val_loss'] epochs = range(1, len(loss) + 1) plt.figure() plt.plot(epochs, loss, 'bo', label='Training loss') plt.plot(epochs, val_loss, 'b', label='Validation loss') plt.title('Training and validation loss') plt.legend() plt.show()