[TIL] 데이터분석 데브코스 49일차 (2) - 이상탐지/Isolation Forest(고립 숲)/kaggle로 이상탐지 실습

이상탐지 (Anomaly Detection)

: 데이터에서 비정상적인 패턴, 이상치, 또는 예외적인 사례를 탐지하는 과정

• 데이터에서 일반적으로 볼 수 있는 특성에서 많이 벗어난 데이터를 식별하는 과정에서 사용
• 보안, 금융, 의료 등의 분야에서 중요한 역할

 

이상 탐지 방법

전통적인 통계 기반 방법

IQR (Interquartile Range)

• 1사분위수(Q1)와 3사분위수(Q3)간의 차이로
• Q1-1.5*IQR 미만 혹은 Q3+1.5*IQR 초과 데이터를 이상치로 간주

Z-Score

• 데이터 포인트가 평균으로부터 표준 편차의 몇 배만큼 떨어져 있는지를 나타내는 수치
• 이 수치가 3이상인 데이터를 이상치로 간주

 

 

 

머신러닝을 활용한 방법

: 고차원&대량의 데이터의 패턴을 익히고 그 패턴에서 벗어난 데이터를 이상치로 취급하는 과정

• 장점
   
  • 복잡한 데이터의 패턴에 적응적
  • (대량의 데이터의 경우) 지속적 학습 및 개선 가능
• 단점
  • 처리할 모델의 복잡성 증가
  • 사람이 해석하기 어려운 패턴 존재할 수 있음
  • 양질의 데이터가 많이 필요함

 

 

 

이상 탐지 모델 : Isolation Forest

: 하나의 데이터를 고립시키도록 특정 특성과 그것의 분할 값을 기준으로 나누어 이상치를 탐지하는 모델

 

• 정상 데이터
  • 밀도가 높은 구역에 몰려있어서 많은 분할 과정이 지나야 고립됨 
• 이상치
  • 밀도가 낮은 지역에 홀로 있기 때문에 낮은 수준의 분할 과정으로도 쉽게 고립 가능

 

 

동작 과정

1. 데이터 준비

전체 데이터에서 무작위로 서브 데이터셋을 선택 (어떤 데이터는 여러 서브 데이터셋에 속할 수 있음)

 

 

2. 트리 생성

하나의 서브 데이터셋마다 개별 Isolation Tree를 준비

 

 

3. 고립 시작

각 트리에 소속된 데이터셋을 하나씩 고립시킴

• 임의의 특성과 임의의 특성 분할값을 선택
• 데이터셋이 리프 노드에 도달할 때까지 반복
• 각 데이터 포인트는 리프 노드까지 도달한 깊이를 저장하고 있음

 

 

4. 고립 완료

하나의 트리에 소속된 모든 데이터셋을 전부 고립시킴

 

 

5. 트리 계산 완료

모든 트리에서 고립 과정이 완료되면, 여러 트리에 소속된 어떤 데이터 포인트는 서로 다른 트리에서 계산된 깊이가 저장됨

 

6. 이상치 점수 계산

평균 깊이를 바탕으로 각 데이터 포인트의 이상치 점수를 계산

 

7. 이상치 추출

이상치 점수와 특정 임계값을 비교해 이상치 추출

 

 

 

 

 

 

 

Isolation Tree를 활용한 이상치 탐지 실습

`IsolationForest()` : 고립숲 모델 구축 함수

• `n_estimatores` : 고립숲을 구성하는 고립 트리의 개수(=서브 데이터셋 개수)
  • 더 많은 트리를 사용할수록 이상치 탐지의 안정석 & 정확도 향상
  • but, 트리 개수가 많을수록 계산 시간 & 메모리 사용량 증가
• `max_samples` : 서브 데이터셋에 포함될 최대 데이터 포인트의 개수
  • `'auto'` : 자동으로 설정해줌
  • 샘플수가 너무 적으면 : 모델 성능 저하
  • 샘플수가 너무 많으면 : 계산 비용 증가
