특성 엔지니어링 (Feature Engineering)

원본 데이터를 모델이 학습하기 좋은 형태로 변환하는 과정. 모델 성능의 핵심 결정 요인.

왜 특성 엔지니어링이 중요한가

"데이터와 특성이 머신러닝의 한계를 결정하고,
 알고리즘은 그 한계에 얼마나 가까이 갈 수 있는지를 결정한다."
 - Pedro Domingos

단계	영향도
좋은 데이터	높음
좋은 특성	높음
좋은 알고리즘	중간
하이퍼파라미터 튜닝	낮음

수치형 특성 처리

스케일링 (Scaling)

from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler
import numpy as np

# 1. Standardization (Z-score)
# μ=0, σ=1로 변환
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# X_scaled = (X - μ) / σ

# 2. Min-Max Scaling
# [0, 1] 범위로 변환
scaler = MinMaxScaler()
X_scaled = scaler.fit_transform(X)
# X_scaled = (X - min) / (max - min)

# 3. Robust Scaling
# 이상치에 강건 (중앙값과 IQR 사용)
scaler = RobustScaler()
X_scaled = scaler.fit_transform(X)
# X_scaled = (X - median) / IQR

스케일링 선택 가이드:

스케일러	사용 상황
StandardScaler	정규분포, 일반적
MinMaxScaler	경계가 중요할 때, Neural Net
RobustScaler	이상치가 많을 때
MaxAbsScaler	희소 행렬

변환 (Transformation)

from sklearn.preprocessing import PowerTransformer, QuantileTransformer
import numpy as np

# 1. Log 변환 (오른쪽 꼬리 분포)
X_log = np.log1p(X)  # log(1+x) for x >= 0

# 2. Box-Cox 변환 (양수 데이터만)
from scipy import stats
X_boxcox, lambda_ = stats.boxcox(X)

# 3. Yeo-Johnson 변환 (음수도 가능)
pt = PowerTransformer(method='yeo-johnson')
X_transformed = pt.fit_transform(X)

# 4. Quantile 변환 (균등/정규 분포로)
qt = QuantileTransformer(output_distribution='normal')
X_transformed = qt.fit_transform(X)

이산화 (Binning)

연속 변수를 범주형으로 변환.

from sklearn.preprocessing import KBinsDiscretizer
import pandas as pd

# 1. 등간격 비닝
discretizer = KBinsDiscretizer(n_bins=5, encode='ordinal', strategy='uniform')
X_binned = discretizer.fit_transform(X)

# 2. 등빈도 비닝 (각 빈에 동일한 샘플 수)
discretizer = KBinsDiscretizer(n_bins=5, encode='ordinal', strategy='quantile')
X_binned = discretizer.fit_transform(X)

# 3. 커스텀 빈 경계
df['age_group'] = pd.cut(df['age'], 
                         bins=[0, 18, 35, 50, 65, 100],
                         labels=['child', 'young', 'middle', 'senior', 'elderly'])

범주형 특성 처리

인코딩 (Encoding)

from sklearn.preprocessing import LabelEncoder, OneHotEncoder, OrdinalEncoder
import pandas as pd

# 1. Label Encoding (순서가 있는 범주)
le = LabelEncoder()
y_encoded = le.fit_transform(y)
# ['low', 'medium', 'high'] -> [0, 1, 2]

# 2. Ordinal Encoding (지정된 순서)
oe = OrdinalEncoder(categories=[['low', 'medium', 'high']])
X_encoded = oe.fit_transform(X)

# 3. One-Hot Encoding (순서 없는 범주)
ohe = OneHotEncoder(sparse_output=False, drop='first')  # drop으로 다중공선성 방지
X_encoded = ohe.fit_transform(X)

# Pandas get_dummies
df_encoded = pd.get_dummies(df, columns=['category'], drop_first=True)

타겟 인코딩 (Target Encoding)

범주별 타겟 평균으로 인코딩. 과적합 주의.

from sklearn.model_selection import KFold
import numpy as np

def target_encode(df, col, target, smoothing=1.0):
    """타겟 인코딩 with smoothing"""
    global_mean = df[target].mean()
    agg = df.groupby(col)[target].agg(['mean', 'count'])

    # Smoothing: 샘플 수가 적으면 전역 평균에 가깝게
    smooth = 1 / (1 + np.exp(-(agg['count'] - smoothing)))
    agg['smoothed'] = smooth * agg['mean'] + (1 - smooth) * global_mean

    return df[col].map(agg['smoothed'])

