Traitement des séries temporelles par les agents
Fenêtres glissantes, agrégation et détection d’anomalies sur des données de capteurs en continu.
Traitement des séries temporelles par les agents est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 2 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.
Séries temporelles dans les agents IoT
Les données des capteurs arrivent sous la forme d’une série temporelle : une suite de paires (horodatage, valeur). Les relevés bruts des capteurs contiennent du bruit, des lacunes et parfois des pics. Les agents qui agissent sur des données brutes sans les traiter déclenchent souvent de fausses alertes ou ne détectent pas certains événements réels.
Le traitement des séries temporelles transforme les signaux bruts en informations exploitables.
Créer un tampon à fenêtre glissante
Une fenêtre glissante conserve en mémoire uniquement les N derniers relevés. Lorsque la fenêtre est pleine, le relevé le plus ancien est supprimé à l’arrivée du plus récent. C’est la base de toute analyse de séries temporelles dans les agents.
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()}')Moyenne mobile
Une moyenne mobile simple (SMA) lisse le bruit en calculant la moyenne des N dernières valeurs. Elle réduit l’effet des anomalies ponctuelles des capteurs et fait ressortir la tendance sous-jacente. Utilisez-la comme référence pour détecter les anomalies.
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:]]}')Détection des pics
Un pic est un relevé qui s’écarte de la tendance récente de plus de N écarts-types (méthode du score-z). Il s’agit de l’approche la plus courante pour détecter les anomalies dans les données de capteurs.
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 pour l’analyse des séries temporelles
Pour une analyse plus sophistiquée, chargez les données des capteurs dans un DataFrame pandas avec un DatetimeIndex. Pandas fournit des opérations intégrées de fenêtrage glissant, de rééchantillonnage et d’interpolation, bien plus rapides à écrire que des boucles manuelles.
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())Gérer les horodatages manquants
Les réseaux de capteurs manquent fréquemment des relevés en raison de problèmes de connectivité. Pandas peut détecter et combler les lacunes : resample crée une grille régulière, tandis que interpolate comble les valeurs manquantes par interpolation linéaire ou par propagation vers l’avant. Consignez toujours le nombre de valeurs imputées.
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_regularRééchantillonnage : agrégation d’une minute à cinq minutes
Le rééchantillonnage réduit la fréquence des données à haute fréquence pour obtenir une résolution plus grossière. Cela réduit le bruit et les besoins de stockage. Utilisez resample('5T').agg() pour calculer le minimum, le maximum, la moyenne et l’écart-type sur chaque fenêtre de cinq minutes.
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)Détecter les tendances par régression linéaire
La température augmente-t-elle régulièrement ou les données sont-elles simplement bruitées ? Ajustez une régression linéaire sur la fenêtre glissante. Une pente positive indique une tendance à la hausse ; une pente qui dépasse un seuil déclenche une alerte avant même que le seuil ne soit atteint.
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))Décision de l’agent à partir d’une série temporelle
Combinez la détection des pics, la détection des tendances et la moyenne mobile pour prendre une décision composite. L’agent utilise une hiérarchie de règles : les pics déclenchent des actions immédiates, les tendances déclenchent des avertissements et le LLM traite les cas ambigus.
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)Enregistrer les séries temporelles dans une base de données
Pour les analyses à long terme, enregistrez les relevés des capteurs dans une base de données de séries temporelles. TimescaleDB (extension de PostgreSQL) et InfluxDB sont des choix courants. L’utilisation de psycopg2 avec TimescaleDB permet d’interroger les données avec du SQL standard ainsi qu’avec des fonctions spécifiques au temps.
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()Période de temporisation des alertes
Sans temporisation, une anomalie persistante déclenche des centaines d’alertes par minute. Implémentez une temporisation par rubrique : lorsqu’une alerte est envoyée pour une rubrique, bloquez les alertes supplémentaires pour cette rubrique pendant N secondes.
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')Vérification des connaissances
Qu’indique un score-z supérieur à 3,0 lorsqu’il est utilisé pour détecter des pics ?
Récapitulatif : traitement des données de séries temporelles
Vous avez étudié l’ensemble des outils de traitement des séries temporelles pour les agents :
- Tampon à fenêtre glissante : deque avec maxlen pour un flux économe en mémoire
- Moyenne mobile : SMA et EMA pour réduire le bruit
- Détection des pics : méthode du score-z appliquée aux statistiques de la fenêtre glissante
- Pandas : rééchantillonnage, interpolation et agrégation avec DatetimeIndex
- Détection des tendances : pente d’une régression linéaire pour l’alerte précoce
- Temporisation des alertes : suppression des alertes répétées pour les anomalies persistantes
Ensuite : réponses automatisées des agents aux événements des capteurs — files d’actions, déduplication et règles d’action.
Questions Fréquemment Posées
La leçon « Traitement des séries temporelles par les agents » est-elle gratuite ?
Oui — le texte complet de « Traitement des séries temporelles par les agents » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Traitement des séries temporelles par les agents » ?
Fenêtres glissantes, agrégation et détection d’anomalies sur des données de capteurs en continu. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Agents ?
Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 2 sur 4.
Combien de temps prend la leçon « Traitement des séries temporelles par les agents » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?
Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Protocole MQTT pour l’intégration d’agents
- Traitement des séries temporelles par les agents
- Réponse automatisée aux événements des capteurs
- Déploiement en périphérie d’agents légers