Anomaly Detection for Process Monitoring
# Anomaly Detection for Process Monitoring
## Introduction & Motivation
Early detection of abnormal process behavior prevents defects, equipment damage, and safety incidents. ML-based anomaly detection monitors process data, identifying deviations from normal operation for rapid response. Critical for semiconductor manufacturing, chemical processing, and quality control.
Motivation: Detect process anomalies for rapid intervention.
Applications: Equipment monitoring, process safety, quality control, predictive maintenance.
---
## Core Concepts & Theory
### Normal Behavior Modeling
Baseline pattern characterization.
### Deviation Detection
Distance-based and statistical methods.
### Anomaly Types
Point, contextual, and collective anomalies.
### Threshold Tuning
Balancing sensitivity and false positives.
---
## Mathematical Formulation
Isolation Forest Score:
$$ ext{anomaly\_score} = 2^{-E[\ell]/c(n)}$$
Mahalanobis Distance:
$$D^2 = (\mathbf{x} - \boldsymbol{\mu})^T \Sigma^{-1} (\mathbf{x} - \boldsymbol{\mu})$$
Robust Z-Score:
$$z_i = \frac{x_i - ext{median}}{MAD}$$
---
## Advanced Theory & Extensions
### Deep Learning Approaches
Autoencoders and GANs.
### Ensemble Methods
Combining multiple detectors.
### Contextual Anomalies
Time and context-dependent deviations.
---
## Computational Considerations
Isolation Forest: O(n·t·log n) for n samples, t trees.
Reconstruction: O(D·H) for D features, H hidden units.
Threshold Optimization: O(N·T) for N samples, T thresholds.
---
## Practical Implementation Strategies
### Unsupervised Detection
Normal data characterization.
### Semi-Supervised Learning
Rare labeled anomalies.
### Real-Time Monitoring
Streaming data processing.
---
## Benchmark Datasets & Evaluation
Yahoo Webscope: Benchmark time series.
UCSD Pedestrian: Video surveillance data.
Industrial Datasets: Plant monitoring data.
---
## Key Challenges & Limitations
### Class Imbalance
Rare anomalies in data.
### Concept Drift
Changing normal behavior.
### False Positives
Operational alerts.
---
## Hyperparameter Tuning
Contamination ratio: 0.01-0.1.
Isolation Forest trees: 100-500.
Threshold: 1-3 standard deviations.
---
## Real-World Applications & Case Studies
Equipment Monitoring: Bearing failure detection.
Process Safety: Runaway detection.
Quality Control: Defect identification.
---
## Integration with Other Methods
Anomaly detection + time series; + control systems; + alerting.
---
## Summary & Key Takeaways
ML-based anomaly detection enables proactive process management.
Principles:
1. Baseline: Model normal operations.
2. Detection: Identify deviations.
3. Characterization: Classify anomaly types.
4. Alerting: Trigger interventions.
5. Tuning: Balance sensitivity and specificity.
---
## Appendix: Practical Labs
### Lab 1: Isolation Forest
import numpy as np
class IsolationTree:
def __init__(self, max_depth=10):
self.max_depth = max_depth
self.split_feature = None
self.split_value = None
self.left = None
self.right = None
def build_tree(self, X, current_depth=0):
"""Build isolation tree"""
if current_depth >= self.max_depth or len(X) <= 1:
return
# Random feature and split
n_features = X.shape[1]
feature_idx = np.random.randint(n_features)
split_value = np.random.uniform(X[:, feature_idx].min(), X[:, feature_idx].max())
self.split_feature = feature_idx
self.split_value = split_value
# Split data
left_mask = X[:, feature_idx] < split_value
if np.sum(left_mask) > 0:
self.left = IsolationTree(self.max_depth)
self.left.build_tree(X[left_mask], current_depth + 1)
if np.sum(~left_mask) > 0:
self.right = IsolationTree(self.max_depth)
self.right.build_tree(X[~left_mask], current_depth + 1)
def isolation_path_length(self, x, current_depth=0):
"""Compute path length for sample"""
if self.split_feature is None:
return current_depth
if x[self.split_feature] < self.split_value:
if self.left is None:
return current_depth + 1
return self.left.isolation_path_length(x, current_depth + 1)
else:
if self.right is None:
return current_depth + 1
return self.right.isolation_path_length(x, current_depth + 1)
class IsolationForest:
def __init__(self, n_trees=100, sample_size=256):
self.n_trees = n_trees
self.sample_size = sample_size
self.trees = []
def fit(self, X):
"""Fit forest"""
for _ in range(self.n_trees):
# Random sample
sample_idx = np.random.choice(len(X), self.sample_size, replace=False)
X_sample = X[sample_idx]
tree = IsolationTree(max_depth=10)
tree.build_tree(X_sample)
self.trees.append(tree)
def anomaly_scores(self, X):
"""Compute anomaly scores"""
scores = np.zeros(len(X))
for x in X:
path_lengths = [tree.isolation_path_length(x) for tree in self.trees]
scores[x] = np.mean(path_lengths)
return scores
# Test
X_normal = np.random.randn(100, 2)
forest = IsolationForest(n_trees=10)
forest.fit(X_normal)
# Normal and anomaly points
X_test = np.vstack([X_normal[:5], np.array([[5, 5], [6, 6]])])
scores = forest.anomaly_scores(X_test)
print(f"✓ Isolation forest anomaly scores:")
print(f" Normal points: {scores[:5]}")
print(f" Anomalies: {scores[-2:]}")### Lab 2: Autoencoder for Anomaly Detection
import numpy as np
class AnomalyAutocoder:
def __init__(self, input_dim=10, encoding_dim=3):
self.input_dim = input_dim
self.encoding_dim = encoding_dim
# Encoder weights
self.W_enc = np.random.randn(input_dim, encoding_dim) * 0.1
self.b_enc = np.zeros(encoding_dim)
# Decoder weights
self.W_dec = np.random.randn(encoding_dim, input_dim) * 0.1
self.b_dec = np.zeros(input_dim)
def relu(self, x):
return np.maximum(0, x)
def encode(self, X):
"""Encode to latent space"""
return self.relu(X @ self.W_enc + self.b_enc)
def decode(self, Z):
"""Decode from latent space"""
return Z @ self.W_dec + self.b_dec
def forward(self, X):
"""Forward pass"""
Z = self.encode(X)
X_reconstructed = self.decode(Z)
return X_reconstructed
def train(self, X, epochs=100, lr=0.01):
"""Train autoencoder"""
for epoch in range(epochs):
# Forward
X_recon = self.forward(X)
# Loss
loss = np.mean((X - X_recon) ** 2)
# Backprop (simplified)
dX_recon = -2 * (X - X_recon) / len(X)
dW_dec = self.encode(X).T @ dX_recon
self.W_dec -= lr * dW_dec
def reconstruction_error(self, X):
"""Anomaly score: reconstruction error"""
X_recon = self.forward(X)
errors = np.sum((X - X_recon) ** 2, axis=1)
return errors
# Test
X_train = np.random.randn(100, 10)
autoencoder = AnomalyAutocoder(input_dim=10, encoding_dim=3)
autoencoder.train(X_train, epochs=50)
# Normal and anomalous data
X_test_normal = np.random.randn(5, 10)
X_test_anomaly = np.random.uniform(5, 10, (2, 10))
X_test = np.vstack([X_test_normal, X_test_anomaly])
errors = autoencoder.reconstruction_error(X_test)
print(f"✓ Reconstruction errors:")
print(f" Normal: {errors[:5]}")
print(f" Anomalies: {errors[-2:]}")### Lab 3: Statistical Anomaly Detection
import numpy as np
class StatisticalAnomalyDetector:
def __init__(self, threshold=3.0):
self.threshold = threshold
self.mean = None
self.std = None
def fit(self, X):
"""Fit normal data"""
self.mean = np.mean(X, axis=0)
self.std = np.std(X, axis=0)
def detect_anomalies_zscore(self, X):
"""Z-score based anomaly detection"""
z_scores = np.abs((X - self.mean) / (self.std + 1e-10))
# Anomaly if any dimension exceeds threshold
anomalies = np.max(z_scores, axis=1) > self.threshold
return anomalies
def detect_anomalies_mahalanobis(self, X):
"""Mahalanobis distance based"""
cov = np.cov(X.T)
cov_inv = np.linalg.inv(cov + np.eye(X.shape[1]) * 1e-6)
distances = []
for x in X:
diff = x - self.mean
dist_sq = diff @ cov_inv @ diff
distances.append(np.sqrt(dist_sq))
distances = np.array(distances)
threshold = np.mean(distances) + 2 * np.std(distances)
return distances > threshold
detector = StatisticalAnomalyDetector(threshold=2.5)
X_normal = np.random.randn(100, 5)
detector.fit(X_normal)
X_test_normal = np.random.randn(10, 5)
X_test_anomaly = np.random.uniform(3, 5, (3, 5))
X_test = np.vstack([X_test_normal, X_test_anomaly])
anomalies_zscore = detector.detect_anomalies_zscore(X_test)
anomalies_maha = detector.detect_anomalies_mahalanobis(X_test)
print(f"✓ Statistical anomaly detection:")
print(f" Z-score detections: {np.sum(anomalies_zscore)}")
print(f" Mahalanobis detections: {np.sum(anomalies_maha)}")### Lab 4: Integrated Monitoring System
import numpy as np
class ProcessMonitoringSystem:
def __init__(self, n_sensors=10, alarm_threshold=0.05):
self.n_sensors = n_sensors
self.alarm_threshold = alarm_threshold
self.baseline_mean = np.zeros(n_sensors)
self.baseline_std = np.ones(n_sensors)
self.anomaly_count = 0
self.alert_history = []
def learn_baseline(self, historical_data):
"""Learn normal operations"""
self.baseline_mean = np.mean(historical_data, axis=0)
self.baseline_std = np.std(historical_data, axis=0)
def monitor_reading(self, sensor_data):
"""Check single sensor reading"""
z_scores = np.abs((sensor_data - self.baseline_mean) / (self.baseline_std + 1e-10))
# Compute anomaly score
max_z = np.max(z_scores)
anomaly_likelihood = 1.0 / (1.0 + np.exp(-5*(max_z - 2)))
return anomaly_likelihood
def generate_alert(self, anomaly_score, sensor_values):
"""Generate alert if threshold exceeded"""
if anomaly_score > self.alarm_threshold:
alert = {
'timestamp': len(self.alert_history),
'anomaly_score': anomaly_score,
'sensor_values': sensor_values
}
self.alert_history.append(alert)
self.anomaly_count += 1
return True
return False
def monitor_continuous(self, sensor_stream, n_readings=100):
"""Monitor continuous stream"""
alarms = []
for i in range(n_readings):
reading = sensor_stream[i]
score = self.monitor_reading(reading)
if self.generate_alert(score, reading):
alarms.append(i)
return alarms
# Test
baseline = np.random.randn(100, 10)
monitor = ProcessMonitoringSystem(n_sensors=10)
monitor.learn_baseline(baseline)
# Test stream with anomalies
stream = np.random.randn(50, 10)
stream[30:35] = np.random.uniform(3, 5, (5, 10)) # Insert anomalies
alarms = monitor.monitor_continuous(stream)
print(f"✓ Detected {len(alarms)} anomalies at indices: {alarms}")---