Last active
September 16, 2022 02:49
-
-
Save jetnew/41fb278c69d3761dd43141f2eb5e676f to your computer and use it in GitHub Desktop.
LSTM Autoencoder using Keras
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from keras.layers import LSTM, Dense, RepeatVector, TimeDistributed | |
from keras.models import Sequential | |
class LSTM_Autoencoder: | |
def __init__(self, optimizer='adam', loss='mse'): | |
self.optimizer = optimizer | |
self.loss = loss | |
self.n_features = 1 | |
def build_model(self): | |
timesteps = self.timesteps | |
n_features = self.n_features | |
model = Sequential() | |
# Encoder | |
model.add(LSTM(timesteps, activation='relu', input_shape=(timesteps, n_features), return_sequences=True)) | |
model.add(LSTM(16, activation='relu', return_sequences=True)) | |
model.add(LSTM(1, activation='relu')) | |
model.add(RepeatVector(timesteps)) | |
# Decoder | |
model.add(LSTM(timesteps, activation='relu', return_sequences=True)) | |
model.add(LSTM(16, activation='relu', return_sequences=True)) | |
model.add(TimeDistributed(Dense(n_features))) | |
model.compile(optimizer=self.optimizer, loss=self.loss) | |
model.summary() | |
self.model = model | |
def fit(self, X, epochs=3, batch_size=32): | |
self.timesteps = X.shape[1] | |
self.build_model() | |
input_X = np.expand_dims(X, axis=2) | |
self.model.fit(input_X, input_X, epochs=epochs, batch_size=batch_size) | |
def predict(self, X): | |
input_X = np.expand_dims(X, axis=2) | |
output_X = self.model.predict(input_X) | |
reconstruction = np.squeeze(output_X) | |
return np.linalg.norm(X - reconstruction, axis=-1) | |
def plot(self, scores, timeseries, threshold=0.95): | |
sorted_scores = sorted(scores) | |
threshold_score = sorted_scores[round(len(scores) * threshold)] | |
plt.title("Reconstruction Error") | |
plt.plot(scores) | |
plt.plot([threshold_score]*len(scores), c='r') | |
plt.show() | |
anomalous = np.where(scores > threshold_score) | |
normal = np.where(scores <= threshold_score) | |
plt.title("Anomalies") | |
plt.scatter(normal, timeseries[normal][:,-1], s=3) | |
plt.scatter(anomalous, timeseries[anomalous][:,-1], s=5, c='r') | |
plt.show() | |
lstm_autoencoder = LSTM_Autoencoder(optimizer='adam', loss='mse') | |
lstm_autoencoder.fit(normal_timeseries, epochs=3, batch_size=32) | |
scores = lstm_autoencoder.predict(test_timeseries) | |
lstm_autoencoder.plot(scores, test_timeseries, threshold=0.95) |
Can you tell me what time series data you are using with your model? Thanks!
Hi, you may refer to my repository here where I used the Numenta Anomaly Benchmark (machine_temperature_system_failure.csv), for temperature sensor data of an internal component of a large, industrial machine.
Hi, I want to use LSTM-Autoencoder to compress input data (dimension reduction), do you know how I can retrieve the compressed sequence (time-series)?
Thank you so much in advance.
Hi @miladgoodarzi, you can consider iterating through model.layers. Reference: https://stackoverflow.com/questions/50151157/keras-how-to-get-layer-index-when-already-know-layer-name. Cheers!
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Can you tell me what time series data you are using with your model? Thanks!