에이전트의 시계열 데이터 처리
스트리밍 센서 데이터에서 이동 구간, 집계, 이상 탐지를 수행합니다.
에이전트의 시계열 데이터 처리은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
IoT Agent의 시계열
센서 데이터는 시계열로 도착합니다. 시계열은 (타임스탬프, 값) 쌍의 연속입니다. 원시 센서 판독값에는 잡음과 누락 구간, 간헐적인 급증이 포함됩니다. 처리 없이 원시 데이터에 따라 동작하는 Agent는 잘못된 경보를 발생시키거나 실제 이벤트를 놓치는 경우가 많습니다.
시계열 처리는 원시 신호를 실행 가능한 인사이트로 변환합니다.
롤링 윈도 버퍼 구축
롤링 윈도는 메모리에 마지막 N개의 판독값만 유지합니다. 윈도가 가득 차면 새 판독값이 들어올 때 가장 오래된 판독값이 제거됩니다. 이는 Agent의 모든 시계열 분석을 위한 기반입니다.
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개 값의 평균을 내 잡음을 완화합니다. 개별 센서 오류의 영향을 줄이고 underlying 추세를 드러냅니다. 이상 감지를 위한 기준선으로 사용하십시오.
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)시계열 분석을 위한 판다스
더 정교한 분석을 위해 센서 데이터를 DatetimeIndex가 있는 판다스 DataFrame으로 불러오십시오. 판다스는 직접 반복문을 작성하는 것보다 훨씬 빠르게 작성할 수 있는 롤링, 리샘플링 및 보간 작업을 기본으로 제공합니다.
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())누락된 타임스탬프 처리
센서 네트워크에서는 연결 문제로 인해 판독값이 자주 누락됩니다. 판다스는 누락 구간을 감지하고 채울 수 있습니다. 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 결정
급증 감지, 추세 감지 및 이동 평균을 결합해 종합적인 결정을 내리십시오. 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가 널리 사용됩니다. PostgreSQL 연결 라이브러리를 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')지식 확인
급증 감지에 사용할 때 3.0보다 큰 z-점수는 무엇을 나타냅니까?
복습: 시계열 데이터 처리
Agent를 위한 전체 시계열 처리 도구 모음을 학습했습니다.
- 롤링 윈도 버퍼: 메모리를 효율적으로 사용하는 스트리밍을 위한 maxlen이 설정된 덱
- 이동 평균: 잡음 감소를 위한 SMA와 EMA
- 급증 감지: 롤링 윈도 통계에 적용하는 z-점수 방식
- 판다스: DatetimeIndex를 이용한 resample, interpolate 및 집계
- 추세 감지: 조기 경보를 위한 선형 회귀 기울기
- 경보 쿨다운: 지속되는 이상 상태에 대한 반복 경보 억제
다음 주제: 센서 이벤트에 대한 자동 Agent 응답 — 동작 큐, 중복 제거 및 동작 정책
자주 묻는 질문
“에이전트의 시계열 데이터 처리” 강의는 무료인가요?
네 — “에이전트의 시계열 데이터 처리” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“에이전트의 시계열 데이터 처리”에서 뭘 배우나요?
스트리밍 센서 데이터에서 이동 구간, 집계, 이상 탐지를 수행합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.
“에이전트의 시계열 데이터 처리” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 에이전트 통합을 위한 MQTT 프로토콜
- 에이전트의 시계열 데이터 처리
- 센서 이벤트에 대한 자동 응답
- 경량 에이전트의 엣지 배포