Bearbetning av tidsseriedata i agenter
Rullande fönster, aggregering och avvikelsedetektering i strömmande sensordata.
Bearbetning av tidsseriedata i agenter är en gratis lektion i AI-agenter på CoddyKit. Detta är lektion 2 av 4. Ni kan läsa hela lektionen gratis nedan och sedan öva praktiskt i webbläsaren med en inbyggd kodredigerare och en AI-handledare som är tillgänglig dygnet runt. Den ingår i lärvägen för AI-agenter, och Era framsteg synkroniseras mellan webben och CoddyKit-appen. Kursen i AI-agenter innehåller totalt 4 lektioner.
Tidsserier i IoT-agenter
Sensor data anländer som en tidsserie: en sekvens av par med (tidsstämpel, värde). Råa sensormätvärden innehåller brus, luckor och enstaka spikar. Agenter som agerar utifrån rådata utan bearbetning utlöser ofta falsklarm eller missar verkliga händelser.
Bearbetning av tidsserier omvandlar råa signaler till användbara insikter.
Bygga en buffert med rullande fönster
Ett rullande fönster behåller endast de senaste N mätvärdena i minnet. När fönstret är fullt tas det äldsta mätvärdet bort när det nyaste anländer. Detta är grunden för all tidsserieanalys i agenter.
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()}')Glidande medelvärde
Ett enkelt glidande medelvärde (SMA) jämnar ut brus genom att beräkna medelvärdet av de senaste N värdena. Det minskar effekten av enskilda sensorstörningar och synliggör den underliggande trenden. Använd det som baslinje för anomalidetektering.
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:]]}')Spikdetektering
En spik är ett mätvärde som avviker från den senaste trenden med mer än N standardavvikelser (z-score-metoden). Detta är den vanligaste metoden för anomalidetektering i sensordata.
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 tidsserieanalys
För mer avancerad analys kan ni läsa in sensordata i en pandas DataFrame med ett DatetimeIndex. Pandas har inbyggda operationer för rullande fönster, omsampling och interpolering som går mycket snabbare att skriva än manuella loopar.
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())Hantera saknade tidsstämplar
Sensor nätverk saknar ofta mätvärden på grund av anslutningsproblem. Pandas kan identifiera och fylla luckor: resample skapar ett regelbundet rutnät, interpolate fyller i saknade värden linjärt eller med framåtfyllning. Logga alltid hur många värden som imputerades.
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_regularOmsampling: aggregering från 1 minut till 5 minuter
Omsampling samplar ned data med hög frekvens till en grövre upplösning. Detta minskar brus och lagringsbehov. Använd resample('5T').agg() för att beräkna min, max, medelvärde och std över varje femminutersfönster.
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)Trenddetektering med linjär regression
Stiger temperaturen stadigt eller är den bara brusig? Anpassa en linjär regression över det rullande fönstret. En positiv lutning visar en uppåtgående trend; en lutning som överskrider ett tröskelvärde utlöser ett larm innan tröskeln ens har nåtts.
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))Agentbeslut utifrån tidsserier
Kombinera spikdetektering, trenddetektering och glidande medelvärde för att fatta ett sammansatt beslut. Agenten använder en regelhierarki: spikar utlöser omedelbara åtgärder, trender utlöser varningar och LLM hanterar tvetydiga fall.
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)Lagra tidsserier i en databas
För långsiktig analys bör ni lagra sensormätvärden i en tidsseriedatabas. TimescaleDB (ett PostgreSQL-tillägg) och InfluxDB är populära alternativ. Med psycopg2 och TimescaleDB kan ni fråga efter data med vanlig SQL samt tidspecifika funktioner.
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()Nedkylningsperiod för larm
Utan en nedkylningsperiod utlöser en ihållande anomali hundratals larm per minut. Implementera en nedkylningsperiod per topic: när ett larm har skickats för en topic ska ytterligare larm för den topicen undertryckas i N sekunder.
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')Kunskapskontroll
Vad visar ett z-score över 3,0 när det används för spikdetektering?
Sammanfattning: bearbetning av tidsseriedata
Ni har gått igenom hela verktygslådan för bearbetning av tidsserier i agenter:
- Buffert med rullande fönster: deque med maxlen för minnessnål strömning
- Glidande medelvärde: SMA och EMA för brusreducering
- Spikdetektering: z-score-metoden mot statistik för det rullande fönstret
- Pandas: resample, interpolate och aggregering med DatetimeIndex
- Trenddetektering: lutning från linjär regression för tidig varning
- Nedkylningsperiod för larm: undertryck upprepade larm vid ihållande anomalier
Nästa steg: automatiserade agentsvar på sensorhändelser — åtgärdsköer, deduplicering och åtgärdspolicyer.
Lär dig AI-agenter med en AI-lärare – gratis
Skriv och kör riktig kod i webbläsaren, få omedelbar hjälp av en AI-lärare dygnet runt och fortsätt där du slutade – på webben eller i appen.
- Kurser
- 60
- Lektioner
- 239
Vanliga frågor
Är lektionen ”Bearbetning av tidsseriedata i agenter” gratis?
Ja – hela texten till ”Bearbetning av tidsseriedata i agenter” kan läsas gratis här på webben. Om Ni vill öva interaktivt med en inbyggd kodredigerare och en AI-handledare som är tillgänglig dygnet runt och låsa upp resten av kursen i AI-agenter, kan Ni uppgradera till CoddyKit PRO. Kursen i AI-agenter innehåller totalt 4 lektioner.
Vad lär jag mig i ”Bearbetning av tidsseriedata i agenter”?
Rullande fönster, aggregering och avvikelsedetektering i strömmande sensordata. Ni övar på AI-agenter med praktisk kod som körs direkt i webbläsaren, medan en AI-handledare som är tillgänglig dygnet runt svarar på Era frågor under lektionen.
Behöver jag någon erfarenhet för att börja lära mig AI-agenter?
Du behöver inga förkunskaper. Utbildningen i AI-agenter på CoddyKit är upplagd för allt från nybörjare till avancerade elever, så att du kan börja här eller från början och gå fram i din egen takt. Detta är lektion 2 av 4.
Hur lång tid tar lektionen ”Bearbetning av tidsseriedata i agenter”?
De flesta CoddyKit-lektioner tar cirka 5–10 minuter. Varje lektion är kort och interaktiv, så att du gör stadiga framsteg och kan fortsätta precis där du slutade – på webben eller i appen.
Kan jag skriva och köra kod i den här AI-agenter-lektionen?
Ja. Varje AI-agenter-lektion innehåller en inbyggd kodredigerare, så att du kan skriva och köra riktig kod direkt i webbläsaren och få omedelbar AI-feedback – utan lokal installation.
Alla lektioner i den här kursen
- MQTT-protokollet för agentintegration
- Bearbetning av tidsseriedata i agenter
- Automatiserade svar på sensorhändelser
- Distribuera lättviktiga agenter vid kanten