Identifying Anomalies in ECG Signals with Autoencoders

Introduction

Electrocardiogram (ECG) signals record the electrical activity of the heart over time. Identifying abnormal beats (arrhythmias) is crucial for diagnosing cardiac conditions. Because abnormal beats are often rare compared to normal beats, supervised learning can suffer from severe class imbalance. In this tutorial, we will use an unsupervised approach: we will train an LSTM Autoencoder exclusively on normal ECG signals to reconstruct them. Anomalous signals will result in a high reconstruction error, allowing us to detect them without needing labeled anomaly data.

Architecture Overview

graph TD
    ECG(Input ECG Signal: 140 time steps) --> Encoder[LSTM Encoder Layer]
    Encoder --> Bottleneck(Latent Representation: Compressed State)
    Bottleneck --> Repeat[Repeat Vector: 140 times]
    Repeat --> Decoder[LSTM Decoder Layer]
    Decoder --> TimeDist[TimeDistributed Dense Layer]
    TimeDist --> Reconstructed(Reconstructed ECG Signal)
    
    Reconstructed -.-> |Compare to Input| MAE(Calculate Mean Absolute Error)
    MAE --> Threshold{Is MAE > Threshold?}
    Threshold -->|Yes| Anomaly[Flag as Anomaly]
    Threshold -->|No| Normal[Classify as Normal]
        

Prerequisites

• Python 3.8+
• TensorFlow/Keras or PyTorch (We will use TensorFlow in this guide)
• Pandas, NumPy, Matplotlib
• The ECG5000 dataset (or a similar time-series dataset like MIT-BIH)

Step 1: Install Dependencies

pip install tensorflow pandas numpy matplotlib scikit-learn

Step 2: Load and Preprocess the Data

We'll assume the dataset is loaded into a Pandas DataFrame. We need to separate normal from anomalous data, and normalize the signals.

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler

# Load data (assuming the last column is the label: 1 for normal, 0 for anomalous)
# dataframe = pd.read_csv('ecg.csv', header=None)
# raw_data = dataframe.values
# labels = raw_data[:, -1]
# data = raw_data[:, 0:-1]

# For demonstration, let's create dummy data
np.random.seed(42)
data = np.random.randn(5000, 140)
labels = np.random.choice([0, 1], size=(5000,), p=[0.1, 0.9]) # 90% normal, 10% anomalous

# Separate normal and anomalous
train_data, test_data, train_labels, test_labels = train_test_split(data, labels, test_size=0.2, random_state=42)

# Keep only normal data for training the autoencoder
train_data_normal = train_data[train_labels == 1]
test_data_normal = test_data[test_labels == 1]
test_data_anomalous = test_data[test_labels == 0]

# Normalize to [0, 1]
scaler = MinMaxScaler()
train_data_normal = scaler.fit_transform(train_data_normal)
test_data_normal = scaler.transform(test_data_normal)
test_data_anomalous = scaler.transform(test_data_anomalous)

# Reshape for LSTM: [samples, time_steps, features]
# For ECG, we treat each reading as 1 feature at a specific time step
train_data_normal = train_data_normal.reshape((train_data_normal.shape[0], train_data_normal.shape[1], 1))
test_data_normal = test_data_normal.reshape((test_data_normal.shape[0], test_data_normal.shape[1], 1))
test_data_anomalous = test_data_anomalous.reshape((test_data_anomalous.shape[0], test_data_anomalous.shape[1], 1))

Step 3: Build the LSTM Autoencoder Model

Because ECGs are sequential time-series data, standard Dense layers struggle to capture temporal dependencies. Long Short-Term Memory (LSTM) networks are vastly superior for this. We build an Encoder-Decoder architecture using LSTMs.

import tensorflow as tf
from tensorflow.keras import layers, models

input_dim = train_data_normal.shape[1] # 140 time steps
features = 1

# Build LSTM Autoencoder
autoencoder = models.Sequential([
    # Encoder
    layers.Input(shape=(input_dim, features)),
    layers.LSTM(32, activation='relu', return_sequences=True),
    layers.LSTM(16, activation='relu', return_sequences=False), # Bottleneck latent state
    
    # We must repeat the latent state for every time step to feed it to the Decoder
    layers.RepeatVector(input_dim),
    
    # Decoder
    layers.LSTM(16, activation='relu', return_sequences=True),
    layers.LSTM(32, activation='relu', return_sequences=True),
    
    # Output layer to reconstruct the original signal
    layers.TimeDistributed(layers.Dense(features))
])

autoencoder.compile(optimizer='adam', loss='mae')

Step 4: Training

We train the model to output the exact same signal it took as input, using only the normal ECG data.

history = autoencoder.fit(
    train_data_normal, train_data_normal,
    epochs=50,
    batch_size=128,
    validation_data=(test_data_normal, test_data_normal),
    shuffle=True,
    verbose=0
)
print("Training Complete")

Step 5: Detecting Anomalies

To detect an anomaly, we pass a signal through the autoencoder and measure the Mean Absolute Error (MAE) of the reconstruction. If the MAE is above a certain threshold (calculated from the training distribution), we flag it as an anomaly.

# Calculate MAE on normal training data to find a threshold
reconstructions_train = autoencoder.predict(train_data_normal)
# reconstructions are shape (N, 140, 1), reshape to (N, 140) to calculate loss
train_loss = tf.keras.losses.mae(reconstructions_train.squeeze(-1), train_data_normal.squeeze(-1))

# Set threshold to be 1 standard deviation above the mean training loss
threshold = np.mean(train_loss) + np.std(train_loss)
print(f"Anomaly Threshold: {threshold:.4f}")

# Test on anomalous data
reconstructions_anomalous = autoencoder.predict(test_data_anomalous)
anomalous_loss = tf.keras.losses.mae(reconstructions_anomalous.squeeze(-1), test_data_anomalous.squeeze(-1))

# Predict (True if MAE > threshold)
preds_anomalous = tf.math.greater(anomalous_loss, threshold)
accuracy = tf.reduce_mean(tf.cast(preds_anomalous, tf.float32)).numpy()
print(f"Accuracy on detecting true anomalies: {accuracy * 100:.2f}%")

Conclusion

LSTM Autoencoders provide a powerful mechanism for unsupervised anomaly detection in medical time-series data. By learning the temporal latent manifold of "normal" physiology, they naturally flag pathological deviations—whether those are premature ventricular contractions, atrial fibrillation, or subtle ischemic changes—without the need for exhaustive expert annotation of every possible disease state.