[파이썬 기초] 데이터 전처리 심화 (결측값, 이상값, 정규화, 인코딩)

실제 데이터는 결측값, 이상값, 다양한 척도와 범주형 변수가 혼재합니다.

머신러닝 모델은 이런 데이터를 그대로 받아들이지 못하는 경우가 많아, 전처리가 분석 품질을 좌우하는 핵심 단계가 됩니다.

pip install numpy pandas scikit-learn

---

결측값 처리

결측값 확인

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "sample_id": ["s01", "s02", "s03", "s04", "s05"],
    "coverage":  [30.5, np.nan, 22.1, np.nan, 45.8],
    "gc_content":[0.51, 0.49, np.nan, 0.52, 0.48],
    "species":   ["Homo sapiens", "Mus musculus", None, "Homo sapiens", "Mus musculus"]
})

print(df.isnull().sum())                          # 열별 결측값 수
print(df.isnull().sum() / len(df) * 100)          # 결측 비율 (%)
print(df[df.isnull().any(axis=1)])                # 결측값 있는 행
print(df.columns[df.isnull().any()])              # 결측값 있는 열

결측값 제거

df.dropna()                          # 결측값 있는 행 모두 제거
df.dropna(subset=["coverage"])       # 특정 열에 결측값 있는 행만 제거
df.dropna(axis=1)                    # 결측값 있는 열 제거
df.dropna(thresh=3)                  # 유효한 값이 3개 미만인 행 제거

 

결측 비율이 높은 열을 선택적으로 제거할 때는 임계값 기준으로 판단합니다.

threshold = 0.3    # 결측 비율 30% 이상이면 열 제거
df = df.loc[:, df.isnull().mean() < threshold]

결측값 대체 (Imputation)

단순 대체

# 수치형: 평균, 중앙값으로 대체
df["coverage"].fillna(df["coverage"].mean(), inplace=True)
df["coverage"].fillna(df["coverage"].median(), inplace=True)

# 범주형: 최빈값으로 대체
df["species"].fillna(df["species"].mode()[0], inplace=True)

# 시계열: 앞/뒤 값으로 대체
df["coverage"].fillna(method="ffill", inplace=True)
df["coverage"].fillna(method="bfill", inplace=True)

scikit-learn SimpleImputer

from sklearn.impute import SimpleImputer

# 수치형: 중앙값으로 대체
num_imputer = SimpleImputer(strategy="median")
X_imputed = num_imputer.fit_transform(df[["coverage", "gc_content"]])

# 범주형: 최빈값으로 대체
cat_imputer = SimpleImputer(strategy="most_frequent")
X_cat = cat_imputer.fit_transform(df[["species"]])

KNN Imputer

주변 샘플의 값을 참고해서 결측값을 대체합니다.

단순 대체보다 정교하지만 계산 비용이 높습니다.

from sklearn.impute import KNNImputer

knn_imputer = KNNImputer(n_neighbors=3)
X_knn = knn_imputer.fit_transform(df[["coverage", "gc_content"]])

---

이상값 탐지 및 처리

IQR 방법

사분위 범위를 기반으로 이상값을 탐지합니다.

coverage = np.array([28.5, 30.1, 29.8, 31.2, 28.9, 150.0, 29.5, 30.8, -5.0, 31.0])

Q1 = np.percentile(coverage, 25)
Q3 = np.percentile(coverage, 75)
IQR = Q3 - Q1

lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR

outlier_mask = (coverage < lower) | (coverage > upper)
print(f"이상값: {coverage[outlier_mask]}")     # [150.  -5.]
print(f"정상값: {coverage[~outlier_mask]}")

Pandas로 열 단위 적용

def remove_outliers_iqr(df, col):
    Q1 = df[col].quantile(0.25)
    Q3 = df[col].quantile(0.75)
    IQR = Q3 - Q1
    return df[(df[col] >= Q1 - 1.5 * IQR) & (df[col] <= Q3 + 1.5 * IQR)]

df_clean = remove_outliers_iqr(df, "coverage")

Z-score 방법

평균에서 표준편차 몇 배 떨어졌는지로 이상값을 판단합니다.

데이터가 정규분포에 가까울 때 적합합니다.

from scipy import stats

z_scores = np.abs(stats.zscore(coverage))
outlier_mask = z_scores > 3
print(f"이상값 인덱스: {np.where(outlier_mask)[0]}")

Isolation Forest

머신러닝 기반 이상값 탐지 방법입니다.

고차원 데이터나 복잡한 분포에서도 잘 작동합니다.

from sklearn.ensemble import IsolationForest

np.random.seed(42)
X = np.random.randn(100, 2)
X = np.vstack([X, [[10, 10], [-10, -10]]])    # 이상값 2개 추가

iso = IsolationForest(contamination=0.02, random_state=42)
labels = iso.fit_predict(X)    # 정상: 1, 이상: -1

print(f"이상값 수: {(labels == -1).sum()}")
X_clean = X[labels == 1]

 

contamination은 이상값 비율 추정치입니다.

사전 지식이 없으면 "auto"로 설정합니다.

이상값 처리 방법

# 1. 제거
df_clean = df[~outlier_mask]

# 2. 경계값으로 대체 (Winsorizing)
from scipy.stats.mstats import winsorize
winsorized = winsorize(coverage, limits=[0.05, 0.05])  # 상/하위 5%를 경계값으로

# 3. NaN으로 변환 후 중앙값으로 대체
coverage_clean = coverage.copy().astype(float)
coverage_clean[outlier_mask] = np.nan
coverage_clean = pd.Series(coverage_clean).fillna(
    pd.Series(coverage_clean).median()
).values

---

정규화 / 스케일링

StandardScaler (표준화)

평균 0, 표준편차 1로 변환합니다.