• `contamination` : 전체 데이터셋에서 이상치 데이터가 차지할 비율
  • 이 비율로 정상 데이터와 이상치 사이의 이상치 점수 임계값 설정 가능
  • `'auto'` : 데이터 분포 특성에 따라 합리적 비율을 자동 추출(이상치의 정확한 비율을 모르거나 데이터가 복잡할 경우 사용)

 

 

 

1단계. 데이터 생성

import numpy as np
import matplotlib.pyplot as plt

정상 데이터 : 100개

비정상 데이터 : 20개

seed = 1234
np.random.seed(seed)

# 정상 데이터
X_normal = 0.3 * np.random.randn(100, 2)

# 이상치 데이터
X_outliers = np.random.uniform(low=-4, high=4, size=(20, 2))

# 전체 데이터
X = np.vstack([X_normal, X_outliers])


# 시각화
plt.figure(figsize=(15, 9))

plt.scatter(X_normal[:, 0], X_normal[:, 1], c='yellow', edgecolor='k', s=20, label="Normal data")
plt.scatter(X_outliers[:, 0], X_outliers[:, 1], c='red', edgecolor='k', s=20, label="Outliers")

plt.title("Normal Data and Outliers")
plt.xlabel("Feature 1")
plt.ylabel("Feature 2")

plt.legend()
plt.show()

 

 

 

 

2단계. Isolation Forest 모델 생성 및 학습

# Isolation Forest 모델 생성을 위한 매개 변수
n_estimators = 100  # 사용할 트리의 수
max_samples = 'auto'  # 각 트리당 최대 샘플 수
contamination = 0.15  # 전체 데이터셋에서 이상치가 차지하는 비율의 추정치
                      # 이 값을 이용해 추후 이상치를 찾아냄
                      # 이상치 점수가 이 임계값보다 높은 데이터 포인트는 이상치로 간주

# Isolation Forest 생성 및 학습
from sklearn.ensemble import IsolationForest
IForest = IsolationForest(n_estimators=n_estimators,
                          max_samples=max_samples,
                          contamination=contamination,
                          random_state=seed)
IForest.fit(X)

 

 

 

 

 

 

3단계. Isolation Forest를 활용한 이상치 추정

# 모든 데이터에 대한 이상치 점수 계산
# 이 값은 트리 내에서의 평균 경로 depth를 기반으로 계산됨
# 정규화 된 값으로,
# 점수가 낮으면 평균 길이가 짧다고 보면 되며 이상치일 가능성이 ↑

scores_pred = IForest.decision_function(X)
print(scores_pred)

 

 

시각화 : 점수가 낮을수록 데이터 포인트를 둘러싼 원이 커진다.

# 이상치 점수를 바탕으로 시각화
plt.figure(figsize=(15, 9))

plt.scatter(X_normal[:, 0], X_normal[:, 1], c='yellow', edgecolor='k', s=20, label="Normal data")
plt.scatter(X_outliers[:, 0], X_outliers[:, 1], c='red', edgecolor='k', s=20, label="Outliers")

# 이상치 점수에 따라 점의 크기를 조정
# 파란색 원이 클 수록 이상치일 확률이 큰 것을 의미
sizes = (scores_pred.max() - scores_pred) * 500
plt.scatter(X[:, 0], X[:, 1], s=sizes, edgecolor='b', facecolors='none', label="Anomaly scores")

plt.title("Isolation Forest Anomaly Detection")
plt.xlabel("Feature 1")
plt.ylabel("Feature 2")

plt.legend()
plt.show()

 

 

임계값(`contamination`) 0.15(=15%)를 기준으로 예측

• 정상 데이터 = 1 
• 이상치 = -1

# 이상치 여부 결정
IForest.predict(X)

 

 

이걸 마스킹(masking)하면

• 정상 데이터 = `True`
• 이상치 = `False`

is_outlier = IForest.predict(X) == -1
is_outlier

 

 

시각화해보면

• 정상 데이터 : 하얀색 점
• 이상치 : 검정색 점