# K-Fold로 과적합 방지
def target_encode_kfold(df, col, target, n_folds=5):
    encoded = np.zeros(len(df))
    kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)

    for train_idx, val_idx in kf.split(df):
        mean_map = df.iloc[train_idx].groupby(col)[target].mean()
        encoded[val_idx] = df.iloc[val_idx][col].map(mean_map)

    # NaN은 전역 평균으로 대체
    global_mean = df[target].mean()
    encoded = np.where(np.isnan(encoded), global_mean, encoded)

    return encoded

빈도 인코딩

def frequency_encode(df, col):
    """빈도 인코딩"""
    freq = df[col].value_counts(normalize=True)
    return df[col].map(freq)

해시 인코딩 (고카디널리티)

from sklearn.feature_extraction import FeatureHasher

# 해시 벡터화 (큰 카디널리티에 유용)
hasher = FeatureHasher(n_features=100, input_type='string')
X_hashed = hasher.transform(df['high_cardinality_col'].values.reshape(-1, 1))

텍스트 특성

기본 텍스트 특성

from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer

# 1. Bag of Words
count_vec = CountVectorizer(max_features=1000, stop_words='english')
X_bow = count_vec.fit_transform(texts)

# 2. TF-IDF
tfidf_vec = TfidfVectorizer(max_features=1000, ngram_range=(1, 2))
X_tfidf = tfidf_vec.fit_transform(texts)

텍스트 통계 특성

def text_features(text):
    """텍스트 통계 특성 추출"""
    return {
        'char_count': len(text),
        'word_count': len(text.split()),
        'avg_word_length': np.mean([len(w) for w in text.split()]),
        'unique_words': len(set(text.split())),
        'uppercase_ratio': sum(1 for c in text if c.isupper()) / len(text),
        'digit_ratio': sum(1 for c in text if c.isdigit()) / len(text),
        'punctuation_count': sum(1 for c in text if c in '.,!?;:'),
    }

임베딩 (Embeddings)

# Word2Vec / FastText
from gensim.models import Word2Vec

sentences = [text.split() for text in texts]
model = Word2Vec(sentences, vector_size=100, window=5, min_count=1)

def get_sentence_embedding(text, model):
    words = text.split()
    vectors = [model.wv[w] for w in words if w in model.wv]
    return np.mean(vectors, axis=0) if vectors else np.zeros(model.vector_size)

# Transformer 임베딩
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(texts)

시계열 특성

시간 기반 특성

def datetime_features(df, date_col):
    """날짜/시간 특성 추출"""
    df = df.copy()

    df['year'] = df[date_col].dt.year
    df['month'] = df[date_col].dt.month
    df['day'] = df[date_col].dt.day
    df['dayofweek'] = df[date_col].dt.dayofweek
    df['hour'] = df[date_col].dt.hour
    df['is_weekend'] = df[date_col].dt.dayofweek >= 5
    df['is_month_start'] = df[date_col].dt.is_month_start
    df['is_month_end'] = df[date_col].dt.is_month_end
    df['quarter'] = df[date_col].dt.quarter

    # 주기적 인코딩 (연속성 유지)
    df['month_sin'] = np.sin(2 * np.pi * df['month'] / 12)
    df['month_cos'] = np.cos(2 * np.pi * df['month'] / 12)
    df['hour_sin'] = np.sin(2 * np.pi * df['hour'] / 24)
    df['hour_cos'] = np.cos(2 * np.pi * df['hour'] / 24)

    return df

지연 특성 (Lag Features)

def create_lag_features(df, col, lags=[1, 7, 30]):
    """지연 특성 생성"""
    for lag in lags:
        df[f'{col}_lag_{lag}'] = df[col].shift(lag)
    return df

def create_rolling_features(df, col, windows=[7, 30]):
    """롤링 통계 특성"""
    for window in windows:
        df[f'{col}_rolling_mean_{window}'] = df[col].rolling(window).mean()
        df[f'{col}_rolling_std_{window}'] = df[col].rolling(window).std()
        df[f'{col}_rolling_min_{window}'] = df[col].rolling(window).min()
        df[f'{col}_rolling_max_{window}'] = df[col].rolling(window).max()
    return df

특성 선택 (Feature Selection)

필터 방법

from sklearn.feature_selection import SelectKBest, f_classif, mutual_info_classif

