DeepSAD.py 파일 상세 분석

import json
import torch

from base.base_dataset import BaseADDataset
from networks.main import build_network, build_autoencoder
from optim.DeepSAD_trainer import DeepSADTrainer
from optim.ae_trainer import AETrainer


class DeepSAD(object):
    """A class for the Deep SAD method.

    Attributes:
        eta: Deep SAD hyperparameter eta (must be 0 < eta).
        c: Hypersphere center c.
        net_name: A string indicating the name of the neural network to use.
        net: The neural network phi.
        trainer: DeepSADTrainer to train a Deep SAD model.
        optimizer_name: A string indicating the optimizer to use for training the Deep SAD network.
        ae_net: The autoencoder network corresponding to phi for network weights pretraining.
        ae_trainer: AETrainer to train an autoencoder in pretraining.
        ae_optimizer_name: A string indicating the optimizer to use for pretraining the autoencoder.
        results: A dictionary to save the results.
        ae_results: A dictionary to save the autoencoder results.
    """

    def __init__(self, eta: float = 1.0):
        """Inits DeepSAD with hyperparameter eta."""

        self.eta = eta
        self.c = None  # hypersphere center c

        self.net_name = None
        self.net = None  # neural network phi

        self.trainer = None
        self.optimizer_name = None

        self.ae_net = None  # autoencoder network for pretraining
        self.ae_trainer = None
        self.ae_optimizer_name = None

        self.results = {
            'train_time': None,
            'test_auc': None,
            'test_time': None,
            'test_scores': None,
        }

        self.ae_results = {
            'train_time': None,
            'test_auc': None,
            'test_time': None
        }

    def set_network(self, net_name):
        """Builds the neural network phi."""
        self.net_name = net_name
        self.net = build_network(net_name)

    def train(self, dataset: BaseADDataset, optimizer_name: str = 'adam', lr: float = 0.001, n_epochs: int = 50,
              lr_milestones: tuple = (), batch_size: int = 128, weight_decay: float = 1e-6, device: str = 'cuda',
              n_jobs_dataloader: int = 0):
        """Trains the Deep SAD model on the training data."""

        self.optimizer_name = optimizer_name
        self.trainer = DeepSADTrainer(self.c, self.eta, optimizer_name=optimizer_name, lr=lr, n_epochs=n_epochs,
                                      lr_milestones=lr_milestones, batch_size=batch_size, weight_decay=weight_decay,
                                      device=device, n_jobs_dataloader=n_jobs_dataloader)
        # Get the model
        self.net = self.trainer.train(dataset, self.net)
        self.results['train_time'] = self.trainer.train_time
        self.c = self.trainer.c.cpu().data.numpy().tolist()  # get as list

    def test(self, dataset: BaseADDataset, device: str = 'cuda', n_jobs_dataloader: int = 0):
        """Tests the Deep SAD model on the test data."""

        if self.trainer is None:
            self.trainer = DeepSADTrainer(self.c, self.eta, device=device, n_jobs_dataloader=n_jobs_dataloader)

        self.trainer.test(dataset, self.net)

        # Get results
        self.results['test_auc'] = self.trainer.test_auc
        self.results['test_time'] = self.trainer.test_time
        self.results['test_scores'] = self.trainer.test_scores

    def pretrain(self, dataset: BaseADDataset, optimizer_name: str = 'adam', lr: float = 0.001, n_epochs: int = 100,
                 lr_milestones: tuple = (), batch_size: int = 128, weight_decay: float = 1e-6, device: str = 'cuda',
                 n_jobs_dataloader: int = 0):
        """Pretrains the weights for the Deep SAD network phi via autoencoder."""

        # Set autoencoder network
        self.ae_net = build_autoencoder(self.net_name)

        # Train
        self.ae_optimizer_name = optimizer_name
        self.ae_trainer = AETrainer(optimizer_name, lr=lr, n_epochs=n_epochs, lr_milestones=lr_milestones,
                                    batch_size=batch_size, weight_decay=weight_decay, device=device,
                                    n_jobs_dataloader=n_jobs_dataloader)
        self.ae_net = self.ae_trainer.train(dataset, self.ae_net)

        # Get train results
        self.ae_results['train_time'] = self.ae_trainer.train_time

        # Test
        self.ae_trainer.test(dataset, self.ae_net)

        # Get test results
        self.ae_results['test_auc'] = self.ae_trainer.test_auc
        self.ae_results['test_time'] = self.ae_trainer.test_time

        # Initialize Deep SAD network weights from pre-trained encoder
        self.init_network_weights_from_pretraining()

    def init_network_weights_from_pretraining(self):
        """Initialize the Deep SAD network weights from the encoder weights of the pretraining autoencoder."""

        net_dict = self.net.state_dict()
        ae_net_dict = self.ae_net.state_dict()

        # Filter out decoder network keys
        ae_net_dict = {k: v for k, v in ae_net_dict.items() if k in net_dict}
        # Overwrite values in the existing state_dict
        net_dict.update(ae_net_dict)
        # Load the new state_dict
        self.net.load_state_dict(net_dict)

    def save_model(self, export_model, save_ae=True):
        """Save Deep SAD model to export_model."""

        net_dict = self.net.state_dict()
        ae_net_dict = self.ae_net.state_dict() if save_ae else None

        torch.save({'c': self.c,
                    'net_dict': net_dict,
                    'ae_net_dict': ae_net_dict}, export_model)

    def load_model(self, model_path, load_ae=False, map_location='cpu'):
        """Load Deep SAD model from model_path."""

        model_dict = torch.load(model_path, map_location=map_location)

        self.c = model_dict['c']
        self.net.load_state_dict(model_dict['net_dict'])

        # load autoencoder parameters if specified
        if load_ae:
            if self.ae_net is None:
                self.ae_net = build_autoencoder(self.net_name)
            self.ae_net.load_state_dict(model_dict['ae_net_dict'])

    def save_results(self, export_json):
        """Save results dict to a JSON-file."""
        with open(export_json, 'w') as fp:
            json.dump(self.results, fp)

    def save_ae_results(self, export_json):
        """Save autoencoder results dict to a JSON-file."""
        with open(export_json, 'w') as fp:
            json.dump(self.ae_results, fp)

