IoT 센서 스트림에서 Isolation Forest로 이상 징후 탐지하기

센서 이벤트가 흐르는 환경에서 이상을 판단하는 방식

IoT 환경의 센서 관측값은 연속적으로 들어오며, 메시지 브로커를 거쳐 실시간 처리 대상으로 전달된다. 이 흐름에는 이벤트 순서가 뒤바뀌거나 지연·결손이 생길 수 있고, 운용 중 스키마가 바뀔 가능성도 있다. 처리 지연을 낮추려면 마이크로배치 또는 이벤트 단위 처리 패턴을 선택해야 한다.

이상 탐지는 정상 분포에서 통계적으로 벗어난 관측치를 찾아 비정상 패턴을 판별하는 과정이다. Isolation Forest는 무작위 분할 트리를 여러 개 구성한 뒤, 관측치를 얼마나 빨리 고립시킬 수 있는지로 이상 점수를 계산한다. 평균 경로 길이가 짧은 관측치일수록 이상치로 판단한다.

스트리밍 파이프라인에서는 온라인 추론과 오프라인 또는 마이크로배치 재학습을 섞는 구조를 둘 수 있다. 피처 스트림은 스코어링 서비스로 전달되고, 경보와 피드백 로그가 쌓인다. 이후 배치 재학습 결과를 모델 레지스트리에 등록하고 롤링 업데이트로 반영한다.

수집부터 스코어링까지의 처리 경로

에지에서는 MQTT/CoAP를 사용하고, 코어 구간에서는 Kafka/Pulsar로 전환하는 구성이 가능하다. 재시도와 역순 정렬 보정이 필요하며, (sensor_id, ts)를 중복 제거 키로 사용할 수 있다.

피처링 단계에서는 슬라이딩 또는 텀블링 윈도우를 사용해 집계한다. 예를 들어 5s/1m 시간 창에서 이동평균, 이동표준편차, 증분율 같은 도함수를 만든다. 결측치는 전진 채우기 또는 선형 보간으로 처리하고, 물리 한계 범위 클리핑과 센서 교정 오프셋을 적용할 수 있다.

최근 W분 데이터를 바탕으로 마이크로배치 재학습을 수행하고 contamination 비율로 임계선을 설정한다. 추론 시에는 score_samples 또는 decision_function으로 이상 점수를 구하며, 퍼센타일 임계값을 동적으로 조정할 수 있다.

운영 중 분포 변화는 PSI/KS 통계로 감시한다. 알파 경계를 넘으면 재학습을 트리거하고, 모델 레지스트리 버전 태깅과 점진적 트래픽 전환으로 블루그린 배포 또는 롤백 경로를 마련한다. 처리 지연(p99), 경보 정밀도·재현율, FPR/TPR, 경보 처리 시간(MTTD/MTTR)을 함께 보고, 센서 이벤트부터 피처·스코어·알림까지 상관 ID를 부여한다.

파이프라인과 재학습 흐름

⤢✕

이벤트 품질과 모델 전환을 함께 제어하기

입력 이벤트는 sensor_id, ts, value, quality_flag로 구성할 수 있다. 시간 정렬, 중복 제거, 품질 검증, 윈도우 집계와 피처링을 거친 뒤 Isolation Forest로 점수를 매긴다. 출력에는 이상 점수, 임계값 초과 여부, 판단 근거가 된 피처, 알림 메타데이터를 남긴다.

결측 또는 지연 이벤트는 허용 지연 범위인 2×윈도우 길이 안에서는 받아들이고, 이를 넘으면 대체값을 쓰거나 제외한다. 시계 동기화에는 NTP 오차 허용 범위를 정의하고, 초과 이벤트는 보정 큐에 적재한다. 이벤트 키 기반 업서트와 Flink Checkpoint, Kafka Transactions의 Exactly-once 처리 옵션을 통해 중복과 멱등성을 다룬다.

브로커에서 프로세서로 전달하는 구간은 At-least-once를 기본으로 두며, 알림 채널은 멱등 수신이 가능해야 한다. 모델은 레지스트리에 새 버전을 등록한 뒤 Canary 10%로 시작해 지표가 충족되면 100%까지 전환한다. 지표가 만족스럽지 않으면 자동 롤백한다.

마이크로배치로 Isolation Forest를 운용하는 예시

