0Pricing
AI Agents · 课时

智能体中的时间序列数据处理

对流式传感器数据执行滚动窗口、聚合和异常检测

智能体中的时间序列数据处理 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

物联网 Agent 中的时间序列

传感器数据以时间序列的形式到达：它是一系列由（时间戳、数值）组成的键值对。原始传感器读数包含噪声、间隙和偶发尖峰。未经处理就根据原始数据采取行动的智能体，往往会触发误报或错过真实事件。

时间序列处理会将原始信号转换为可执行的洞察。

构建滚动窗口缓冲区

滚动窗口只在内存中保留最近的 N 个读数。当窗口已满时，新的读数到达后，最早的读数会被丢弃。这是智能体进行所有时间序列分析的基础。

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

移动平均值

简单移动平均值（SMA）通过对最近 N 个值求平均来平滑噪声。它可以减弱单个传感器故障读数的影响，并揭示潜在趋势。请将其用作异常检测的基准。

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:]]}')

尖峰检测

尖峰是指偏离近期趋势超过 N 个标准差的读数（z 分数方法）。这是传感器数据最常用的异常检测方法。

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 进行时间序列分析

如需进行更复杂的分析，请将传感器数据加载到带有 DatetimeIndex 的 Pandas DataFrame 中。Pandas 提供内置的滚动、重采样和插值操作，比手动编写循环快得多。

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

处理缺失时间戳

由于连接问题，传感器网络经常会漏掉读数。Pandas 可以检测并填补间隙：resample 会创建规则网格，interpolate 会使用线性插值或前向填充来填补缺失值。请始终记录填补了多少个值。

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

重采样：从 1 分钟聚合到 5 分钟

重采样会将高频数据降采样到更粗的时间分辨率，从而减少噪声和存储需求。请使用 resample('5T').agg()，计算每个 5 分钟窗口内的最小值、最大值、平均值和标准差。

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)

使用线性回归检测趋势

温度是在稳定上升，还是只是存在噪声？请在滚动窗口上拟合线性回归。正斜率表示上升趋势；当斜率超过阈值时，即使温度尚未达到阈值，也会触发告警。

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 决策

结合尖峰检测、趋势检测和移动平均值，做出综合决策。智能体使用规则层级：尖峰触发立即操作，趋势触发警告，而 LLM 负责处理含义不明确的情况。

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)

将时间序列持久化到数据库

如需进行长期分析，请将传感器读数持久化到时间序列数据库中。TimescaleDB（PostgreSQL 扩展）和 InfluxDB 都是常用选择。将 psycopg2 与 TimescaleDB 配合使用，可以通过标准 SQL 以及时间专用函数查询数据。

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

告警冷却期

如果没有冷却期，持续存在的异常每分钟可能触发数百次告警。请按主题实现冷却机制：某个主题发送告警后，在接下来的 N 秒内抑制该主题的其他告警。

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

知识检查

在用于尖峰检测时，z 分数高于 3.0 表示什么？

回顾：时间序列数据处理

您已经学习了智能体所需的完整时间序列处理工具集：

  • 滚动窗口缓冲区：使用带有最大长度的双端队列，实现节省内存的流式处理
  • 移动平均值：使用 SMA 和 EMA 减少噪声
  • 尖峰检测：针对滚动窗口统计数据使用 z 分数方法
  • Pandas：结合 DatetimeIndex 使用重采样、插值和聚合
  • 趋势检测：使用线性回归斜率进行提前预警
  • 告警冷却：抑制持续异常产生的重复告警

下一部分：对传感器事件进行自动化智能体响应——操作队列、去重和操作策略。

常见问题解答

「智能体中的时间序列数据处理」课时是免费的吗？

是的 — 「智能体中的时间序列数据处理」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「智能体中的时间序列数据处理」这节课中我会学到什么？

对流式传感器数据执行滚动窗口、聚合和异常检测 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「智能体中的时间序列数据处理」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 用于智能体集成的 MQTT 协议
  2. 智能体中的时间序列数据处理
  3. 对传感器事件的自动响应
  4. 轻量级智能体的边缘部署
← 返回 AI Agents