1. 전체 코드 구조

DeepSAD.py는 Deep Semi-supervised Anomaly Detection 알고리즘의 핵심 구현 파일입니다. 전체적인 코드 구조는 다음과 같습니다:

# 1. 필요한 모듈 임포트
# 2. DeepSAD 클래스 정의
# 3. DeepSADTrainer 클래스 정의 (DeepSAD가 상속받는다)
# 4. AETrainer 클래스 정의 (자동인코더 학습용)

class DeepSAD(object):
    """Deep SAD 메서드의 클래스"""
    
    def __init__(self, eta: float = 1.0):
        """DeepSAD 초기화"""
        # 하이퍼파라미터 설정
    
    def set_network(self, net_name):
        """신경망 구조 설정"""
        
    def train(self, dataset: BaseADDataset, ...):
        """DeepSAD 모델 학습"""
        
    def test(self, dataset: BaseADDataset, ...):
        """DeepSAD 모델 테스트"""
        
    def pretrain(self, dataset: BaseADDataset, ...):
        """자동인코더를 통한 사전학습"""
        
    def init_network_weights_from_pretraining(self):
        """사전학습 결과로 네트워크 초기화"""
        
    def save_model(self, export_model, save_ae=True):
        """모델 저장"""
        
    def load_model(self, model_path, load_ae=False, map_location='cpu'):
        """모델 로드"""
        
    def save_results(self, export_json):
        """결과 저장"""
        
    def save_ae_results(self, export_json):
        """자동인코더 결과 저장"""

2. DeepSAD 클래스 실행 순서

2.1 전체 실행 흐름도:

1. DeepSAD 인스턴스 생성
2. 네트워크 설정 (set_network)
3. 사전학습 (pretrain) - 선택사항
4. DeepSAD 학습 (train)
5. DeepSAD 테스트 (test)
6. 결과 저장

2.2 자세한 실행 단계:

단계 1: DeepSAD 인스턴스 생성

# DeepSAD 클래스의 __init__ 메서드 호출
deep_sad = DeepSAD(eta=1.0)  # eta = 1.0

# 초기화된 멤버 변수:
- self.eta = 1.0
- self.c = None (하이퍼구 중심)
- self.net_name = None
- self.net = None (신경망)
- self.trainer = None
- self.optimizer_name = None
- self.ae_net = None (자동인코더)
- self.ae_trainer = None
- self.ae_optimizer_name = None

단계 2: 네트워크 설정

# set_network 메서드 호출
deep_sad.set_network("mnist_LeNet")

# 내부 처리:
# - self.net_name = "mnist_LeNet"
# - self.net = build_network("mnist_LeNet")  # 신경망 생성

단계 3: 사전학습 (선택적 단계)

# pretrain 메서드 호출 (사전학습이 필요한 경우)
deep_sad.pretrain(dataset, optimizer_name="adam", lr=0.001, n_epochs=100)

# 내부 처리:
# 1. 자동인코더 네트워크 생성
# 2. AETrainer 생성 및 학습
# 3. 사전학습 결과 저장
# 4. DeepSAD 네트워크 가중치 초기화

단계 4: DeepSAD 학습

