결측치 처리 전략: 삭제부터 다중 대체(Multiple Imputation)까지

예시: 설문 응답을 입력하는 직원이 랜덤하게 타이핑 실수를 해서 값이 빠짐
     센서가 무작위로 간헐적 오작동

특징: 결측이 있는 행과 없는 행의 분포가 동일
처리: 삭제해도 편향이 생기지 않음 (데이터 손실만 문제)

예시: 고소득자일수록 소득 항목을 비워두는 경향
     → 결측 여부가 '직업' 변수와 상관 (직업은 관측됨)

     젊은 사용자일수록 주소를 입력하지 않음
     → 결측 여부가 '나이' 변수와 상관 (나이는 관측됨)

특징: 결측 패턴을 다른 관측 변수로 예측할 수 있음
처리: 단순 삭제하면 편향 발생! 대체(Imputation)가 필요

예시: 체중이 많이 나가는 사람일수록 체중 항목을 비워둠
     → 결측 여부가 체중 값 자체에 의존 (체중은 관측 안 됨)

     우울증이 심한 환자일수록 설문에 응답하지 않음
     → 결측 여부가 우울증 심각도에 의존

특징: 결측 패턴을 관측된 데이터만으로는 설명할 수 없음
처리: 어떤 통계적 방법으로도 완전히 보정 불가. 도메인 지식 필요

import pandas as pd

df = pd.read_csv('data.csv')

# 컬럼별 결측 수와 비율
missing = df.isnull().sum()
missing_pct = (missing / len(df) * 100).round(1)

missing_info = pd.DataFrame({
    'missing_count': missing,
    'missing_pct': missing_pct
}).sort_values('missing_pct', ascending=False)

print(missing_info[missing_info['missing_count'] > 0])

               missing_count  missing_pct
cabin                    687         77.1
age                      177         19.9
embarked                   2          0.2

import missingno as msno
import matplotlib.pyplot as plt

# 매트릭스 플롯: 흰색이 결측
msno.matrix(df)
plt.show()

# 히트맵: 결측 간 상관관계
msno.heatmap(df)
plt.show()

# 막대 그래프: 컬럼별 비결측 수
msno.bar(df)
plt.show()

# 결측이 하나라도 있는 행을 모두 삭제
df_clean = df.dropna()

# 특정 컬럼 기준으로만 삭제
df_clean = df.dropna(subset=['age', 'embarked'])

print(f"원본: {len(df)}행 → 삭제 후: {len(df_clean)}행")
# 원본: 891행 → 삭제 후: 714행 (subset 사용 시)
# 원본: 891행 → 삭제 후: 183행 (전체 dropna 시)

삭제가 안전한 조건:
[1] MCAR인 경우 (삭제해도 편향 없음)
[2] 결측 비율이 매우 낮은 경우 (5% 이하)
[3] 데이터가 충분히 많은 경우 (삭제 후에도 샘플 수 충분)
[4] 결측 행이 소수이고 특이값인 경우

삭제가 위험한 조건:
[1] MAR/MNAR인 경우 → 편향 발생
[2] 결측 비율이 높은 경우 → 정보 손실 과다
[3] 데이터가 적은 경우 → 통계적 검정력 저하
[4] 여러 컬럼에 산발적 결측 → 행 삭제 시 대부분의 행이 사라짐

# 결측 비율 50% 이상인 컬럼 제거
threshold = 0.5
cols_to_drop = missing_pct[missing_pct > threshold * 100].index
df_clean = df.drop(columns=cols_to_drop)

print(f"제거된 컬럼: {list(cols_to_drop)}")
# 제거된 컬럼: ['cabin']

from sklearn.impute import SimpleImputer
import numpy as np

# 수치형: 평균으로 대체
mean_imputer = SimpleImputer(strategy='mean')
df['age_imputed'] = mean_imputer.fit_transform(df[['age']])

# 수치형: 중앙값으로 대체
median_imputer = SimpleImputer(strategy='median')
df['age_imputed'] = median_imputer.fit_transform(df[['age']])

# 범주형: 최빈값으로 대체
mode_imputer = SimpleImputer(strategy='most_frequent')
df['embarked_imputed'] = mode_imputer.fit_transform(df[['embarked']])

# 상수로 대체
const_imputer = SimpleImputer(strategy='constant', fill_value=0)
df['cabin_imputed'] = const_imputer.fit_transform(df[['cabin']])

