[2025 ABC 프로젝트 멘토링 8기] 3주차 - CNN으로 시계열 이상 탐지 (PyTorch)

PyTorch를 사용하여 1D CNN 오토인코더 기반 시계열 이상 탐지 베이스라인 모델을 구현합니다.
ABC프로젝트멘토링
PyTorch
Author

Beomdo Park

Published

June 8, 2025

Modified

May 26, 2026

안녕하세요, ABC 프로젝트 멘토링 8기 세 번째 기술노트입니다. 이번 주는 시계열 데이터의 ’패턴’을 학습할 수 있는 딥러닝, 그중에서도 CNN을 활용한 이상 탐지의 첫걸음을 PyTorch로 구현해 보겠습니다.

Tip이전 포스트

Week2 포스트에서 머신러닝 기반 이상 탐지 기법을 다뤘습니다. 이번 포스트는 이를 기반으로 PyTorch를 사용한 딥러닝 접근법을 소개합니다.

1. 시계열 데이터를 CNN에 입력하는 방법: 윈도잉(Windowing)

시계열 데이터를 CNN 모델에 입력하려면 연속된 데이터를 일정한 길이의 조각(window)으로 나누는 ‘슬라이딩 윈도우’ 기법이 필요합니다. 이 방법은 데이터의 시간적 패턴을 학습하는 데 유용합니다.

슬라이딩 윈도우 구현

아래는 numpy를 사용해 슬라이딩 윈도우를 구현하는 간단한 Python 함수입니다:

import numpy as np

def sliding_window(data, window_size, step_size=1):
    """시계열 데이터를 슬라이딩 윈도우로 변환"""
    n_windows = (len(data) - window_size) // step_size + 1
    return np.array([data[i:i+window_size] for i in range(0, n_windows * step_size, step_size)])

# 예제 데이터
data = np.sin(np.linspace(0, 20, 100))
windowed_data = sliding_window(data, window_size=10)
print("윈도우 형태:", windowed_data.shape)
윈도우 형태: (91, 10)

2. 기본 이상 탐지 모델: CNN 오토인코더 (Autoencoder)

오토인코더란?

오토인코더는 데이터를 압축(인코더)했다가 다시 복원(디코더)하도록 학습하는 딥러닝 모델입니다. 정상 데이터는 잘 복원되지만, 이상 데이터는 복원이 잘 되지 않아 재구성 오차가 커지는 특징을 활용합니다.

모델 구조

  • 인코더 (Encoder): Conv1D와 MaxPooling1D 층을 사용해 입력 데이터의 특징을 추출하고 압축합니다.
  • 디코더 (Decoder): ConvTranspose1D (또는 Upsample + Conv1D) 층을 사용해 데이터를 복원합니다.

PyTorch 구현

아래는 PyTorch를 사용한 간단한 1D CNN 오토인코더 모델 구현입니다:

import torch
import torch.nn as nn

class CNNAutoencoder(nn.Module):
    def __init__(self, input_shape): # input_shape: (sequence_length, num_features)
        super(CNNAutoencoder, self).__init__()
        # Encoder
        # input_shape[1]은 특성 수 (in_channels로 사용)
        self.encoder_conv1 = nn.Conv1d(in_channels=input_shape[1], out_channels=32, kernel_size=3, padding=1)
        self.encoder_relu1 = nn.ReLU()
        self.encoder_pool1 = nn.MaxPool1d(kernel_size=2, stride=2) # 시퀀스 길이 1/2로 감소
        self.encoder_conv2 = nn.Conv1d(in_channels=32, out_channels=16, kernel_size=3, padding=1)
        self.encoder_relu2 = nn.ReLU()
        self.encoder_pool2 = nn.MaxPool1d(kernel_size=2, stride=2) # 시퀀스 길이 1/4로 감소

        # Decoder
        # 인코더에서 시퀀스 길이가 1/4로 줄었으므로, 디코더에서 원래 길이로 복원
        self.decoder_conv_t1 = nn.ConvTranspose1d(in_channels=16, out_channels=16, kernel_size=4, stride=2, padding=1, output_padding=1)
        self.decoder_relu1 = nn.ReLU()
        self.decoder_conv_t2 = nn.ConvTranspose1d(in_channels=16, out_channels=32, kernel_size=3, stride=2, padding=1, output_padding=1)
        self.decoder_relu2 = nn.ReLU()
        self.decoder_conv_final = nn.Conv1d(in_channels=32, out_channels=input_shape[1], kernel_size=3, padding=1) # 원본 특성 수로 복원
        self.sigmoid = nn.Sigmoid()

    def forward(self, x):
        # Encoder
        x = self.encoder_conv1(x)
        x = self.encoder_relu1(x)
        x = self.encoder_pool1(x)
        x = self.encoder_conv2(x)
        x = self.encoder_relu2(x)
        encoded = self.encoder_pool2(x)
        
        # Decoder
        x = self.decoder_conv_t1(encoded)
        x = self.decoder_relu1(x)
        x = self.decoder_conv_t2(x)
        x = self.decoder_relu2(x)
        x = self.decoder_conv_final(x)
        decoded = self.sigmoid(x)
        return decoded