# train 메서드 호출
deep_sad.train(dataset, optimizer_name="adam", lr=0.001, n_epochs=50)

# 내부 처리:
# 1. DeepSADTrainer 생성
# 2. 학습 시작 (하이퍼구 중심 c 계산)
# 3. 네트워크 파라미터 업데이트
# 4. 최종 하이퍼구 중심 c 계산 및 저장

단계 5: DeepSAD 테스트

# test 메서드 호출
deep_sad.test(dataset)

# 내부 처리:
# 1. DeepSADTrainer 생성 (기존 학습된 c 사용)
# 2. 테스트 데이터에 대한 점수 계산
# 3. AUC 계산 및 저장

3. DeepSAD 알고리즘 원리 상세 분석

3.1 핵심 원리:

# DeepSAD의 기본 아이디어:
# 1. 정상 데이터를 하이퍼구(Hypersphere) 안에 포함
# 2. 이상치는 하이퍼구 밖으로 벗어남
# 3. 각 데이터 포인트의 이상치 점수 = ||φ(x) - c||²

3.2 수학적 표현:

목적 함수:

L = Σ(||φ(x_i) - c||²) + η * max(0, r² - ||φ(x_i) - c||²)

점수 계산:

Score(x) = ||φ(x) - c||²

하이퍼구 중심 업데이트:

c = (1/N) * Σφ(x_i)  (정상 데이터에 대해)

4. DeepSAD 학습 과정 상세 분석

4.1 학습 단계별 구현:

학습 메서드 내부 로직:

def train(self, dataset: BaseADDataset, optimizer_name: str = 'adam', 
          lr: float = 0.001, n_epochs: int = 50, ...):
    
    # 1. 트레이너 초기화
    self.trainer = DeepSADTrainer(self.c, self.eta, ...)
    
    # 2. 트레이너의 train 메서드 호출
    self.net = self.trainer.train(dataset, self.net)
    
    # 3. 학습 시간 저장
    self.results['train_time'] = self.trainer.train_time
    
    # 4. 하이퍼구 중심 c 계산 및 저장
    self.c = self.trainer.c.cpu().data.numpy().tolist()

4.2 DeepSADTrainer의 학습 로직:

# DeepSADTrainer.train() 내부 구현 요약:
class DeepSADTrainer(BaseTrainer):
    def train(self, dataset: BaseADDataset, net: BaseNet) -> BaseNet:
        # 1. 데이터 로더 생성
        train_loader = dataset.loaders(batch_size=self.batch_size, shuffle=True)
        
        # 2. 하이퍼구 중심 c 초기화 (필요 시)
        if self.c is None:
            # 정상 데이터만으로 c 계산
            self.initialize_center(dataset, net)
        
        # 3. 반복 학습 (N epochs)
        for epoch in range(self.n_epochs):
            # 3.1 배치별 학습
            for batch_idx, (data, labels) in enumerate(train_loader):
                # 3.2 정상 데이터만 필터링
                normal_data = data[labels == 0]
                
                # 3.3 신경망 통과 (특징 벡터 계산)
                features = net(normal_data)  # φ(x)
                
                # 3.4 손실 계산 (DeepSAD 목적 함수)
                loss = self.compute_loss(features, self.c)
                
                # 3.5 역전파 및 파라미터 업데이트
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()
                
                # 3.6 하이퍼구 중심 업데이트 (정상 데이터 특징 평균)
                self.update_center(features)
        
        return net

5. DeepSAD 테스트 과정 상세 분석

5.1 테스트 메서드 내부 로직:

def test(self, dataset: BaseADDataset, device: str = 'cuda', n_jobs_dataloader: int = 0):
    # 1. 트레이너 초기화 (기존 학습된 c 사용)
    if self.trainer is None:
        self.trainer = DeepSADTrainer(self.c, self.eta, device=device, ...)
    
    # 2. 테스트 실행
    self.trainer.test(dataset, self.net)
    
    # 3. 결과 저장
    self.results['test_auc'] = self.trainer.test_auc
    self.results['test_time'] = self.trainer.test_time
    self.results['test_scores'] = self.trainer.test_scores

5.2 DeepSADTrainer의 테스트 로직:

class DeepSADTrainer(BaseTrainer):
    def test(self, dataset: BaseADDataset, net: BaseNet):
        # 1. 테스트 데이터 로더 생성
        test_loader = dataset.loaders(batch_size=self.batch_size, shuffle=False)
        
        # 2. 모든 테스트 샘플에 대해 점수 계산
        all_scores = []
        all_labels = []
        
        for batch_idx, (data, labels) in enumerate(test_loader):
            # 1. 신경망 통과로 특징 벡터 계산
            features = net(data)  # φ(x)
            
            # 2. 이상치 점수 계산
            scores = torch.norm(features - self.c, p=2, dim=1)  # ||φ(x) - c||²
            
            # 3. 결과 저장
            all_scores.extend(scores.cpu().numpy())
            all_labels.extend(labels.cpu().numpy())
        
        # 4. AUC 계산
        self.test_auc = roc_auc_score(all_labels, all_scores)
        
        # 5. 테스트 점수 저장
        self.test_scores = list(zip(range(len(all_scores)), all_labels, all_scores))