# 정상 데이터와 이상치 데이터 분리
X_normal_detected = X[~is_outlier]
X_outliers_detected = X[is_outlier]

# 시각화
plt.figure(figsize=(15, 9))
plt.scatter(X_normal_detected[:, 0], X_normal_detected[:, 1], c='white', edgecolor='k', s=20, label="Detected Normal Data")
plt.scatter(X_outliers_detected[:, 0], X_outliers_detected[:, 1], c='red', edgecolor='k', s=20, label="Detected Outliers")

plt.title("Detected Anomaly Data using Isolation Forest")
plt.xlabel("Feature 1")
plt.ylabel("Feature 2")
plt.legend()
plt.show()

 

 

 

4단계. 분류 문제로서 학습한 모델 평가

이상치 데이터의 정답을 알고 있다면, 이상치를 양성(positive)로 놓고 분류 형태의 문제로 성능을 평가할 수 있다.

# 실제 레이블을 생성 (0: 정상, 1: 이상치)
y_true = np.zeros(X.shape[0])
y_true[-X_outliers.shape[0]:] = 1  # 마지막 20개 데이터(이상치)에 대해 1 할당!

y_true

 

 

성능 평가를 위해 레이블링

• 정상 데이터 : 1 → 0
• 이상치 : -1 → 1

# 모델이 예측한 값을 바탕으로 이상치 레이블링
y_pred = IForest.predict(X) # 여기서는 정상이 1, 이상치가 -1
y_pred = np.where(y_pred == 1, 0, 1)  # 정상: 1 -> 0, 이상치: -1 -> 1

y_pred

 

 

# 성능 평가
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

accuracy = accuracy_score(y_true, y_pred)
precision = precision_score(y_true, y_pred)
recall = recall_score(y_true, y_pred)
f1 = f1_score(y_true, y_pred)

print(f'정확도 Accuracy : {accuracy*100:.2f} %')
print(f'정밀도 Precision : {precision*100:.2f} %')
print(f'재현율 Recall : {recall*100:.2f} %')
print(f'F1 : {f1*100:.2f} %')

► 결과 해석

• 정밀도 100% : 학습시킨 모델이 이상치라고 예측한 것들 중 실제로도 이상치였던 것

 

 

 

 

 

 

 

 

---

이상탐지 실습

 

실습 데이터

신용카드 거래에서의 사기 탐지를 위해 설계된 데이터셋

 

🔗 실습 링크 : https://www.kaggle.com/datasets/mlg-ulb/creditcardfraud

Credit Card Fraud Detection

 

 

변수명	의미
Time	첫 거래 후 다음의 각 거래 사이 경과 시간(초)
Amount	거래액
V1 ~ V28	개인 정보 데이터를 PCA로 차원 축소하여 얻은 수치형 입력변수
Class	정상 거래 여부 (종속변수) - 0 : 정상 거래 - 1 : 비정상 거래

 

 

 

 

문제 정의

: 주어진 [독립변수]거래 관련 데이터들(시간, 거래액)을 바탕으로 이상 거래 데이터를 탐지하는 것

 

 

 

 

 

1단계. 데이터 로드

import numpy as np

seed = 1234
np.random.seed(seed)

import pandas as pd

# 데이터 경로 지정 및 읽어오기
data_path = '/content/sampled_creditcard.csv'
credit = pd.read_csv(data_path)

 

 

 

 

2단계. EDA

1) 기본 정보 및 기초 통계 분석

# 기본 정보
print('#'*20, '기본 정보', '#'*20)
credit.info() # info() 안에서 자동으로 print를 진행

# 기초 통계량
summary_statistics = credit.describe(include='all')
print('#'*20, '기초 통계량', '#'*20)
print(summary_statistics)

 

 

 

2) 이상치 여부 시각화 - 범주형 데이터

# 이상치 여부 시각화 (Class : 종속변수)
y_columns = ['Class']
y = credit[y_columns]

import matplotlib.pyplot as plt
plt.figure(figsize=(4, 4))

y.value_counts(normalize=True).plot(kind='bar', color=['b', 'r'])