# 1. 분산 기반
from sklearn.feature_selection import VarianceThreshold
selector = VarianceThreshold(threshold=0.01)
X_selected = selector.fit_transform(X)

# 2. 통계 검정 기반
selector = SelectKBest(score_func=f_classif, k=10)
X_selected = selector.fit_transform(X, y)

# 3. 상호 정보량
selector = SelectKBest(score_func=mutual_info_classif, k=10)
X_selected = selector.fit_transform(X, y)

# 4. 상관관계 기반
def remove_highly_correlated(df, threshold=0.9):
    corr_matrix = df.corr().abs()
    upper = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(bool))
    to_drop = [col for col in upper.columns if any(upper[col] > threshold)]
    return df.drop(columns=to_drop)

래퍼 방법

from sklearn.feature_selection import RFE, RFECV

# Recursive Feature Elimination
rfe = RFE(estimator=RandomForestClassifier(), n_features_to_select=10)
X_selected = rfe.fit_transform(X, y)

# RFE with Cross-Validation
rfecv = RFECV(estimator=RandomForestClassifier(), cv=5, scoring='accuracy')
X_selected = rfecv.fit_transform(X, y)
print(f"최적 특성 수: {rfecv.n_features_}")

임베디드 방법

from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LassoCV

# 트리 기반 특성 중요도
model = RandomForestClassifier()
model.fit(X, y)

importance = pd.DataFrame({
    'feature': feature_names,
    'importance': model.feature_importances_
}).sort_values('importance', ascending=False)

# L1 정규화 (Lasso)
lasso = LassoCV(cv=5)
lasso.fit(X, y)
selected_features = np.array(feature_names)[lasso.coef_ != 0]

특성 조합

다항 특성

from sklearn.preprocessing import PolynomialFeatures

poly = PolynomialFeatures(degree=2, interaction_only=True, include_bias=False)
X_poly = poly.fit_transform(X)

도메인 지식 기반

# 금융 예시
df['debt_to_income'] = df['debt'] / df['income']
df['credit_utilization'] = df['balance'] / df['credit_limit']

# 이커머스 예시
df['avg_order_value'] = df['total_revenue'] / df['num_orders']
df['items_per_order'] = df['total_items'] / df['num_orders']

파이프라인 구성

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer

# 전처리 파이프라인
numeric_features = ['age', 'income']
categorical_features = ['gender', 'education']

numeric_transformer = Pipeline([
    ('scaler', StandardScaler()),
    ('poly', PolynomialFeatures(degree=2))
])

categorical_transformer = Pipeline([
    ('onehot', OneHotEncoder(drop='first', handle_unknown='ignore'))
])

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

# 전체 파이프라인
model = Pipeline([
    ('preprocessor', preprocessor),
    ('classifier', RandomForestClassifier())
])

model.fit(X_train, y_train)

흔히 하는 실수

1. 테스트 데이터로 fit (데이터 누수!)

가장 치명적인 실수.

# 나쁜 예: 전체 데이터로 스케일러 학습
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)  # 테스트 정보 누출!
X_train, X_test = train_test_split(X_scaled, ...)

# 좋은 예: 훈련 데이터만으로 fit
X_train, X_test, y_train, y_test = train_test_split(X, y)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)  # 훈련만으로 fit
X_test_scaled = scaler.transform(X_test)        # 변환만

2. 타겟 인코딩에서 데이터 누수

# 나쁜 예: 전체 데이터의 타겟 평균 사용
df['category_encoded'] = df.groupby('category')['target'].transform('mean')

# 좋은 예: K-Fold로 out-of-fold 인코딩
def safe_target_encode(df, col, target, n_folds=5):
    encoded = np.zeros(len(df))
    kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)

    for train_idx, val_idx in kf.split(df):
        mean_map = df.iloc[train_idx].groupby(col)[target].mean()
        encoded[val_idx] = df.iloc[val_idx][col].map(mean_map)

    return encoded

3. 결측치 처리 전 분할

# 나쁜 예: 전체 데이터 평균으로 채움
df['col'].fillna(df['col'].mean(), inplace=True)  # 테스트 정보 누출
X_train, X_test = train_test_split(df)

# 좋은 예: 훈련 데이터 통계로 채움
X_train, X_test = train_test_split(df)
train_mean = X_train['col'].mean()
X_train['col'].fillna(train_mean, inplace=True)
X_test['col'].fillna(train_mean, inplace=True)  # 훈련 평균 사용