대부분의 머신러닝 알고리즘에서 기본으로 사용합니다.

from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

X_train, X_test = train_test_split(np.random.randn(100, 5), test_size=0.2)

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)    # transform만 적용

# 역변환
X_original = scaler.inverse_transform(X_train_scaled)

 

테스트 데이터에는 반드시 transform()만 사용합니다.

fit_transform()을 사용하면 테스트 데이터 정보가 전처리에 반영되어 데이터 누출(data leakage)이 발생합니다.

MinMaxScaler

0~1 범위로 변환합니다.

이상값에 민감합니다.

from sklearn.preprocessing import MinMaxScaler

minmax = MinMaxScaler(feature_range=(0, 1))
X_mm = minmax.fit_transform(X_train)

 

중앙값과 IQR을 기준으로 스케일링합니다. 이상값의 영향을 적게 받습니다.

from sklearn.preprocessing import RobustScaler

robust = RobustScaler()
X_robust = robust.fit_transform(X_train)

로그 변환

오른쪽으로 치우친(right-skewed) 분포를 정규분포에 가깝게 만듭니다.

유전자 발현량 데이터에서 자주 사용합니다.

expression = np.array([1, 5, 10, 100, 1000, 5000])

log2_expr  = np.log2(expression + 1)    # log2 변환 (+1은 0 값 처리)
log10_expr = np.log10(expression + 1)   # log10 변환
ln_expr    = np.log1p(expression)       # 자연로그 (log(1+x))

 

RNA-seq 발현량 데이터는 log2(TPM+1) 또는 log2(CPM+1) 변환을 적용하는 것이 일반적입니다.

스케일러 비교

스케일러	변환 방식	특징
StandardScaler	(x - 평균) / 표준편차	가장 일반적, 이상값에 민감
MinMaxScaler	(x - min) / (max - min)	0~1 범위, 이상값에 매우 민감
RobustScaler	(x - 중앙값) / IQR	이상값에 강건
로그 변환	log(x+1)	치우친 분포 보정, 발현량 데이터

---

인코딩 (Encoding)

머신러닝 모델은 문자열을 처리하지 못하므로 범주형 변수를 수치로 변환해야 합니다.

Label Encoding

범주를 정수로 변환합니다. 순서가 있는 범주형 변수에 적합합니다.

from sklearn.preprocessing import LabelEncoder

le = LabelEncoder()
species = ["Homo sapiens", "Mus musculus", "Homo sapiens", "Rattus norvegicus"]
encoded = le.fit_transform(species)
print(encoded)       # [0 1 0 2]
print(le.classes_)   # ['Homo sapiens' 'Mus musculus' 'Rattus norvegicus']

# 역변환
print(le.inverse_transform([0, 2]))

 

순서가 없는 범주형 변수에 Label Encoding을 사용하면 모델이 숫자 간 크기 관계를 잘못 해석할 수 있습니다.

이런 경우 One-Hot Encoding을 사용합니다.

 

One-Hot Encoding

각 범주를 별도의 이진(0/1) 열로 변환합니다.

순서가 없는 범주형 변수에 적합합니다.

from sklearn.preprocessing import OneHotEncoder

ohe = OneHotEncoder(sparse_output=False, handle_unknown="ignore")
X_ohe = ohe.fit_transform(df[["species"]])
print(ohe.get_feature_names_out())

# pandas get_dummies (간편하게 사용)
df_encoded = pd.get_dummies(df, columns=["species"], drop_first=False)

drop_first=True로 설정하면 더미 변수 함정(dummy variable trap)을 피할 수 있습니다. 선형 모델에서는 설정하는 것이 좋고, 트리 기반 모델에서는 필수는 아닙니다.

Ordinal Encoding

순서가 있는 범주형 변수에 사용합니다. Label Encoding과 달리 순서를 명시적으로 지정할 수 있습니다.

from sklearn.preprocessing import OrdinalEncoder

quality = [["low"], ["medium"], ["high"], ["medium"], ["low"]]
order = [["low", "medium", "high"]]

oe = OrdinalEncoder(categories=order)
encoded = oe.fit_transform(quality)
print(encoded.ravel())    # [0. 1. 2. 1. 0.]

Target Encoding

범주별 타겟 변수의 평균값으로 인코딩합니다. 범주 수가 많은 변수에 유용합니다. 과적합 위험이 있으므로 교차 검증과 함께 사용해야 합니다.

# pandas로 간단히 구현
target_mean = df.groupby("species")["coverage"].mean()
df["species_encoded"] = df["species"].map(target_mean)

---

ColumnTransformer: 열 유형별 일괄 처리

수치형 열과 범주형 열에 다른 전처리를 적용할 때 사용합니다.

Pipeline과 조합하면 교차 검증 시 데이터 누출을 방지할 수 있습니다.

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

numeric_features = ["coverage", "gc_content"]
categorical_features = ["species"]

numeric_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

categorical_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OneHotEncoder(handle_unknown="ignore"))
])

preprocessor = ColumnTransformer([
    ("num", numeric_pipeline, numeric_features),
    ("cat", categorical_pipeline, categorical_features)
])

full_pipeline = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", RandomForestClassifier(n_estimators=100, random_state=42))
])

---

전처리 적용 순서

순서	단계	설명
1	데이터 탐색	결측값, 분포, 이상값 파악
2	결측값 처리	제거 또는 대체
3	이상값 처리	탐지 후 제거, 대체, 유지
4	인코딩	범주형 변수를 수치로 변환
5	스케일링	수치형 변수의 척도 통일
6	특성 선택/생성	불필요한 특성 제거, 새 특성 생성

 

스케일링은 학습 데이터로만 fit하고 테스트 데이터에는 transform만 적용하는 원칙을 항상 지켜야 합니다.