Procesamiento de datos de series temporales en agentes
Ventanas móviles, agregación y detección de anomalías en datos de sensores transmitidos en tiempo real.
Procesamiento de datos de series temporales en agentes es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 2 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
Series temporales en agentes IoT
Los datos de los sensores llegan como una serie temporal: una secuencia de pares (marca de tiempo, valor). Las lecturas sin procesar de los sensores contienen ruido, brechas y picos ocasionales. Los agentes que actúan sobre datos sin procesar y sin procesarlos suelen activar falsas alarmas o no detectar eventos reales.
El procesamiento de series temporales transforma las señales sin procesar en información útil para tomar decisiones.
Construcción de un búfer de ventana deslizante
Una ventana deslizante conserva en memoria solo las últimas N lecturas. Cuando la ventana está llena, la lectura más antigua se elimina al llegar la más reciente. Esta es la base de todo análisis de series temporales en agentes.
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()}')Promedio móvil
Un promedio móvil simple (SMA) reduce el ruido al promediar los últimos N valores. Disminuye el efecto de los fallos puntuales de los sensores y revela la tendencia subyacente. Úselo como referencia para detectar anomalías.
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:]]}')Detección de picos
Un pico es una lectura que se desvía de la tendencia reciente en más de N desviaciones estándar (método de puntuación z). Este es el enfoque más común para detectar anomalías en datos de sensores.
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 para el análisis de series temporales
Para realizar análisis más sofisticados, cargue los datos de los sensores en un DataFrame de pandas con un DatetimeIndex. Pandas proporciona operaciones integradas de ventana deslizante, remuestreo e interpolación, que son mucho más rápidas de escribir que los bucles manuales.
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())Gestión de marcas de tiempo ausentes
Las redes de sensores suelen omitir lecturas debido a problemas de conectividad. Pandas puede detectar y rellenar las brechas: resample crea una cuadrícula regular y interpolate rellena los valores ausentes de forma lineal o mediante relleno hacia delante. Registre siempre cuántos valores se imputaron.
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_regularRemuestreo: agregación de 1 minuto a 5 minutos
El remuestreo reduce la frecuencia de los datos de alta frecuencia a una resolución más amplia. Esto disminuye el ruido y los requisitos de almacenamiento. Use resample('5T').agg() para calcular el mínimo, el máximo, la media y la desviación estándar en cada ventana de 5 minutos.
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)Detección de tendencias con regresión lineal
¿La temperatura aumenta de forma constante o solo presenta ruido? Ajuste una regresión lineal sobre la ventana deslizante. Una pendiente positiva indica una tendencia ascendente; una pendiente que supera un umbral activa una alerta incluso antes de alcanzar dicho umbral.
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))Decisión del agente a partir de series temporales
Combine la detección de picos, la detección de tendencias y el promedio móvil para tomar una decisión compuesta. El agente usa una jerarquía de reglas: los picos activan acciones inmediatas, las tendencias activan advertencias y el LLM gestiona los casos ambiguos.
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)Persistencia de series temporales en una base de datos
Para realizar análisis a largo plazo, persista las lecturas de los sensores en una base de datos de series temporales. TimescaleDB (extensión de PostgreSQL) e InfluxDB son opciones populares. El uso de psycopg2 con TimescaleDB permite consultar los datos mediante SQL estándar y funciones específicas para series temporales.
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()Período de enfriamiento de las alertas
Sin un período de enfriamiento, una anomalía persistente activa cientos de alertas por minuto. Implemente un período de enfriamiento por topic: una vez enviada una alerta para un topic, suprima las alertas posteriores de ese topic durante N segundos.
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')Comprobación de conocimientos
¿Qué indica una puntuación z superior a 3,0 cuando se utiliza para detectar picos?
Repaso: procesamiento de datos de series temporales
Ha cubierto todo el conjunto de herramientas para procesar series temporales en agentes:
- Búfer de ventana deslizante: deque con maxlen para consumir flujos con un uso eficiente de la memoria
- Promedio móvil: SMA y EMA para reducir el ruido
- Detección de picos: método de puntuación z aplicado a las estadísticas de la ventana deslizante
- Pandas: resample, interpolate y aggregate con DatetimeIndex
- Detección de tendencias: pendiente de regresión lineal para obtener alertas tempranas
- Período de enfriamiento de alertas: supresión de alertas repetidas para anomalías persistentes
A continuación: respuestas automatizadas de los agentes a eventos de sensores — colas de acciones, deduplicación y políticas de acción.
Preguntas frecuentes
¿La lección «Procesamiento de datos de series temporales en agentes» es gratis?
Sí — el texto completo de «Procesamiento de datos de series temporales en agentes» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Procesamiento de datos de series temporales en agentes»?
Ventanas móviles, agregación y detección de anomalías en datos de sensores transmitidos en tiempo real. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 2 de 4.
¿Cuánto tiempo toma la lección «Procesamiento de datos de series temporales en agentes»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Protocolo MQTT para la integración de agentes
- Procesamiento de datos de series temporales en agentes
- Respuesta automatizada a eventos de sensores
- Despliegue en el edge de agentes ligeros