plt.xticks(range(2), ['Nomal', 'Fraud'])
plt.title('Nomal VS Fraud')
plt.tight_layout()
plt.show()

 

정상 데이터 & 이상치 개수 확인해보기

# 개수 확인
num_Normal = (y==0).sum().values[0]
num_Fraud = (y==1).sum().values[0]
print(f'Number of Normal data : {num_Normal}')
print(f'Number of Fraud data : {num_Fraud}')

 

 

 

3) 시각화 - 수치형 데이터

# Time과 Amount 데이터

time_amount_columns = ['Time', 'Amount']
time_amount = credit[time_amount_columns]

# 분포 시각화
plt.figure(figsize=(6, 9))

np.random.seed(seed)
for idx, numeric in enumerate(time_amount_columns) :
    col = (np.random.random(), np.random.random(), np.random.random())

    plt.subplot(2, 1, idx+1)
    plt.hist(time_amount[numeric], bins=50, color=col, edgecolor='black')
    plt.title(numeric)
    plt.tight_layout()

plt.tight_layout()
plt.show()

► 알 수 있는 정보

• Time(시간) 데이터
  • 기준 시간 정보가 있다면 주기성을 갖고 있을 수 있음
  • but, 주어진 데이터는 추가적인 정보가 부족 + 큰 outlier가 없어 보이므로 Min-Max Scaling 적용
    (※ 시계열 데이터의 전처리 조건 : 시작 지점이 정해져있어야 한다.)

• Amount(거래액) 데이터
  • 치우침이 있지만 신용 카드 거래 금액으로 의미가 없는 데이터는 아님
  • 치우침을 처리하기 위해 Log Scaling 적용

 

 

4) Amount(거래액) 데이터 Log Scaling 적용

• 데이터 분포가 정규 모형으로 변형
• 꼬리 부분이 줄어들고 대체적으로 고르게 분포하게 된다.
• Log(0)의 값을 피하기 위해 +1값 추가 : `np.log(credit['Amount'] + 1)`
  • Amount의 경우 1정도의 크기가 큰 문제가 되지는 않지만, 이 값이 영향을 미치는 데이터인지 확인해야한다.

# Amount에 Log scaling 적용
credit['Log_Amount'] = np.log(credit['Amount'] + 1)
credit

 

시각화 해보면 고르게 분포된 것을 확인할 수 있다.

# Time과 Amount 데이터
log_amount_columns = ['Log_Amount']
log_amount = credit[log_amount_columns]

# 분포 시각화
plt.figure(figsize=(4, 4))

np.random.seed(seed)
for idx, numeric in enumerate(log_amount_columns) :
    col = (np.random.random(), np.random.random(), np.random.random())

    plt.hist(log_amount[numeric], bins=50, color=col, edgecolor='black')
    plt.title(numeric)
    plt.tight_layout()

plt.tight_layout()
plt.show()

 

 

 

5) V1~V28 분포 파악

V1 ~ V28은 PCA로 차원축소를 진행했기 때문에 각각이 어떤 의미인지는 알 수 없음

따라서 분포를 통해 해당 변수가 쓸모있는지 없는지 판단해야함

# V1 ~ V28 데이터
Vs_columns = ['V1',  'V2',  'V3',  'V4',  'V5',  'V6',  'V7',  'V8',  'V9',  'V10',
              'V11', 'V12', 'V13', 'V14', 'V15', 'V16', 'V17', 'V18', 'V19', 'V20',
              'V21', 'V22', 'V23', 'V24', 'V25', 'V26', 'V27', 'V28']
Vs_data = credit[Vs_columns]

# 분포 시각화
plt.figure(figsize=(30, 24))

np.random.seed(seed)
for idx, numeric in enumerate(Vs_columns) :
    col = (np.random.random(), np.random.random(), np.random.random())

    plt.subplot(5, 6, idx+1)
    plt.hist(Vs_data[numeric], bins=50, color=col, edgecolor='black')
    plt.title(numeric)
    plt.tight_layout()