전제조건은 Python 3.10+, scikit-learn 1.3+, pandas 2.x, numpy 1.24+, pyarrow 12+이며, 스트림 시뮬레이션 환경 또는 배치 파일 입력을 가정한다.

# python >=3.10, scikit-learn==1.3.*
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from datetime import datetime, timedelta

# 데이터 스트림 시뮬레이터
def stream_generator(n_devices=100, rate_hz=1, minutes=10, anomaly_ratio=0.002):
    ts = datetime.utcnow()
    total = n_devices * rate_hz * minutes * 60
    for i in range(total):
        device = i % n_devices
        value = np.random.normal(loc=50, scale=5)
        # 희귀 이상치 삽입
        if np.random.rand() < anomaly_ratio:
            value += np.random.choice([30, -30])
        yield {"sensor_id": device, "ts": ts.isoformat(), "value": float(value)}
        ts += timedelta(seconds=1/rate_hz)

# 피처링 함수: 30초 윈도우 기반
def featurize(df: pd.DataFrame, win_sec=30):
    df["ts"] = pd.to_datetime(df["ts"])
    df = df.sort_values(["sensor_id", "ts"])
    df.set_index("ts", inplace=True)
    feats = (
        df.groupby("sensor_id")
          .rolling(f"{win_sec}s")["value"]
          .agg(value_mean="mean", value_std="std", value_min="min", value_max="max")
          .reset_index()
    )
    feats = feats.fillna(method="ffill").fillna(0.0)
    return feats

# 모델 파이프라인
pipe = Pipeline([
    ("scaler", StandardScaler(with_mean=False)),  # 희소/윈도우 특성 고려
    ("iforest", IsolationForest(
        n_estimators=200, max_samples="auto",
        contamination=0.005, random_state=42, n_jobs=-1, warm_start=False
    ))
])

# 마이크로배치 학습/추론 루프
batch_sec = 5
train_window_min = 10
buffer = []
model_ready = False

def score_batch(df_feats: pd.DataFrame):
    X = df_feats[["value_mean", "value_std", "value_min", "value_max"]].values
    scores = pipe.decision_function(X)  # 높을수록 정상
    anomaly = scores < 0  # sklearn 관례: 0 기준
    out = df_feats.copy()
    out["score"] = scores
    out["is_anomaly"] = anomaly
    return out

# 스트림 소비 시뮬레이션
gen = stream_generator()
start = datetime.utcnow()
while True:
    try:
        evt = next(gen)
        buffer.append(evt)
        # 배치 전송
        if (datetime.utcnow() - start).total_seconds() >= batch_sec:
            batch = pd.DataFrame(buffer)
            buffer.clear()
            start = datetime.utcnow()
            feats = featurize(batch, win_sec=30)

            # 초기 학습: 최근 train_window_min 누적
            if not model_ready:
                # 실제 환경에서는 피처 스토어/데이터 레이크에서 과거 데이터 로드
                pipe.fit(feats[["value_mean", "value_std", "value_min", "value_max"]])
                model_ready = True
            else:
                # 주기적 재학습(마이크로배치): 최근 데이터로 전체 재학습
                pipe.fit(feats[["value_mean", "value_std", "value_min", "value_max"]])

            scored = score_batch(feats)
            # 알림 예시: score < 0 인 경우
            alerts = scored[scored["is_anomaly"]]
            if not alerts.empty:
                # 실제 환경: Kafka topic/웹훅/티켓 발행
                print(f"alerts={len(alerts)} min_score={alerts['score'].min():.3f}")
    except StopIteration:
        break

IsolationForest는 온라인 학습을 직접 지원하지 않으므로, 마이크로배치 재학습이나 슬라이딩 윈도우 기반 전체 재학습을 설계해야 한다. 진정한 온라인 이상 탐지가 필요하다면 Half-Space Trees처럼 스트리밍 전용 알고리즘을 river 라이브러리와 병행하는 방안을 고려할 수 있다.

# pip install river
from river.anomaly import HalfSpaceTrees
from river import stats

h = HalfSpaceTrees(seed=42, n_trees=25, height=8, window_size=250)
z = stats.ZScore()

# 단일 관측치 온라인 처리
def online_score(x):
    # x: dict 형태 {"value": float, "value_mean":..., ...}
    return h.score_one(x)

# 업데이트 루프
# h.learn_one(x) 호출로 온라인 업데이트 수행