# 모델 생성 및 컴파일은 data-generation 셀 이후로 이동합니다.
# input_shape도 window_size를 사용하도록 수정됩니다.

3. 모델 학습 및 이상치 탐지

데이터 생성

Week2에서 사용한 샘플 데이터를 기반으로 정상/비정상 데이터를 생성합니다:

# numpy는 sliding_window_implementation 셀에서 이미 import 됨

# 데이터 생성
np.random.seed(42)
data = np.sin(0.2 * np.arange(0, 100)) + np.random.normal(0, 0.1, 100)
outliers = [20, 50, 80]
data[outliers] += [3, -3, 2]

# 슬라이딩 윈도우 적용
window_size = 10
windows = sliding_window(data, window_size) # (N, L) -> (N, window_size)
windows = windows[..., np.newaxis]  # (N, L, C) -> (N, window_size, 1)
# PyTorch Conv1d는 (N, C, L) 입력을 기대하므로 차원 변경
windows = windows.transpose(0, 2, 1) # (N, C, L) -> (N, 1, window_size)
print(f"윈도우 데이터 형태 (N, C, L): {windows.shape}")
윈도우 데이터 형태 (N, C, L): (91, 1, 10)
import torch.optim as optim # PyTorch 옵티마이저

# 모델 생성
# input_shape은 (window_size, 1) 이어야 합니다. (sequence_length, num_features)
# data-generation 셀에서 windows는 (N, 1, window_size) 형태로 준비됨.
# CNNAutoencoder의 __init__은 input_shape=(window_size, 1)을 받아 input_shape[1]=1을 in_channels로 사용.
model_input_shape = (window_size, 1) # (sequence_length, num_features)
model = CNNAutoencoder(model_input_shape) # cnn-autoencoder-definition 셀에서 정의된 클래스 사용

optimizer = optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.MSELoss() # 평균 제곱 오차 손실

print("PyTorch 모델 구조:")
print(model)
PyTorch 모델 구조:
CNNAutoencoder(
  (encoder_conv1): Conv1d(1, 32, kernel_size=(3,), stride=(1,), padding=(1,))
  (encoder_relu1): ReLU()
  (encoder_pool1): MaxPool1d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  (encoder_conv2): Conv1d(32, 16, kernel_size=(3,), stride=(1,), padding=(1,))
  (encoder_relu2): ReLU()
  (encoder_pool2): MaxPool1d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  (decoder_conv_t1): ConvTranspose1d(16, 16, kernel_size=(4,), stride=(2,), padding=(1,), output_padding=(1,))
  (decoder_relu1): ReLU()
  (decoder_conv_t2): ConvTranspose1d(16, 32, kernel_size=(3,), stride=(2,), padding=(1,), output_padding=(1,))
  (decoder_relu2): ReLU()
  (decoder_conv_final): Conv1d(32, 1, kernel_size=(3,), stride=(1,), padding=(1,))
  (sigmoid): Sigmoid()
)

모델 학습

정상 데이터만 사용해 모델을 학습합니다:

# torch는 cnn-autoencoder-definition 셀에서 이미 import 됨
from torch.utils.data import TensorDataset, DataLoader

# 정상 데이터로 학습
# 'outliers'는 원본 'data' 배열의 인덱스입니다.
# 'windows' 배열에서 이상치가 포함된 윈도우를 식별하여 제외합니다.
contaminated_window_indices = set()
# 'outliers', 'window_size', 'windows' 변수는 이전 셀들에서 정의되어 있어야 합니다.
for outlier_data_idx in outliers: 
    start_contaminated_win_idx = max(0, outlier_data_idx - window_size + 1)
    end_contaminated_win_idx = outlier_data_idx 
    
    for win_idx in range(start_contaminated_win_idx, end_contaminated_win_idx + 1):
        if win_idx < len(windows): # 윈도우 인덱스가 유효한 범위 내에 있는지 확인
            contaminated_window_indices.add(win_idx)

normal_windows_mask = np.ones(len(windows), dtype=bool)
if contaminated_window_indices: # set이 비어있지 않은 경우에만 인덱싱
    normal_windows_mask[list(contaminated_window_indices)] = False

normal_windows_np = windows[normal_windows_mask]

if len(normal_windows_np) == 0:
    print("경고: 학습에 사용할 정상 윈도우가 없습니다. Outlier 정의, window_size 또는 데이터 길이를 확인하세요.")