6. DeepSAD의 성능 분석

6.1 성능 요소:

1) 정확도 (AUC):

# AUC = Area Under ROC Curve
# 정상/이상치 분류 성능 측정
# 값이 클수록 성능 좋음 (0.5 ~ 1.0)

# 예시: AUC = 0.92 (좋은 성능)

2) 하이퍼구 중심의 정확성:

# c 계산 정확도가 성능에 직접 영향
# 정상 데이터 특징 평균이 정확할수록 성능 향상

3) 차원 선택의 영향:

# rep_dim = 32 (최적) vs rep_dim = 16/64 (성능 저하)

6.2 성능 향상 방법:

1) 최적 차원 선택:

# 다양한 rep_dim으로 실험 후 최적 선택
rep_dims = [16, 32, 64]
best_auc = 0
optimal_dim = 32

for dim in rep_dims:
    # 각 차원으로 학습 및 테스트
    model = DeepSAD(eta=1.0)
    model.set_network(f"mnist_LeNet_{dim}")
    model.train(dataset)
    model.test(dataset)
    
    auc = model.results['test_auc']
    if auc > best_auc:
        best_auc = auc
        optimal_dim = dim

2) 사전학습 활용:

# 자동인코더를 통한 가중치 초기화로 성능 향상
# 사전학습이 있는 경우 AUC 향상 가능

7. 실제 실행 예시:

7.1 전체 실행 흐름:

# 1. DeepSAD 모델 생성
deep_sad = DeepSAD(eta=1.0)

# 2. 네트워크 설정
deep_sad.set_network("mnist_LeNet")  # rep_dim = 32

# 3. 사전학습 (선택사항)
deep_sad.pretrain(dataset, optimizer_name="adam", lr=0.001, n_epochs=100)

# 4. DeepSAD 학습
deep_sad.train(dataset, optimizer_name="adam", lr=0.001, n_epochs=50)

# 5. DeepSAD 테스트
deep_sad.test(dataset)

# 6. 결과 확인
print("AUC:", deep_sad.results['test_auc'])
print("하이퍼구 중심 c:", deep_sad.c)

7.2 핵심 실행 흐름 상세:

Step-by-step 실행:

1. DeepSAD 인스턴스 생성: DeepSAD(eta=1.0)
2. 네트워크 설정: set_network("mnist_LeNet") → rep_dim = 32
3. 하이퍼구 중심 초기화: c = torch.zeros(32)
4. 학습 시작: train() 메서드 실행
5. 특징 추출: net(input) → 32차원 특징 벡터
6. 하이퍼구 중심 업데이트: c = mean(features)
7. 점수 계산: ||φ(x) - c||² → 이상치 점수
8. 성능 평가: AUC 계산

8. DeepSAD의 주요 특징 요약

8.1 장점:

• 비지도 학습: 정상 데이터만으로도 이상치 탐지 가능
• 심층 특징 추출: 신경망을 통한 복잡한 패턴 인식
• 정확성: 정상 데이터 분포 모델링으로 높은 정확도

8.2 단점:

• 정상 데이터 의존: 정상 데이터가 부족하면 성능 저하
• 계산 복잡성: 하이퍼구 중심 계산에 시간 소요
• 차원 선택 중요: 최적 차원이 필요

8.3 성능 특성:

• AUC 범위: 일반적으로 0.8~0.95 (MNIST 기준)
• 최적 차원: 32차원 (MNIST에서 일반적으로 최적)
• 학습 시간: 10~30분 (MNIST, 50 epochs)

9. 실행 순서 요약

최종 실행 흐름:

1. DeepSAD(eta=1.0) → __init__
2. set_network("mnist_LeNet") → 네트워크 생성 (rep_dim=32)
3. pretrain(dataset) → 자동인코더 학습 (선택)
4. train(dataset) → DeepSAD 학습 (c 계산, 네트워크 학습)
5. test(dataset) → 테스트 및 AUC 계산
6. 결과 저장 및 출력

DeepSAD는 정상 데이터 분포를 하이퍼구로 모델링하여 이상치 탐지 성능을 높이는 알고리즘입니다. 핵심적인 특징은 하이퍼구 중심의 정확한 계산과 최적의 차원 선택에 있습니다.