평균 대체의 문제를 직관적으로 보자:

원래 분포: [20, 25, 30, 35, 40, NaN, NaN, NaN]
평균 = 30

대체 후:   [20, 25, 30, 35, 40, 30, 30, 30]

→ 분산이 줄어든다 (평균 주변에 값이 몰림)
→ 변수 간 상관관계가 왜곡된다
→ 결측 비율이 높을수록 문제가 심각해진다

from sklearn.impute import SimpleImputer
import numpy as np

# 결측 지시 변수 추가
df['age_is_missing'] = df['age'].isnull().astype(int)

# 그 다음 대체 수행
imputer = SimpleImputer(strategy='median')
df['age_imputed'] = imputer.fit_transform(df[['age']])

# sklearn의 MissingIndicator 활용
from sklearn.impute import MissingIndicator

indicator = MissingIndicator()
missing_flags = indicator.fit_transform(df[['age', 'cabin', 'embarked']])

# True/False 배열 → 어떤 컬럼에 결측이 있었는지 기록

효과적인 경우:
- MNAR이 의심될 때 (결측 자체가 정보)
- 트리 기반 모델과 함께 사용할 때 (분기 조건으로 활용 가능)
- 결측 비율이 10~50% 사이일 때

효과가 없는 경우:
- MCAR일 때 (결측이 무작위이므로 정보 없음)
- 결측 비율이 너무 낮을 때 (변수의 분산이 거의 0)

결측이 있는 행 → 다른 (결측 없는) 피처들로 거리 계산
                → 가장 가까운 K개 이웃 찾기
                → 이웃들의 해당 피처 값 평균으로 대체

from sklearn.impute import KNNImputer

# K=5인 KNN Imputer
knn_imputer = KNNImputer(n_neighbors=5, weights='distance')

# 수치형 컬럼만 선택
numeric_cols = df.select_dtypes(include=[np.number]).columns
df_imputed = pd.DataFrame(
    knn_imputer.fit_transform(df[numeric_cols]),
    columns=numeric_cols
)

KNN Imputer 사용 시 체크리스트:

[1] 스케일링 필수 — KNN은 거리 기반이다. 피처 스케일이 다르면
    거리가 왜곡된다. StandardScaler 또는 MinMaxScaler를 먼저 적용.
    (/ml/feature-scaling/ 참고)

[2] 범주형 변수 처리 — KNN Imputer는 수치형만 지원한다.
    범주형은 먼저 인코딩하거나 (/ml/categorical-encoding/),
    별도 SimpleImputer(mode)로 처리한다.

[3] 계산 비용 — 모든 결측 행에 대해 전체 데이터와 거리를 계산한다.
    데이터가 10만 행 이상이면 느려진다.

[4] K 값 선택 — 너무 작으면 노이즈에 민감, 너무 크면 평균 대체와 비슷해짐.
    보통 5~10이 적절.

Round 1:
  age를 [sex, fare, pclass, embarked]로 예측하는 모델 학습 → age 결측 대체
  fare를 [sex, age(대체됨), pclass, embarked]로 예측 → fare 결측 대체
  ...

Round 2:
  age를 [sex, fare(대체됨), pclass, embarked]로 다시 예측 → age 대체값 업데이트
  fare를 [sex, age(업데이트), pclass, embarked]로 다시 예측 → fare 대체값 업데이트
  ...

Round N: 대체값이 수렴할 때까지 반복

from sklearn.experimental import enable_iterative_imputer  # 필수!
from sklearn.impute import IterativeImputer
from sklearn.linear_model import BayesianRidge

# 기본 추정기는 BayesianRidge
iterative_imputer = IterativeImputer(
    estimator=BayesianRidge(),
    max_iter=10,
    random_state=42
)

df_imputed = pd.DataFrame(
    iterative_imputer.fit_transform(df[numeric_cols]),
    columns=numeric_cols
)

from sklearn.ensemble import RandomForestRegressor
from sklearn.linear_model import BayesianRidge

# 선형 관계 → BayesianRidge (기본, 빠름)
imp_linear = IterativeImputer(estimator=BayesianRidge())

# 비선형 관계 → RandomForest (느리지만 정확)
imp_forest = IterativeImputer(
    estimator=RandomForestRegressor(n_estimators=100, random_state=42),
    max_iter=10,
    random_state=42
)