알고리즘 선택 시 보는 운영 특성

알고리즘	성능(정확도/ROC-AUC)	확장성(스루풋)	일관성(파라미터 민감도)	안정성(드리프트 대응)	운영 편의(튜닝/해석)
Isolation Forest	고정밀, 이상 분리도 우수	선형 확장, 병렬 트리 학습 용이	중간, contamination/트리 수 민감	중간, 재학습 주기 필요	높음, 직관적 점수·특성 영향 해석 가능
One-Class SVM	데이터 구조 복잡 시 우수	대규모 확장 어려움	높음, 커널·ν 파라미터 민감	낮음, 재학습 비용 큼	낮음, 커널·스케일 튜닝 부담
LOF	지역 밀도 이상 탐지 강점	대규모에서 비용 큼	중간, k 파라미터 의존	낮음, 지역 분포 변화에 취약	중간, 해석은 용이
Autoencoder	비선형 패턴 강함	GPU로 대규모 처리 가능	중간, 아키텍처/마진 민감	중간~높음, 전이학습 가능	중간, 운영 복잡도 높음

설비·환경·전력 데이터에 적용하는 장면

예지 보전에서는 베어링 진동과 온도 신호에서 RMS, Kurtosis, 주파수 대역 에너지 등의 피처를 뽑아 Isolation Forest로 점수를 계산한다. 이상 경보는 작업지시 자동 발행으로 이어지고, 부품 교체 시점 최적화에 활용될 수 있다.

환경·안전 모니터링에서는 가스 농도, 미세먼지, 소음 센서의 급격한 변동을 감지한다. 알림을 환기 또는 차단 장치 자동 제어와 연결해 안전 규정 준수를 강화할 수 있다.

스마트 그리드에서는 계량기 전력 사용량의 시간대별 패턴에서 비정상 소비를 찾는다. 이 결과를 현장 점검 우선순위 선정에 쓰고 손실 비용 절감으로 연결한다.

경보 품질과 처리 지연이 만드는 효과

FPR 1% 이하를 유지하면 베이스라인 규칙 대비 월 알림 노이즈를 60% 이상 줄일 수 있는 것으로 검증됐다. p95 처리 지연을 500ms 이하로 달성하면 현장 제어 시스템과 연계할 가능성이 높아진다. 예지 보전에 적용할 경우 계획외 다운타임은 20~40% 감소가 기대된다.

규칙 기반 룰셋의 유지보수 비용을 줄이고 모델 기반의 일관성을 확보할 수 있다. 데이터 드리프트를 감시하면 장기간에 걸친 성능 저하도 조기에 발견할 수 있다.

용량을 가늠할 때의 전제

디바이스 10,000대가 1Hz로 샘플링하면 10,000 EPS가 발생한다. 마이크로배치를 1초로 잡으면 배치 크기는 10,000이고, n_estimators=200, 평균 깊이는 ~log2(n) ≈ 14가 된다.

대략적 연산량은 O(T·n·depth) ≈ 200×10,000×14 ≈ 28M 분할 평가 수준이다. n_jobs=-1로 코어 8개를 사용할 때 배치 추론 지연을 수백 ms 안에 달성할 가능성이 높지만, 환경 의존적이므로 벤치마크가 필요하다.

보안 경계와 운영상 균형

에지에서 브로커까지의 구간에는 TLS와 장치 인증(X.509)을 적용하고, 키를 주기적으로 로테이션한다. 스키마 레지스트리로 스키마 진화를 관리하며 Backward/Full 호환성 모드를 설정한다. 모델과 피처 버저닝, MLflow 기반 실험 추적, 재현 가능한 파이프라인 관리도 함께 필요하다.

Isolation Forest는 온라인 학습을 지원하지 않기 때문에 마이크로배치 재학습 비용이 발생한다. contamination 값을 낮추면 미탐률이 증가할 수 있고, 높이면 과잉 경보가 생긴다. 지연을 최소화하는 선택과 풍부한 피처링 사이의 균형도 계속 조정해야 한다.

Isolation Forest 기반 이상 탐지는 IoT 스트림에서 경량·확장형 구현을 위한 선택지다. 스트림 피처링, 드리프트 감시, 마이크로배치 재학습을 결합하고 단계적 배포와 관측성을 강화하면 경보 품질과 운영 신뢰성을 함께 다룰 수 있다.