Zeitreihendaten in Agenten verarbeiten
Gleitende Fenster, Aggregation und Anomalieerkennung für gestreamte Sensordaten.
Zeitreihendaten in Agenten verarbeiten ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Zeitreihen in IoT-Agenten
Die Sensordaten treffen als Zeitreihe ein: als Folge von Paaren aus (timestamp, value). Rohe Sensormesswerte enthalten Rauschen, Lücken und gelegentliche Spitzen. Agenten, die ohne Verarbeitung auf Rohdaten reagieren, lösen häufig Fehlalarme aus oder übersehen tatsächliche Ereignisse.
Die Zeitreihenverarbeitung wandelt rohe Signale in verwertbare Erkenntnisse um.
Einen Puffer für ein gleitendes Fenster erstellen
Ein gleitendes Fenster speichert nur die letzten N Messwerte im Arbeitsspeicher. Sobald das Fenster voll ist, wird der älteste Messwert entfernt, wenn ein neuer hinzukommt. Dies bildet die Grundlage für jede Zeitreihenanalyse in Agenten.
from collections import deque
from datetime import datetime
class SensorBuffer:
def __init__(self, window_size: int = 60):
self.window_size = window_size
self._data = deque(maxlen=window_size)
def add(self, value: float, timestamp: datetime = None):
ts = timestamp or datetime.utcnow()
self._data.append({'ts': ts, 'value': value})
def values(self) -> list:
return [d['value'] for d in self._data]
def timestamps(self) -> list:
return [d['ts'] for d in self._data]
def is_full(self) -> bool:
return len(self._data) == self.window_size
buf = SensorBuffer(window_size=60)
buf.add(22.5)
buf.add(22.8)
buf.add(23.1)
print(f'Buffer: {len(buf._data)} readings, values: {buf.values()}')Gleitender Mittelwert
Ein einfacher gleitender Mittelwert (SMA) glättet Rauschen, indem er den Mittelwert der letzten N Werte bildet. Dadurch wird der Einfluss einzelner Sensorfehler reduziert und der zugrunde liegende Trend sichtbar. Verwenden Sie ihn als Basis für die Anomalieerkennung.
import statistics
def simple_moving_average(values: list, window: int = 10) -> list:
if len(values) < window:
return []
return [
statistics.mean(values[i - window:i])
for i in range(window, len(values) + 1)
]
def exponential_moving_average(values: list, alpha: float = 0.2) -> list:
"""EMA weights recent values more heavily."""
if not values:
return []
ema = [values[0]]
for v in values[1:]:
ema.append(alpha * v + (1 - alpha) * ema[-1])
return ema
readings = [22.1, 22.3, 22.0, 35.0, 22.2, 22.4, 22.1, 22.3, 22.5, 22.2, 22.4]
sma = simple_moving_average(readings, window=5)
ema = exponential_moving_average(readings, alpha=0.2)
print(f'SMA (last 3): {[round(v,2) for v in sma[-3:]]}')
print(f'EMA (last 3): {[round(v,2) for v in ema[-3:]]}')Spike-Erkennung
Ein Spike ist ein Messwert, der um mehr als N Standardabweichungen vom aktuellen Trend abweicht (z-Score-Verfahren). Dies ist der gängigste Ansatz zur Anomalieerkennung bei Sensordaten.
import statistics
def detect_spikes(
values: list,
window: int = 20,
z_threshold: float = 3.0
) -> list:
"""Returns list of (index, value, z_score) for detected spikes."""
if len(values) < window:
return []
spikes = []
for i in range(window, len(values)):
window_vals = values[i - window:i]
mean = statistics.mean(window_vals)
stdev = statistics.stdev(window_vals)
if stdev == 0:
continue
z_score = abs(values[i] - mean) / stdev
if z_score > z_threshold:
spikes.append({
'index': i,
'value': values[i],
'z_score': round(z_score, 2),
'mean': round(mean, 2)
})
return spikes
data = [22.1, 22.3, 22.0, 22.2, 22.4] * 5 + [55.0] + [22.2, 22.3] * 3
spikes = detect_spikes(data, window=10, z_threshold=3.0)
print('Spikes detected:', spikes)Pandas für die Zeitreihenanalyse
Für anspruchsvollere Analysen laden Sie die Sensordaten in einen pandas DataFrame mit einem DatetimeIndex. Pandas bietet integrierte Operationen für gleitende Fenster, Resampling und Interpolation, die sich deutlich schneller schreiben lassen als manuelle Schleifen.
import pandas as pd
from datetime import datetime, timedelta
# Create a sample time series DataFrame
base_time = datetime(2024, 1, 1, 12, 0, 0)
times = [base_time + timedelta(seconds=i*10) for i in range(20)]
values = [22.1, 22.3, None, 22.0, 22.4, 22.2, 35.0, 22.1,
22.3, 22.2, 22.5, 22.1, None, 22.4, 22.2, 22.3,
22.0, 22.1, 22.4, 22.2]
df = pd.DataFrame({'value': values}, index=pd.DatetimeIndex(times))
df.index.name = 'timestamp'
print('Shape:', df.shape)
print('Missing values:', df['value'].isna().sum())
print(df.head())Fehlende Zeitstempel verarbeiten
In Sensornetzwerken fehlen aufgrund von Verbindungsproblemen häufig Messwerte. Pandas kann Lücken erkennen und füllen: resample erstellt ein regelmäßiges Raster, interpolate füllt fehlende Werte linear oder per Forward-Fill. Protokollieren Sie immer, wie viele Werte imputiert wurden.
import pandas as pd
def fill_missing_readings(df: pd.DataFrame, freq: str = '10S') -> pd.DataFrame:
"""
df: DataFrame with DatetimeIndex and 'value' column
freq: expected sampling frequency ('10S' = 10 seconds, '1T' = 1 minute)
"""
original_count = df['value'].notna().sum()
# Resample to regular grid (introduces NaN for missing periods)
df_regular = df.resample(freq).mean()
missing_count = df_regular['value'].isna().sum()
print(f'Missing readings before fill: {missing_count}')
# Forward fill then linear interpolate
df_regular['value'] = df_regular['value'].interpolate(
method='linear', limit=5 # don't fill gaps longer than 5 periods
)
filled_count = df_regular['value'].notna().sum()
print(f'Filled {filled_count - original_count} missing values')
return df_regularResampling: Aggregation von 1 auf 5 Minuten
Beim Resampling werden Daten mit hoher Frequenz auf eine gröbere Auflösung reduziert. Dadurch verringern sich Rauschen und Speicherbedarf. Verwenden Sie resample('5T').agg(), um für jedes 5-Minuten-Fenster Minimum, Maximum, Mittelwert und Standardabweichung zu berechnen.
import pandas as pd
def resample_to_5min(df: pd.DataFrame) -> pd.DataFrame:
return df.resample('5min').agg({
'value': ['mean', 'min', 'max', 'std', 'count']
}).round(3)
# Example with 1-minute data:
times = pd.date_range('2024-01-01 12:00', periods=30, freq='1min')
import random
random.seed(42)
vals = [22.0 + random.gauss(0, 0.5) for _ in range(30)]
df_1min = pd.DataFrame({'value': vals}, index=times)
df_5min = resample_to_5min(df_1min)
print(df_5min)Trenderkennung mit linearer Regression
Steigt die Temperatur stetig an oder weist sie nur Rauschen auf? Führen Sie eine lineare Regression über das gleitende Fenster durch. Eine positive Steigung weist auf einen Aufwärtstrend hin; überschreitet die Steigung einen Schwellenwert, wird ein Alarm ausgelöst, noch bevor der Schwellenwert erreicht ist.
def detect_trend(
values: list,
slope_threshold: float = 0.1 # units per second
) -> dict:
import statistics
n = len(values)
if n < 2:
return {'trend': 'insufficient_data'}
x = list(range(n))
x_mean = statistics.mean(x)
y_mean = statistics.mean(values)
numerator = sum((xi - x_mean) * (yi - y_mean) for xi, yi in zip(x, values))
denominator = sum((xi - x_mean) ** 2 for xi in x)
slope = numerator / denominator if denominator != 0 else 0.0
return {
'slope': round(slope, 4),
'trend': 'rising' if slope > slope_threshold
else 'falling' if slope < -slope_threshold
else 'stable',
'alert': abs(slope) > slope_threshold * 2
}
readings = [22.0, 22.5, 23.0, 23.5, 24.0, 24.5, 25.0]
print(detect_trend(readings, slope_threshold=0.3))Agentenentscheidung aus einer Zeitreihe
Kombinieren Sie Spike-Erkennung, Trenderkennung und gleitenden Mittelwert zu einer umfassenden Entscheidung. Der Agent verwendet eine Regelhierarchie: Spikes lösen sofortige Aktionen aus, Trends lösen Warnungen aus und das LLM übernimmt mehrdeutige Fälle.
def analyze_sensor_window(
values: list,
topic: str
) -> dict:
if len(values) < 10:
return {'action': 'collecting_data'}
spikes = detect_spikes(values, window=10, z_threshold=3.0)
trend_info = detect_trend(values[-20:], slope_threshold=0.2)
avg = sum(values[-10:]) / 10
if spikes:
return {
'action': 'IMMEDIATE_ALERT',
'reason': f'Spike detected: {spikes[-1]["value"]} (z={spikes[-1]["z_score"]})',
'severity': 'high'
}
if trend_info['alert']:
return {
'action': 'TREND_WARNING',
'reason': f'Rapid {trend_info["trend"]} trend: slope={trend_info["slope"]}',
'severity': 'medium'
}
return {
'action': 'NORMAL',
'avg_last_10': round(avg, 2),
'trend': trend_info['trend']
}
result = analyze_sensor_window([22.0]*15 + [55.0], 'sensors/temp')
print(result)Zeitreihen in einer Datenbank speichern
Für langfristige Analysen speichern Sie Sensormesswerte dauerhaft in einer Zeitreihendatenbank. TimescaleDB (eine PostgreSQL-Erweiterung) und InfluxDB sind beliebte Optionen. Mit psycopg2 und TimescaleDB können Sie Daten mit Standard-SQL und zusätzlichen zeitbezogenen Funktionen abfragen.
import psycopg2
from datetime import datetime
# TimescaleDB connection (standard PostgreSQL connection)
conn = psycopg2.connect(
host='localhost', port=5432, dbname='iot',
user='agent', password='YOUR_DB_PASSWORD'
)
def insert_reading(topic: str, value: float, ts: datetime = None):
ts = ts or datetime.utcnow()
with conn.cursor() as cur:
cur.execute(
'INSERT INTO sensor_readings (time, topic, value) VALUES (%s, %s, %s)',
(ts, topic, value)
)
conn.commit()
def query_last_hour(topic: str) -> list:
with conn.cursor() as cur:
cur.execute(
'SELECT time, value FROM sensor_readings '
'WHERE topic=%s AND time > NOW() - INTERVAL \'1 hour\' '
'ORDER BY time ASC',
(topic,)
)
return cur.fetchall()Cooldown-Zeitraum für Alarme
Ohne einen Cooldown löst eine anhaltende Anomalie Hunderte von Alarmen pro Minute aus. Implementieren Sie einen Topic-spezifischen Cooldown: Sobald ein Alarm für ein Topic gesendet wurde, unterdrücken Sie weitere Alarme für dieses Topic N Sekunden lang.
from datetime import datetime, timedelta
class AlertCooldownManager:
def __init__(self, cooldown_seconds: int = 300):
self.cooldown_seconds = cooldown_seconds
self._last_alert: dict = {} # topic -> last alert datetime
def should_alert(self, topic: str) -> bool:
last = self._last_alert.get(topic)
if last is None:
return True
return (datetime.utcnow() - last).seconds >= self.cooldown_seconds
def mark_alerted(self, topic: str):
self._last_alert[topic] = datetime.utcnow()
def cooldown_remaining(self, topic: str) -> int:
last = self._last_alert.get(topic)
if last is None:
return 0
elapsed = (datetime.utcnow() - last).seconds
return max(0, self.cooldown_seconds - elapsed)
cooldown = AlertCooldownManager(cooldown_seconds=300)
if cooldown.should_alert('sensors/temperature'):
print('Sending alert')
cooldown.mark_alerted('sensors/temperature')
else:
print(f'Cooldown: {cooldown.cooldown_remaining("sensors/temperature")}s remaining')Wissenscheck
Worauf weist ein z-Score über 3,0 bei der Spike-Erkennung hin?
Zusammenfassung: Verarbeitung von Zeitreihendaten
Sie haben das vollständige Toolkit zur Zeitreihenverarbeitung für Agenten kennengelernt:
- Puffer für gleitende Fenster: Deque mit maxlen für speichereffizientes Streaming
- Gleitender Mittelwert: SMA und EMA zur Rauschreduzierung
- Spike-Erkennung: z-Score-Verfahren anhand von Statistiken des gleitenden Fensters
- Pandas: Resampling, Interpolation und Aggregation mit DatetimeIndex
- Trenderkennung: Steigung der linearen Regression für frühzeitige Warnungen
- Cooldown für Alarme: Wiederholte Alarme bei anhaltenden Anomalien unterdrücken
Als Nächstes: automatisierte Agentenantworten auf Sensorereignisse — Aktionswarteschlangen, Deduplizierung und Aktionsrichtlinien.
Häufig gestellte Fragen
Ist die Lektion „Zeitreihendaten in Agenten verarbeiten“ kostenlos?
Ja — der vollständige Text von „Zeitreihendaten in Agenten verarbeiten“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Zeitreihendaten in Agenten verarbeiten“?
Gleitende Fenster, Aggregation und Anomalieerkennung für gestreamte Sensordaten. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Agents zu starten?
Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.
Wie lange dauert die Lektion „Zeitreihendaten in Agenten verarbeiten“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?
Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- MQTT-Protokoll zur Agentenintegration
- Zeitreihendaten in Agenten verarbeiten
- Automatisierte Reaktion auf Sensorereignisse
- Bereitstellung leichtgewichtiger Agenten am Edge