Terzo tentativo di analisi dati di estensimetro con Tensorflow (iniziato qui). Si tratta di un adattamento dell'esempio sul sito di Keras
In questo post si cerca di impostare una anomaly detection mediante autoencoder
La prima anomalia nella serie dati e' in corrispondenza del movimento indicato dalla freccia nel grafico soprastante
I dati sono stati tagliati in modo da includere solo l'inizio dell'anomalia in modo da non istruire troppo la rete
Il modello converge rapidamente con valori di loss e validation loss similari
sovrapponendo il modello ai dati di train si nota una ottima corrispondenza
sottrando i dati reali dal modello si possono estrapolare le anomalie. Indicato dalla freccia l'anomalia derivante dal movimento
di seguito il codice
# -*- coding: utf-8 -*-
"""timeseries_anomaly_detection_detrend3
Automatically generated by Colaboratory.
Original file is located at
https://colab.research.google.com/drive/12Kkjp_xazCmO4HrK0tmzVPyHIoYxJsYo
"""
import numpy as np
import pandas as pd
from tensorflow import keras
from tensorflow.keras import layers
from matplotlib import pyplot as plt
!rm detrend.*
!wget http://c1p81.altervista.org/detrend3.zip
!rm *.csv
!unzip detrend3.zip
df_small_noise=pd.read_csv(r'detrend3.csv', sep=':', header=0, low_memory=False, infer_datetime_format=True, parse_dates={'datetime':[0]}, index_col=['datetime'],usecols=['Data','detrend'])
print(df_small_noise.head())
print(df_small_noise.shape)
#df_small_noise = df_small_noise[:9500]
plt.plot(df_small_noise['detrend'])
plt.show()
# Normalize and save the mean and std we get,
# for normalizing test data.
training_mean = df_small_noise.mean()
training_std = df_small_noise.std()
df_training_value = (df_small_noise - training_mean) / training_std
print("Number of training samples:", len(df_training_value))
TIME_STEPS = 1000
# Generated training sequences for use in the model.
def create_sequences(values, time_steps=TIME_STEPS):
output = []
for i in range(len(values) - time_steps + 1):
output.append(values[i : (i + time_steps)])
return np.stack(output)
x_train = create_sequences(df_training_value.values)
print("Training input shape: ", x_train.shape)
model = keras.Sequential(
[
layers.Input(shape=(x_train.shape[1], x_train.shape[2])),
layers.Conv1D(
filters=32, kernel_size=7, padding="same", strides=2, activation="relu"
),
layers.Dropout(rate=0.2),
layers.Conv1D(
filters=16, kernel_size=7, padding="same", strides=2, activation="relu"
),
layers.Conv1DTranspose(
filters=16, kernel_size=7, padding="same", strides=2, activation="relu"
),
layers.Dropout(rate=0.2),
layers.Conv1DTranspose(
filters=32, kernel_size=7, padding="same", strides=2, activation="relu"
),
layers.Conv1DTranspose(filters=1, kernel_size=7, padding="same"),
]
)
model.compile(optimizer=keras.optimizers.Adam(learning_rate=0.001), loss="mse")
model.summary()
history = model.fit(
x_train,
x_train,
epochs=10,
batch_size=128,
validation_split=0.1,
callbacks=[
keras.callbacks.EarlyStopping(monitor="val_loss", patience=5, mode="min")
],
)
plt.plot(history.history["loss"], label="Training Loss")
plt.plot(history.history["val_loss"], label="Validation Loss")
plt.legend()
plt.show()
# Get train MAE loss.
x_train_pred = model.predict(x_train)
train_mae_loss = np.mean(np.abs(x_train_pred - x_train), axis=1)
plt.hist(train_mae_loss, bins=50)
plt.xlabel("Train MAE loss")
plt.ylabel("No of samples")
plt.show()
# Get reconstruction loss threshold.
threshold = np.max(train_mae_loss)
print("Reconstruction error threshold: ", threshold)
print(x_train.shape)
# Checking how the first sequence is learnt
plt.plot(x_train[288],label='Dati')
plt.plot(x_train_pred[288],label='Modello')
plt.legend()
plt.show()
anomalia = x_train[288] - x_train_pred[288]
plt.plot(anomalia)
plt.show()
