Ik help je graag een TensorFlow-model te ontwikkelen voor real-time anomaliedetectie in IoT-temperatuurdata. Hier is een complete implementatie in het Nederlands: ## 1. Data Preprocessing Pipeline ```python import tensorflow as tf import numpy as np from sklearn.preprocessing import StandardScaler class DataPreprocessor: def __init__(self, window_size=12): # 60 seconden aan data (12 metingen) self.window_size = window_size self.scaler = StandardScaler() self.is_fitted = False def exponential_smoothing(self, data, alpha=0.3): """Pas exponential smoothing toe op de data""" smoothed = [data[0]] for i in range(1, len(data)): smoothed.append(alpha * data[i] + (1 - alpha) * smoothed[-1]) return np.array(smoothed) def preprocess_stream(self, new_measurement): """Verwerk nieuwe metingen in real-time""" if not self.is_fitted: # Initialiseer met eerste batch data self.scaler.partial_fit([new_measurement]) self.is_fitted = True # Normaliseer de meting normalized = self.scaler.transform([[new_measurement]])[0][0] return normalized ``` ## 2. Autoencoder Model voor Anomaliedetectie ```python class AnomalyDetector(tf.keras.Model): def __init__(self, input_dim): super(AnomalyDetector, self).__init__() self.encoder = tf.keras.Sequential([ tf.keras.layers.Dense(32, activation='relu', input_shape=(input_dim,)), tf.keras.layers.Dropout(0.1), tf.keras.layers.Dense(16, activation='relu'), tf.keras.layers.Dense(8, activation='relu') ]) self.decoder = tf.keras.Sequential([ tf.keras.layers.Dense(16, activation='relu', input_shape=(8,)), tf.keras.layers.Dense(32, activation='relu'), tf.keras.layers.Dense(input_dim, activation='linear') ]) def call(self, x): encoded = self.encoder(x) decoded = self.decoder(encoded) return decoded def detect_anomaly(self, window_data, threshold=0.05): """Detecteer anomalieën op basis van reconstructiefout""" reconstruction = self.predict(window_data.reshape(1, -1)) mse = tf.keras.losses.mse(window_data, reconstruction[0]).numpy() return mse > threshold, mse ``` ## 3. Real-time Streaming Pipeline ```python class RealTimeAnomalyDetector: def __init__(self): self.window_size = 12 self.data_window = [] self.preprocessor = DataPreprocessor(self.window_size) self.model = None self.threshold = 0.05 def train_model(self, historical_data): """Train het model met historische data""" # Preprocess historische data processed_data = [] for measurement in historical_data: processed = self.preprocessor.preprocess_stream(measurement) processed_data.append(processed) # Maak sliding windows windows = [] for i in range(len(processed_data) - self.window_size + 1): windows.append(processed_data[i:i+self.window_size]) windows = np.array(windows) # Bouw en train model self.model = AnomalyDetector(self.window_size) self.model.compile(optimizer='adam', loss='mse') self.model.fit(windows, windows, epochs=50, batch_size=32, validation_split=0.1, verbose=1) def process_new_measurement(self, temperature): """Verwerk nieuwe meting in real-time""" start_time = tf.timestamp() # Preprocess processed = self.preprocessor.preprocess_stream(temperature) # Update data window self.data_window.append(processed) if len(self.data_window) > self.window_size: self.data_window.pop(0) # Controleer of window vol is if len(self.data_window) == self.window_size: window_data = np.array(self.data_window) # Detecteer anomalie is_anomaly, score = self.model.detect_anomaly(window_data, self.threshold) end_time = tf.timestamp() latency = (end_time - start_time).numpy() * 1000 # ms return { 'is_anomaly': is_anomaly, 'anomaly_score': score, 'latency_ms': latency, 'timestamp': tf.timestamp().numpy() } return {'status': 'window_not_full'} ``` ## 4. AWS EC2 Deployment Setup ```python # requirements.txt tensorflow==2.10.0 numpy==1.23.0 scikit-learn==1.1.0 fastapi==0.85.0 uvicorn==0.18.0 # deployment_script.py from fastapi import FastAPI import uvicorn from datetime import datetime app = FastAPI() detector = RealTimeAnomalyDetector() @app.post("/detect_anomaly") async def detect_anomaly(temperature: float): result = detector.process_new_measurement(temperature) return { **result, "processed_at": datetime.utcnow().isoformat() } @app.get("/health") async def health_check(): return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000) ``` ## 5. Optimalisaties voor Lage Latentie ```python class OptimizedDetector: def __init__(self): # Gebruik TensorFlow Lite voor snellere inferentie self.interpreter = None self.input_details = None self.output_details = None def convert_to_tflite(self, keras_model): """Converteer naar TensorFlow Lite voor snellere inferentie""" converter = tf.lite.TFLiteConverter.from_keras_model(keras_model) converter.optimizations = [tf.lite.Optimize.DEFAULT] tflite_model = converter.convert() # Save het model with open('anomaly_detector.tflite', 'wb') as f: f.write(tflite_model) # Load voor inferentie self.interpreter = tf.lite.Interpreter(model_content=tflite_model) self.interpreter.allocate_tensors() self.input_details = self.interpreter.get_input_details() self.output_details = self.interpreter.get_output_details() def tflite_predict(self, input_data): """Snellere predict met TFLite""" self.interpreter.set_tensor(self.input_details[0]['index'], input_data) self.interpreter.invoke() return self.interpreter.get_tensor(self.output_details[0]['index']) ``` ## 6. Implementatie Stappen 1. **Data Collectie**: Verzamel minstens 10.000 historische metingen voor training 2. **Model Training**: Train het autoencoder model met normale data 3. **Threshold Bepaling**: Stel drempelwaarde in op basis van validation data 4. **Deployment**: Deploy op EC2 met Docker container 5. **Monitoring**: Implementeer monitoring voor model performance en latency ## 7. Performance Monitoring ```python # monitoring.py class PerformanceMonitor: def __init__(self): self.latencies = [] self.anomaly_count = 0 self.total_measurements = 0 def update_metrics(self, result): self.total_measurements += 1 if 'latency_ms' in result: self.latencies.append(result['latency_ms']) if result.get('is_anomaly', False): self.anomaly_count += 1 def get_stats(self): return { 'avg_latency_ms': np.mean(self.latencies) if self.latencies else 0, 'anomaly_rate': self.anomaly_count / self.total_measurements, 'total_processed': self.total_measurements } ``` Deze implementatie voldoet aan je eisen: - ✅ Realtime verwerking (<1s latency) - ✅ Detectie van plotselinge pieken/dalen - ✅ Geschikt voor zeldzame anomalieën - ✅ AWS EC2 deployment-ready - ✅ Efficiente preprocessing met smoothing Start met het verzamelen van trainingsdata en pas de threshold aan op basis van je specifieke data!