강점:
[1] 변수 간 관계를 보존 — 상관관계 구조가 유지됨
[2] MAR 가정에서 이론적으로 가장 정확한 방법 중 하나
[3] 유연함 — 추정기를 자유롭게 교체 가능
[4] 불확실성 반영 가능 — 여러 번 대체해서 결과의 변동성을 측정

한계:
[1] 계산 비용이 큼 — 피처 수 × 반복 횟수만큼 모델 학습
[2] MNAR에서는 여전히 편향 발생
[3] 수렴이 보장되지 않을 수 있음 (max_iter를 충분히 설정)
[4] 범주형 변수 처리가 복잡 — 사전 인코딩 필요

# Forward Fill: 직전 값으로 채움
df['temperature'] = df['temperature'].ffill()

# Backward Fill: 다음 값으로 채움
df['temperature'] = df['temperature'].bfill()

# 둘 다 적용: forward fill 후 남은 결측을 backward fill
df['temperature'] = df['temperature'].ffill().bfill()

시간  온도
08:00  20.0
09:00  NaN   → ffill: 20.0 (08:00 값)
10:00  NaN   → ffill: 20.0 (08:00 값)
11:00  22.5
12:00  NaN   → ffill: 22.5 (11:00 값)

# 선형 보간: 두 관측값 사이를 직선으로 연결
df['temperature'] = df['temperature'].interpolate(method='linear')

# 시간 기반 보간: 시간 간격을 고려
df['temperature'] = df['temperature'].interpolate(method='time')

# 다항식 보간
df['temperature'] = df['temperature'].interpolate(method='polynomial', order=2)

시간  온도
08:00  20.0
09:00  NaN   → 선형 보간: 20.83
10:00  NaN   → 선형 보간: 21.67
11:00  22.5

| 방법          | 가정                     | 적합한 상황              |
|--------------|-------------------------|------------------------|
| Forward Fill | 값이 변할 때까지 유지     | 주가, 재고, 상태 데이터    |
| Backward Fill| 미래 값으로 소급          | 사후 분석 (실시간 X)      |
| 선형 보간     | 두 점 사이 직선 변화      | 온도, 센서 (연속적 변화)   |
| 시간 보간     | 시간 간격 비례 변화       | 불균등 시간 간격 데이터    |
| 이동 평균     | 주변 값의 평균            | 노이즈가 많은 시계열      |

import xgboost as xgb
import numpy as np

# NaN이 포함된 데이터를 그대로 학습 가능
X_train_with_nan = X_train.copy()  # NaN 포함
model = xgb.XGBClassifier(n_estimators=100)
model.fit(X_train_with_nan, y_train)  # 에러 없이 작동

import lightgbm as lgb

model = lgb.LGBMClassifier(n_estimators=100)
model.fit(X_train_with_nan, y_train)  # NaN 그대로 학습

트리 모델의 자체 처리 vs 사전 대체:

실험적으로, 트리 모델에서는:
- 단순 평균 대체 → 자체 처리보다 성능이 같거나 나빠질 수 있음
- KNN/MICE 대체 → 약간의 성능 향상 가능 (데이터에 따라 다름)
- 결측 지시 변수 추가 → 대부분 도움이 됨

결론: XGBoost/LightGBM을 쓸 때는 NaN을 그대로 두되,
      결측 지시 변수를 추가하는 것이 실전에서 가장 흔한 전략이다.

# 전체 데이터로 대체 → 테스트 데이터 정보가 누수!
imputer = SimpleImputer(strategy='mean')
X_imputed = imputer.fit_transform(X)  # 테스트 데이터 포함 평균

X_train, X_test = train_test_split(X_imputed, test_size=0.2)

from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score

pipe = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler()),
    ('model', LogisticRegression())
])

# 교차 검증에서 각 fold마다 imputer가 훈련 데이터만으로 fit됨
scores = cross_val_score(pipe, X, y, cv=5, scoring='accuracy')
print(f"CV Score: {scores.mean():.4f} (+/- {scores.std():.4f})")

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer, KNNImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder

numeric_features = ['age', 'fare']
categorical_features = ['embarked', 'sex']

# 수치형: KNN Imputer → 스케일링
numeric_transformer = Pipeline([
    ('imputer', KNNImputer(n_neighbors=5)),
    ('scaler', StandardScaler())
])

# 범주형: 최빈값 대체 → 원핫 인코딩
categorical_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('encoder', OneHotEncoder(handle_unknown='ignore'))
])

