AI Agents · Lesson

Time-Series Data Processing in Agents

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

Time-Series Data Processing in Agents is a free AI Agents lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

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()}')

Moving Average

A simple moving average (SMA) smooths out noise by averaging the last N values. It reduces the effect of individual sensor glitches and reveals the underlying trend. Use it as the baseline for anomaly detection.

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 Detection

A spike is a reading that deviates from the recent trend by more than N standard deviations (z-score method). This is the most common anomaly detection approach for sensor data.

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 for Time Series Analysis

For more sophisticated analysis, load sensor data into a pandas DataFrame with a DatetimeIndex. Pandas provides built-in rolling, resampling, and interpolation operations that are much faster to write than manual loops.

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())

Handling Missing Timestamps

Sensor networks frequently miss readings due to connectivity issues. Pandas can detect and fill gaps: resample creates a regular grid, interpolate fills missing values linearly or with forward-fill. Always log how many values were imputed.

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_regular

Resampling: 1-Minute to 5-Minute Aggregation

Resampling downsamples high-frequency data to a coarser resolution. This reduces noise and storage requirements. Use resample('5T').agg() to compute min, max, mean, and std over each 5-minute window.

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)

Trend Detection with Linear Regression

Is the temperature steadily rising or just noisy? Fit a linear regression over the rolling window. A positive slope indicates an upward trend; a slope exceeding a threshold triggers an alert before the threshold is even reached.

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))

Agent Decision From Time Series

Combine spike detection, trend detection, and moving average to make a composite decision. The agent uses a rule hierarchy: spikes trigger immediate actions, trends trigger warnings, and the LLM handles ambiguous cases.

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)

Persisting Time Series to a Database

For long-term analysis, persist sensor readings to a time-series database. TimescaleDB (PostgreSQL extension) and InfluxDB are popular choices. Using psycopg2 with TimescaleDB lets you query data with standard SQL plus time-specific functions.

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()

Alerting Cooldown Period

Without a cooldown, a sustained anomaly triggers hundreds of alerts per minute. Implement a per-topic cooldown: once an alert is sent for a topic, suppress further alerts for that topic for N seconds.

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')

Knowledge Check

What does a z-score above 3.0 indicate when used for spike detection?

Recap: Time Series Data Processing

You covered the full time series processing toolkit for agents:

Rolling window buffer: deque with maxlen for memory-efficient streaming
Moving average: SMA and EMA for noise reduction
Spike detection: z-score method against rolling window statistics
Pandas: resample, interpolate, and aggregate with DatetimeIndex
Trend detection: linear regression slope for early warning
Alerting cooldown: suppress repeated alerts for sustained anomalies

Next: automated agent responses to sensor events — action queues, deduplication, and action policies.

Frequently asked questions

Is the “Time-Series Data Processing in Agents” lesson free?

Yes — the full text of “Time-Series Data Processing in Agents” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.

What will I learn in “Time-Series Data Processing in Agents”?

Rolling windows, aggregation, and anomaly detection on streaming sensor data. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Agents?

No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Time-Series Data Processing in Agents” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Agents lesson?

Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

← Back to AI Agents