else:
    # PyTorch 데이터셋 및 로더 준비
    normal_windows_torch = torch.tensor(normal_windows_np, dtype=torch.float32)
    train_dataset = TensorDataset(normal_windows_torch) # 오토인코더는 입력과 타겟이 동일
    train_loader = DataLoader(train_dataset, batch_size=16, shuffle=True)

    # 모델 학습
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f"Using device: {device}")
    model.to(device)
    
    epochs = 50 # 에포크 수 설정
    print_every_epochs = 10

    model.train() # 학습 모드
    for epoch in range(epochs):
        epoch_loss = 0
        for batch_data_list in train_loader:
            inputs = batch_data_list[0].to(device)
            targets = inputs # 오토인코더의 타겟은 입력과 동일

            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, targets)
            loss.backward()
            optimizer.step()
            
            epoch_loss += loss.item() * inputs.size(0) # 배치 손실 누적 (loss.item()은 평균 손실)
        
        epoch_loss /= len(train_loader.dataset) # 에포크 평균 손실
        if (epoch + 1) % print_every_epochs == 0:
            print(f"Epoch [{epoch+1}/{epochs}], Loss: {epoch_loss:.6f}")
    print("모델 학습 완료.")
Using device: cpu
Epoch [10/50], Loss: 0.485761
Epoch [20/50], Loss: 0.230847
Epoch [30/50], Loss: 0.215341
Epoch [40/50], Loss: 0.208849
Epoch [50/50], Loss: 0.207247
모델 학습 완료.

재구성 오차 계산 및 이상치 탐지

학습된 모델로 데이터를 복원하고, 재구성 오차를 계산합니다:

# torch 및 numpy는 이전 셀들에서 이미 import 됨

# 재구성 오차 계산
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
model.eval() # 평가 모드

# 전체 windows 데이터를 PyTorch 텐서로 변환하고 device로 이동
all_windows_torch = torch.tensor(windows, dtype=torch.float32).to(device)

# 메모리 부족을 방지하기 위해 배치 단위로 처리할 수 있으나, 현재 데이터는 작으므로 한번에 처리
with torch.no_grad(): # 그래디언트 계산 비활성화
    reconstructed_torch = model(all_windows_torch)

# 결과를 CPU로 옮기고 NumPy 배열로 변환
reconstructed_np = reconstructed_torch.cpu().numpy()

# MAE (Mean Absolute Error) 계산
# 원본 windows (numpy 배열)와 reconstructed_np 모두 (N, 1, window_size) 형태
# axis=(1, 2)는 채널과 시퀀스 길이에 대한 평균을 의미
mae = np.mean(np.abs(windows - reconstructed_np), axis=(1, 2))
print(f"계산된 MAE 값 (처음 5개): {mae[:5]}")

# 이상치 탐지를 위한 임계값 설정 (데이터 및 모델 성능에 따라 조정 필요)
# 예: MAE의 평균 + (표준편차 * 특정 배수) 또는 분위수 사용
threshold = np.mean(mae) + 1.5 * np.std(mae) # 표준편차 배수를 2에서 1.5로 줄여 민감도 증가
print(f"이상치 탐지 임계값 (MAE): {threshold:.4f}")

anomalies_indices_in_windows = np.where(mae > threshold)[0] # 윈도우 배열 내의 인덱스

print(f"이상치로 탐지된 윈도우의 수: {len(anomalies_indices_in_windows)}")
print(f"이상치로 탐지된 윈도우 인덱스: {anomalies_indices_in_windows}")

# 윈도우 인덱스를 원본 데이터 인덱스로 대략적으로 매핑 (윈도우의 시작점 기준)
# 실제 이상치 발생 시점과 정확히 일치하지 않을 수 있음
anomalies_approx_original_indices = anomalies_indices_in_windows 
# 좀 더 정확하게는 윈도우의 중간 지점 등을 고려할 수 있으나, 여기서는 시작점으로 단순화
# anomalies_approx_original_indices = [idx + window_size // 2 for idx in anomalies_indices_in_windows]
print(f"원본 데이터의 대략적인 이상치 인덱스 (윈도우 시작점 기준): {anomalies_approx_original_indices}")
계산된 MAE 값 (처음 5개): [0.10953805 0.09630316 0.0786427  0.0780262  0.07156017]
이상치 탐지 임계값 (MAE): 0.9568
이상치로 탐지된 윈도우의 수: 8
이상치로 탐지된 윈도우 인덱스: [17 18 19 20 47 48 49 50]
원본 데이터의 대략적인 이상치 인덱스 (윈도우 시작점 기준): [17 18 19 20 47 48 49 50]

결과 시각화

import matplotlib.pyplot as plt