preprocessor = ColumnTransformer([
    ('num', numeric_transformer, numeric_features),
    ('cat', categorical_transformer, categorical_features)
])

full_pipe = Pipeline([
    ('preprocessor', preprocessor),
    ('model', LogisticRegression())
])

scores = cross_val_score(full_pipe, X, y, cv=5, scoring='accuracy')

from sklearn.datasets import fetch_openml
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer, KNNImputer
from sklearn.experimental import enable_iterative_imputer
from sklearn.impute import IterativeImputer
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
import numpy as np
import pandas as pd

# Titanic 데이터 로드
titanic = fetch_openml('titanic', version=1, as_frame=True)
X = titanic.data[['age', 'fare', 'pclass', 'sibsp', 'parch']].copy()
y = (titanic.target == '1').astype(int)

# 전략별 Pipeline 정의
strategies = {
    'Mean': Pipeline([
        ('imputer', SimpleImputer(strategy='mean')),
        ('model', LogisticRegression(max_iter=1000))
    ]),
    'Median': Pipeline([
        ('imputer', SimpleImputer(strategy='median')),
        ('model', LogisticRegression(max_iter=1000))
    ]),
    'KNN (k=5)': Pipeline([
        ('imputer', KNNImputer(n_neighbors=5)),
        ('model', LogisticRegression(max_iter=1000))
    ]),
    'Iterative (MICE)': Pipeline([
        ('imputer', IterativeImputer(max_iter=10, random_state=42)),
        ('model', LogisticRegression(max_iter=1000))
    ]),
}

# 5-fold CV로 비교
results = {}
for name, pipe in strategies.items():
    scores = cross_val_score(pipe, X, y, cv=5, scoring='accuracy')
    results[name] = f"{scores.mean():.4f} (+/- {scores.std():.4f})"
    print(f"{name:20s}: {results[name]}")

일반적인 결과 경향:

| 전략               | CV Accuracy (예시)    |
|-------------------|-----------------------|
| Mean              | 0.6923 (+/- 0.0180)  |
| Median            | 0.6930 (+/- 0.0175)  |
| KNN (k=5)         | 0.6965 (+/- 0.0162)  |
| Iterative (MICE)  | 0.6978 (+/- 0.0155)  |

관찰:
- 이 데이터에서 결측 비율이 크지 않아 차이가 작다
- 결측 비율이 높을수록 정교한 방법의 이점이 커진다
- 모델에 따라서도 결과가 달라진다 (트리 모델은 차이가 더 줄어듦)

결측 비율 확인
│
├── > 50% → 컬럼 제거 (+ 결측 지시 변수 생성)
│
├── 5~50%
│   ├── 결측 유형은?
│   │   ├── MCAR → 삭제 OK (데이터 충분하면)
│   │   ├── MAR  → 다변량 대체 권장 (KNN, MICE)
│   │   └── MNAR → 결측 지시 변수 + 단순 대체
│   │
│   └── 모델은?
│       ├── 트리 기반 (XGB/LGBM) → NaN 유지 + 지시 변수
│       ├── 선형 모델 → KNN/MICE 대체
│       └── 거리 기반 (KNN, SVM) → 대체 필수 + 스케일링
│
├── < 5% → SimpleImputer (mean/median/mode)로 충분
│
└── 시계열 → ffill / interpolate

전략별 비교 요약:

| 방법             | 복잡도 | 정확도 | 속도  | 가정           |
|-----------------|--------|--------|------|----------------|
| 삭제             | 낮음   | 낮음   | 빠름  | MCAR           |
| Mean/Median     | 낮음   | 보통   | 빠름  | MCAR           |
| 결측 지시 변수    | 낮음   | 보통+  | 빠름  | 없음           |
| KNN Imputer     | 중간   | 높음   | 중간  | MAR, 유사성     |
| Iterative (MICE)| 높음   | 높음   | 느림  | MAR            |
| ffill/보간       | 낮음   | 높음   | 빠름  | 시간적 연속성   |
| 트리 모델 자체    | 없음   | 높음   | 빠름  | 없음           |

원본 데이터 → [결측 처리] → [인코딩] → [스케일링] → [피처 선택] → 학습 준비 완료
     │           │            │           │            │
     │        이 글        #30, #33      #31          #32
     │
     └── 범주형, 수치형, 결측 섞인 혼돈의 테이블