plt.tight_layout()
plt.show()

 

► 알 수 있는 정보

몇 가지 치우침이 있는 변수들있지만 이것들만 따로 Log 처리 해준다면 나름 쓸모있는 변수로 보인다.

 

 

 

6) 상관관계 분석 - 수치형 데이터(V1~V28)

# 상관관계 메트릭스
correlation_matrix = Vs_data.corr()

# 상관관계 메트릭스 시각화
plt.figure(figsize=(24, 24))

plt.matshow(correlation_matrix, fignum=1)
plt.colorbar()
plt.xticks(range(len(correlation_matrix.columns)), correlation_matrix.columns, rotation=90)
plt.yticks(range(len(correlation_matrix.columns)), correlation_matrix.columns)

for (i, j), val in np.ndenumerate(correlation_matrix):
    plt.text(j, i, '{:0.2f}'.format(val), ha='center', va='center', color='black')

plt.show()

 

 

 

 

 

 

3단계. 데이터 전처리

1) 결측치 제거

# 결측치 값 존재 여부 확인
exist_na = credit.isna().values.any()
exist_null = credit.isnull().values.any()
print(exist_na, exist_null)

 

 

2) Time 데이터 처리 - Min-Max Scaling

# Time 변수를 Min Max Scaling
from sklearn.preprocessing import MinMaxScaler
scaler = MinMaxScaler()

credit_time = scaler.fit_transform(credit[['Time']])
credit_time = pd.DataFrame(credit_time)
credit_time.columns = ['Time']
credit_time

 

 

3) Amount 데이터 Log scale로 변경

# Amount 변수를 Log scale로

credit_log_amount= credit[log_amount_columns]
credit_log_amount

 

 

4) 데이터 합치기

credit_combined = pd.concat([credit_time,
                             credit_log_amount,
                             Vs_data],
                            axis=1)

 

 

 

 

 

 

4단계.  Isolation Tree 모델 학습

n_estimators = 100
max_samples = 'auto'
# contamination = 'auto'
contamination = num_Fraud/(num_Normal+num_Fraud) # 'auto'로 하지 않고 실제 이상치 비율로 정해줌

# Isolation Forest 생성 및 학습
from sklearn.ensemble import IsolationForest
IForest = IsolationForest(n_estimators=n_estimators,
                          max_samples=max_samples,
                          contamination=contamination,
                          random_state=seed)
IForest.fit(credit_combined)

 

 

 

 

 

 

5단계. 학습한 모델 평가

y_true = credit['Class']

y_pred = IForest.predict(credit_combined)
y_pred = np.where(y_pred == 1, 0, 1)

 

정답이 있기 때문에 정확도/정밀도/재현율/F1점수로 파악

# 성능 평가
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

accuracy = accuracy_score(y_true, y_pred)
precision = precision_score(y_true, y_pred)
recall = recall_score(y_true, y_pred)
f1 = f1_score(y_true, y_pred)

print(f'정확도 Accuracy : {accuracy*100:.2f} %')
print(f'정밀도 Precision : {precision*100:.2f} %')
print(f'재현율 Recall : {recall*100:.2f} %')
print(f'F1 : {f1*100:.2f} %')

 

 

Q. 정확도에 비해 정밀도와 재현율, F1 점수가 현저히 낮은 이유는 무엇일까?

## confusion matrix 그려보기
from sklearn.metrics import confusion_matrix

result_cm = confusion_matrix(y_true, y_pred)

plt.figure(figsize=(6, 6))

plt.matshow(result_cm, fignum=1)
plt.xticks(range(2), [0, 1])
plt.yticks(range(2), [0, 1])
plt.ylabel('True label')
plt.xlabel('Predicted label')
plt.colorbar()

for (i, j), val in np.ndenumerate(result_cm):
    plt.text(j, i, f'{val}', ha='center', va='center', color='red')

plt.show()

► 알 수 있는 정보

애초에 양성(=이상치)이 34개밖에 없었기 때문에 정밀도와 재현율, F1점수가 작은 것