AI Agents · Lesson

Time-Series Data Processing in Agents

Rolling windows, aggregation, and anomaly detection on streaming sensor data.

Time Series in IoT Agents

Sensor data arrives as a time series: a sequence of (timestamp, value) pairs. Raw sensor readings contain noise, gaps, and occasional spikes. Agents that act on raw data without processing often trigger false alarms or miss real events.

Time series processing transforms raw signals into actionable insights.

Building a Rolling Window Buffer

A rolling window keeps only the last N readings in memory. When the window is full, the oldest reading is dropped as the newest arrives. This is the foundation for all time series analysis in 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()}')

All lessons in this course

← Back to AI Agents