Pemprosesan Data Siri Masa dalam Ejen
Tetingkap bergerak, pengagregatan dan pengesanan anomali pada data sensor penstriman.
Pemprosesan Data Siri Masa dalam Ejen ialah pelajaran Ejen AI percuma di CoddyKit. Ini ialah pelajaran 2 daripada 4. Anda boleh membaca keseluruhan pelajaran di bawah secara percuma — kemudian berlatih secara praktikal dalam pelayar menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7. Pelajaran ini merupakan sebahagian daripada laluan pembelajaran Ejen AI, dan kemajuan anda disegerakkan merentas web serta aplikasi CoddyKit. Kursus Ejen AI merangkumi sejumlah 4 pelajaran.
Siri Masa dalam Ejen IoT
Data sensor tiba sebagai siri masa: urutan pasangan (cap masa, nilai). Bacaan sensor mentah mengandungi hingar, jurang dan lonjakan sekali-sekala. Ejen yang bertindak berdasarkan data mentah tanpa pemprosesan sering mencetuskan amaran palsu atau terlepas peristiwa sebenar.
Pemprosesan siri masa menukarkan isyarat mentah kepada cerapan yang boleh diambil tindakan.
Membina Penimbal Tetingkap Gelongsor
Tetingkap gelongsor hanya menyimpan N bacaan terakhir dalam ingatan. Apabila tetingkap penuh, bacaan paling lama dibuang apabila bacaan terbaharu tiba. Inilah asas bagi semua analisis siri masa dalam ejen.
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()}')Purata Bergerak
Purata bergerak mudah (SMA) melicinkan hingar dengan mempuratakan N nilai terakhir. Kaedah ini mengurangkan kesan gangguan sensor individu dan menonjolkan trend asas. Gunakannya sebagai garis dasar untuk pengesanan anomali.
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:]]}')Pengesanan Lonjakan
Lonjakan ialah bacaan yang menyimpang daripada trend terkini sebanyak lebih daripada N sisihan piawai (kaedah skor-z). Ini ialah pendekatan pengesanan anomali yang paling biasa untuk data sensor.
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 untuk Analisis Siri Masa
Untuk analisis yang lebih canggih, muatkan data sensor ke dalam pandas DataFrame dengan DatetimeIndex. Pandas menyediakan operasi gelongsor, pensampelan semula dan interpolasi terbina dalam yang jauh lebih pantas untuk ditulis berbanding gelung manual.
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())Mengendalikan Cap Masa yang Hilang
Rangkaian sensor sering terlepas bacaan akibat masalah kesambungan. Pandas boleh mengesan dan mengisi jurang: resample mencipta grid tetap, manakala interpolate mengisi nilai yang hilang secara linear atau dengan pengisian ke hadapan. Sentiasa catat bilangan nilai yang diimput.
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_regularPensampelan Semula: Pengagregatan 1 Minit kepada 5 Minit
Pensampelan semula mengurangkan sampel data berfrekuensi tinggi kepada resolusi yang lebih kasar. Ini mengurangkan hingar dan keperluan storan. Gunakan resample('5T').agg() untuk mengira nilai minimum, maksimum, mean dan sisihan piawai bagi setiap tetingkap 5 minit.
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)Pengesanan Trend dengan Regresi Linear
Adakah suhu meningkat secara berterusan atau sekadar bising? Sesuaikan regresi linear pada tetingkap gelongsor. Kecerunan positif menunjukkan trend menaik; kecerunan yang melebihi ambang mencetuskan amaran sebelum ambang itu pun dicapai.
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))Keputusan Ejen daripada Siri Masa
Gabungkan pengesanan lonjakan, pengesanan trend dan purata bergerak untuk membuat keputusan gabungan. Ejen menggunakan hierarki peraturan: lonjakan mencetuskan tindakan serta-merta, trend mencetuskan amaran, dan LLM mengendalikan kes yang samar.
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)Menyimpan Siri Masa dalam Pangkalan Data
Untuk analisis jangka panjang, simpan bacaan sensor dalam pangkalan data siri masa. TimescaleDB (sambungan PostgreSQL) dan InfluxDB ialah pilihan yang popular. Menggunakan psycopg2 dengan TimescaleDB membolehkan Anda membuat pertanyaan data menggunakan SQL standard serta fungsi khusus masa.
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()Tempoh Bertenang Amaran
Tanpa tempoh bertenang, anomali berterusan boleh mencetuskan ratusan amaran seminit. Laksanakan tempoh bertenang bagi setiap topik: setelah amaran dihantar untuk sesuatu topik, tahan amaran seterusnya bagi topik itu selama N saat.
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')Semakan Pengetahuan
Apakah yang ditunjukkan oleh skor-z yang melebihi 3.0 apabila digunakan untuk pengesanan lonjakan?
Imbas Kembali: Pemprosesan Data Siri Masa
Anda telah merangkumi keseluruhan himpunan alat pemprosesan siri masa untuk ejen:
- Penimbal tetingkap gelongsor: deque dengan maxlen untuk penstriman yang menjimatkan ingatan
- Purata bergerak: SMA dan EMA untuk pengurangan hingar
- Pengesanan lonjakan: kaedah skor-z terhadap statistik tetingkap gelongsor
- Pandas: resample, interpolate dan aggregate dengan DatetimeIndex
- Pengesanan trend: kecerunan regresi linear untuk amaran awal
- Tempoh bertenang amaran: menahan amaran berulang bagi anomali berterusan
Seterusnya: tindak balas ejen automatik terhadap peristiwa sensor — baris gilir tindakan, penyahduplikatan dan polisi tindakan.
Pelajari Ejen AI dengan tutor kecerdasan buatan — percuma
Tulis dan jalankan kod sebenar dalam pelayar anda, dapatkan bantuan segera daripada tutor kecerdasan buatan yang tersedia 24/7, dan sambung semula dari tempat anda berhenti di web atau dalam aplikasi.
- Kursus
- 60
- Pelajaran
- 239
Soalan Lazim
Adakah pelajaran “Pemprosesan Data Siri Masa dalam Ejen” percuma?
Ya — teks penuh “Pemprosesan Data Siri Masa dalam Ejen” boleh dibaca secara percuma di web ini. Untuk berlatih secara interaktif menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7, serta membuka kunci baki kursus Ejen AI, tingkat taraf kepada CoddyKit PRO. Kursus Ejen AI merangkumi sejumlah 4 pelajaran.
Apakah yang akan saya pelajari dalam “Pemprosesan Data Siri Masa dalam Ejen”?
Tetingkap bergerak, pengagregatan dan pengesanan anomali pada data sensor penstriman. Anda berlatih Ejen AI menggunakan kod praktikal yang dijalankan terus dalam pelayar, manakala tutor kecerdasan buatan 24/7 menjawab soalan anda semasa anda mengikuti pelajaran.
Adakah saya memerlukan pengalaman untuk memulakan Ejen AI?
Tiada pengalaman terdahulu diperlukan. Pembelajaran Ejen AI di CoddyKit disusun untuk pelajar daripada peringkat pemula hingga lanjutan, jadi anda boleh bermula di sini atau dari awal dan belajar mengikut kadar anda sendiri. Ini ialah pelajaran 2 daripada 4.
Berapa lamakah pelajaran “Pemprosesan Data Siri Masa dalam Ejen” diambil?
Kebanyakan pelajaran CoddyKit mengambil masa kira-kira 5–10 minit. Setiap pelajaran ringkas dan interaktif, jadi anda boleh membuat kemajuan secara berterusan dan menyambung tepat dari tempat anda berhenti di web atau aplikasi.
Bolehkah saya menulis dan menjalankan kod dalam pelajaran Ejen AI ini?
Ya. Setiap pelajaran Ejen AI menyertakan penyunting kod terbina dalam, jadi anda boleh menulis dan menjalankan kod sebenar terus dalam pelayar serta menerima maklum balas kecerdasan buatan serta-merta — tanpa memerlukan persediaan setempat.
Semua pelajaran dalam kursus ini
- Protokol MQTT untuk Integrasi Ejen
- Pemprosesan Data Siri Masa dalam Ejen
- Respons Automatik terhadap Peristiwa Sensor
- Pelaksanaan Ejen Ringan di Pinggir