plt.figure(figsize=(10, 4))
plt.plot(data, label='원본 데이터', alpha=0.7) # 'data'는 data-generation에서 정의됨
plt.scatter(outliers, data[outliers], color='red', s=100, label='실제 이상치 (Ground Truth)', marker='o', edgecolors='black') # 'outliers'는 data-generation에서 정의됨

# anomalies_approx_original_indices가 비어있을 수 있으므로 확인
if len(anomalies_approx_original_indices) > 0:
    # 탐지된 이상치 표시는 윈도우의 시작점을 기준으로 함
    plt.scatter(anomalies_approx_original_indices, data[anomalies_approx_original_indices], 
                color='orange', marker='x', s=80, label='탐지된 이상치 (모델 예측)', alpha=0.8)
else:
    print("탐지된 이상치가 없습니다.")
        
plt.legend()
plt.title('PyTorch CNN 오토인코더 기반 시계열 이상 탐지')
plt.xlabel('시간 스텝')
plt.ylabel('값')
plt.grid(True, linestyle='--', alpha=0.6)
plt.show()

# 재구성 오차(MAE) 시각화
plt.figure(figsize=(10, 3))
plt.plot(mae, label='재구성 오차 (MAE)', color='green')
plt.axhline(threshold, color='red', linestyle='--', label=f'임계값 ({threshold:.2f})')
plt.title('윈도우별 재구성 오차 (MAE) 및 임계값')
plt.xlabel('윈도우 인덱스')
plt.ylabel('MAE')
plt.legend()
plt.grid(True, linestyle='--', alpha=0.6)
plt.show()

PyTorch CNN 오토인코더 기반 이상 탐지 결과

탐지 결과 분석 및 고려사항

시각화 결과와 재구성 오차 검토 시 다음 사항을 고려해야 한다.

  1. 실제 이상치 vs. 탐지 이상치:
    • data-generation 단계에서 의도적으로 넣은 실제 이상치(outliers = [20, 50, 80])와 모델의 탐지 결과는 다를 수 있다.
    • 모든 실제 이상치가 탐지되지 않거나, 정상이 이상치로 잘못 탐지될 가능성이 항상 존재한다.
    • 현 예제는 임계값(np.mean(mae) + 1.5 * np.std(mae)) 조정을 통해 최소 하나의 이상치를 탐지하도록 유도했다. 실제 상황에서는 모델 성능, 데이터 특성, window_size, 임계값 설정에 따라 결과가 크게 달라진다.
  2. 윈도우 경계 효과 (Edge Effects):
    • 시계열 데이터의 시작과 끝 부분 윈도우는 내부 윈도우에 비해 정보가 불완전할 수 있다 (이전/이후 데이터 부재).
    • CNN 모델, 특히 패딩 사용 시, 경계 영역 윈도우는 학습된 주 정상 패턴과 달라 재구성 오차가 상대적으로 커질 수 있다.
    • 결과적으로, 시계열 양 끝부분에서 이상치가 아닌데도 이상치로 탐지되는 경향이 나타날 수 있다. MAE 그래프에서 초반 또는 후반부에 높은 오차가 관찰된다면 이 효과를 의심해볼 수 있다.
  3. window_size의 중요성:
    • window_size는 모델이 학습할 패턴의 길이를 결정한다.
    • 너무 작으면 장기 패턴 파악이 어렵고, 너무 크면 짧은 순간의 이상치를 놓치거나 정상 변동에도 민감하게 반응할 수 있다.
    • 현재 window_size=10으로 설정했다. 데이터 특성에 맞춰 이 값을 조정하며 실험하는 과정이 중요하다.
  4. 모델 및 임계값의 한계:
    • 여기서 사용한 CNN 오토인코더는 비교적 단순한 모델이다.
    • 더 복잡한 패턴이나 다양한 유형의 이상치를 탐지하려면 모델 구조 개선(예: LSTM, Transformer 기반 오토인코더)이나 다른 접근법을 고려해야 한다.
    • 고정 임계값 대신 동적 임계값을 사용하거나, 통계적 검정 기법을 결합하는 것도 탐지 성능을 높이는 데 도움이 될 수 있다.

이런 점들을 고려해 모델 결과를 해석해야 하며, 실제 문제 적용 시에는 충분한 검증과 실험이 필수다.

결론 및 다음 단계

이번 주에는 PyTorch로 간단한 1D CNN 오토인코더를 만들고, 시계열 이상 탐지를 수행했다. 이 모델은 시계열 이상 탐지의 괜찮은 시작점이 될 수 있다. 재구성 오차를 기반으로 이상치를 찾는 과정과, 임계값 설정에 따라 탐지 결과가 어떻게 달라지는지 확인했다.

다음 포스트에서는 실제 산업 데이터를 사용해 모델을 학습시키고, 성능을 개선할 다양한 방법(예: 더 복잡한 모델 구조, 다른 유형의 오토인코더, 동적 임계값 설정 등)을 살펴볼 예정이다.