4. 스케일링 방법 잘못 선택

# 나쁜 예: 이상치 있는 데이터에 StandardScaler
# StandardScaler는 평균/표준편차를 쓰므로 이상치에 민감

# 좋은 예: 이상치에 강건한 RobustScaler
from sklearn.preprocessing import RobustScaler
scaler = RobustScaler()  # 중앙값과 IQR 사용

5. 범주형 변수에 Label Encoding 후 트리 외 모델 사용

# 나쁜 예: 순서 없는 범주에 숫자 부여 후 선형 모델
le = LabelEncoder()
df['color_encoded'] = le.fit_transform(df['color'])  # red=0, blue=1, green=2
model = LogisticRegression()  # 0 < 1 < 2 관계 가정

# 좋은 예: One-Hot Encoding
df_encoded = pd.get_dummies(df, columns=['color'])
# 또는 트리 기반 모델 사용 (Label Encoding OK)

6. 고카디널리티 범주에 One-Hot Encoding

# 나쁜 예: 10000개 카테고리를 one-hot
df_encoded = pd.get_dummies(df, columns=['user_id'])  # 10000개 컬럼!

# 좋은 예: 대안 사용
# 1. Target Encoding
# 2. Frequency Encoding
# 3. Embedding (딥러닝)
# 4. Hash Encoding

7. 특성 선택 없이 모든 특성 사용

# 나쁜 예: 수백 개 특성 그대로
model.fit(X_all_features, y)

# 좋은 예: 특성 선택 수행
# 1. 상관관계 높은 특성 제거
# 2. 분산 낮은 특성 제거
# 3. 모델 기반 선택 (RFE, 특성 중요도)
from sklearn.feature_selection import SelectFromModel
selector = SelectFromModel(RandomForestClassifier(), threshold='median')
X_selected = selector.fit_transform(X, y)

8. 특성 생성 시 미래 정보 사용 (시계열)

# 나쁜 예: 미래 데이터로 rolling 계산
df['rolling_mean'] = df['value'].rolling(7).mean()  # 현재 포함

# 좋은 예: shift로 과거만 사용
df['rolling_mean'] = df['value'].shift(1).rolling(7).mean()  # 현재 제외

9. 텍스트 전처리 불일치

# 나쁜 예: 훈련과 추론에서 다른 전처리
# 훈련: 소문자 변환
# 추론: 소문자 변환 빼먹음

# 좋은 예: 전처리 함수화 및 재사용
def preprocess_text(text):
    text = text.lower()
    text = re.sub(r'[^\w\s]', '', text)
    return text.strip()

# 훈련과 추론 모두 같은 함수 사용

10. 다항 특성 과도하게 생성

# 나쁜 예: 고차 다항식 무분별 사용
poly = PolynomialFeatures(degree=5)  # 특성 폭발
X_poly = poly.fit_transform(X)

# 좋은 예: 상호작용만 또는 낮은 차수
poly = PolynomialFeatures(degree=2, interaction_only=True)
# 또는 도메인 지식 기반으로 선택적 생성

11. 수치형-범주형 혼합 처리 실수

# 나쁜 예: 모든 컬럼에 같은 처리
scaler.fit_transform(df)  # 범주형 컬럼 포함 에러

# 좋은 예: ColumnTransformer 사용
from sklearn.compose import ColumnTransformer

preprocessor = ColumnTransformer([
    ('num', StandardScaler(), numeric_cols),
    ('cat', OneHotEncoder(), categorical_cols)
])
X_processed = preprocessor.fit_transform(df)

12. 파이프라인 없이 전처리

# 나쁜 예: 수동으로 단계별 처리 (실수 가능성 높음)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
selector = SelectKBest()
X_train_selected = selector.fit_transform(X_train_scaled, y_train)
model.fit(X_train_selected, y_train)
# 테스트 시 같은 순서로 해야 하는데...

# 좋은 예: Pipeline으로 캡슐화
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('selector', SelectKBest(k=10)),
    ('model', RandomForestClassifier())
])
pipeline.fit(X_train, y_train)
pipeline.predict(X_test)  # 자동으로 같은 전처리

참고 자료

• Scikit-learn Preprocessing
• Feature Engineering for Machine Learning
• Kaggle Feature Engineering Course
• Common Pitfalls and Recommended Practices (Scikit-learn)