1. 기술 감사 및 비즈니스 맥락

분석 대상은 7건이며, 실제 모델링에 사용할 수 있는 타깃 데이터는 churned가 결측인 C004를 제외한 6건입니다. 표본 수가 매우 작으므로 아래 통계는 운영 규칙 수립을 위한 탐색 결과로만 사용해야 합니다.

추가 데이터 스멜은 다음과 같습니다.

• C006의 age=-5는 결측 또는 입력 오류로 처리해야 합니다.
• monthly_fee는 원화 정수형으로 보이며, 모델링 전에 단위와 통화 기준을 명시해야 합니다.
• C002의 last_login 결측은 단순 누락이 아니라 실제 미로그인 상태일 수도 있습니다. 따라서 결측값 자체가 이탈 신호일 가능성이 있습니다.
• 현재 관측 기간이 2026년 8월이므로, 실제 이탈 예측에서는 last_login이 예측 시점 이후의 정보가 아닌지 반드시 확인해야 합니다.

2. 통계적 처리 전략

결측값 대체

• age: 중앙값 대체
  연령은 비정상값과 이상치에 영향을 받을 수 있으므로 평균보다 중앙값이 안전합니다.
• monthly_fee: 중앙값 대체
  요금제별 대체가 더 정확할 수 있으나, 현재 표본이 작으므로 우선 전체 중앙값을 사용합니다.
• last_login 파생 변수: 중앙값 대체 + 결측 여부 indicator 추가
  “로그인 기록 없음” 자체가 중요한 고객 행동 신호일 수 있습니다.
• churned: 대체하지 않고 해당 행을 학습 데이터에서 제외
  타깃을 임의로 대체하면 라벨 노이즈와 데이터 편향이 발생합니다.

인코딩

• gender, plan, region: One-Hot Encoding
• Label Encoding은 범주 간 순서가 없는데도 인위적인 순서를 부여하므로 사용하지 않습니다.
• 운영 환경에서 신규 범주가 발생할 수 있으므로 handle_unknown="ignore"를 적용합니다.

스케일링

• 수치형 변수에는 RobustScaler 사용
• age, monthly_fee, 날짜 파생 변수는 이상치 또는 비대칭 분포가 발생할 수 있습니다.
• 평균과 표준편차에 민감한 StandardScaler보다 중앙값과 사분위 범위를 사용하는 RobustScaler가 적합합니다.

데이터 누수 방지

• churned를 먼저 분리합니다.
• 전처리 파이프라인은 학습 데이터에 대해서만 fit()합니다.
• 검증·테스트·배치 데이터에는 동일 파이프라인으로 transform()만 수행합니다.
• 시간 기반 이탈 예측이라면 랜덤 분할보다 가입일 또는 예측 시점 기준의 시간 분할을 권장합니다.

3. 구현 블록

"""
Customer churn preprocessing pipeline.

Expected input columns:
    customer_id, signup_date, last_login, age, gender,
    plan, monthly_fee, region, churned
"""

from __future__ import annotations

from typing import Any

import numpy as np
import pandas as pd
from scipy import sparse
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, RobustScaler


NUMERIC_COLUMNS = [
    "age",
    "monthly_fee",
    "signup_year",
    "signup_month",
    "signup_dayofweek",
    "last_login_year",
    "last_login_month",
    "last_login_dayofweek",
    "days_to_last_login",
]

CATEGORICAL_COLUMNS = [
    "gender",
    "plan",
    "region",
]


class BusinessCleaner(BaseEstimator, TransformerMixin):
    """Apply business rules and create model-ready date features."""

    DATE_COLUMNS = ("signup_date", "last_login")

    @staticmethod
    def _parse_date(series: pd.Series) -> pd.Series:
        """Parse mixed date formats while converting invalid dates to NaT."""
        try:
            return pd.to_datetime(
                series,
                format="mixed",
                errors="coerce",
            )
        except TypeError:
            # Compatibility with older pandas versions.
            return pd.to_datetime(series, errors="coerce")

    def fit(
        self,
        X: pd.DataFrame,
        y: pd.Series | None = None,
    ) -> "BusinessCleaner":
        """No statistics are learned here, preventing target leakage."""
        return self

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        """Clean raw records and generate date-derived features."""
        data = X.copy()

        # Identifier is not a predictive feature.
        data = data.drop(columns=["customer_id"], errors="ignore")

        # Normalize categorical values.
        if "gender" in data.columns:
            data["gender"] = (
                data["gender"]
                .astype("string")
                .str.strip()
                .str.upper()
                .replace({"MALE": "M", "FEMALE": "F"})
            )

        if "plan" in data.columns:
            data["plan"] = (
                data["plan"]
                .astype("string")
                .str.strip()
                .str.title()
            )

        if "region" in data.columns:
            data["region"] = (
                data["region"]
                .astype("string")
                .str.strip()
                .str.lower()
                .str.title()
            )

        # Convert numeric columns and invalidate impossible values.
        data["age"] = pd.to_numeric(data["age"], errors="coerce")
        data.loc[~data["age"].between(0, 120), "age"] = np.nan

        data["monthly_fee"] = pd.to_numeric(
            data["monthly_fee"],
            errors="coerce",
        )
        data.loc[data["monthly_fee"] < 0, "monthly_fee"] = np.nan

        # Create numeric date features and remove raw date strings.
        signup = self._parse_date(data["signup_date"])
        last_login = self._parse_date(data["last_login"])

        data["signup_year"] = signup.dt.year
        data["signup_month"] = signup.dt.month
        data["signup_dayofweek"] = signup.dt.dayofweek

        data["last_login_year"] = last_login.dt.year
        data["last_login_month"] = last_login.dt.month
        data["last_login_dayofweek"] = last_login.dt.dayofweek

        data["days_to_last_login"] = (
            last_login - signup
        ).dt.days

        data = data.drop(
            columns=["signup_date", "last_login"],
            errors="ignore",
        )

        # Use compact numerical types where possible.
        for column in NUMERIC_COLUMNS:
            if column in data.columns:
                data[column] = pd.to_numeric(
                    data[column],
                    errors="coerce",
                    downcast="float",
                )

        return data


def make_one_hot_encoder() -> OneHotEncoder:
    """Create a version-compatible memory-efficient encoder."""
    try:
        return OneHotEncoder(
            handle_unknown="ignore",
            sparse_output=True,
            dtype=np.float32,
        )
    except TypeError:
        # Compatibility with scikit-learn versions before 1.2.
        return OneHotEncoder(
            handle_unknown="ignore",
            sparse=True,
            dtype=np.float32,
        )


def build_preprocessing_pipeline() -> Pipeline:
    """Build the complete leakage-safe preprocessing pipeline."""
    numeric_pipeline = Pipeline(
        steps=[
            (
                "imputer",
                SimpleImputer(
                    strategy="median",
                    add_indicator=True,
                ),
            ),
            (
                "scaler",
                RobustScaler(),
            ),
        ]
    )

    categorical_pipeline = Pipeline(
        steps=[
            (
                "imputer",
                SimpleImputer(
                    strategy="most_frequent",
                ),
            ),
            (
                "encoder",
                make_one_hot_encoder(),
            ),
        ]
    )

    preprocessor = ColumnTransformer(
        transformers=[
            (
                "numeric",
                numeric_pipeline,
                NUMERIC_COLUMNS,
            ),
            (
                "categorical",
                categorical_pipeline,
                CATEGORICAL_COLUMNS,
            ),
        ],
        remainder="drop",
    )

    return Pipeline(
        steps=[
            ("business_cleaner", BusinessCleaner()),
            ("preprocessor", preprocessor),
        ]
    )


def validate_raw_target(
    target: pd.Series,
) -> pd.Series:
    """Validate and compact the binary target variable."""
    target = pd.to_numeric(target, errors="coerce")

    # Rows without labels cannot be used for supervised training.
    if target.isna().any():
        raise ValueError(
            "Target contains missing values. "
            "Remove unlabeled rows before fitting."
        )

    invalid_values = ~target.isin([0, 1])
    if invalid_values.any():
        raise ValueError(
            "Target must contain only binary values 0 or 1."
        )

    return target.astype(np.int8)


def validate_transformed_output(
    transformed: Any,
    target: pd.Series,
) -> None:
    """Assert core integrity conditions after transformation."""
    assert transformed.shape[0] == target.shape[0]
    assert transformed.shape[1] > 0

    if sparse.issparse(transformed):
        assert np.isfinite(transformed.data).all()
    else:
        assert np.isfinite(transformed).all()

    assert target.isna().sum() == 0
    assert target.dtype == np.int8


def load_and_prepare_data(
    csv_path: str,
) -> tuple[pd.DataFrame, pd.Series]:
    """Load data, separate target, and remove unlabeled records."""
    raw = pd.read_csv(csv_path)

    required_columns = {
        "customer_id",
        "signup_date",
        "last_login",
        "age",
        "gender",
        "plan",
        "monthly_fee",
        "region",
        "churned",
    }

    missing_columns = required_columns.difference(raw.columns)
    if missing_columns:
        raise ValueError(
            f"Missing required columns: {sorted(missing_columns)}"
        )

    target = pd.to_numeric(raw["churned"], errors="coerce")
    labeled_mask = target.notna()

    X = raw.loc[labeled_mask].drop(columns=["churned"])
    y = validate_raw_target(target.loc[labeled_mask])

    return X, y


def main() -> None:
    """Example batch-training entry point."""
    X, y = load_and_prepare_data("customers.csv")

    pipeline = build_preprocessing_pipeline()

    # Fit only on X and y; target is never passed into the cleaner.
    pipeline.fit(X, y)

    transformed = pipeline.transform(X)
    validate_transformed_output(transformed, y)

    print(f"Rows processed: {transformed.shape[0]}")
    print(f"Features generated: {transformed.shape[1]}")
    print("Validation passed: no null or non-finite transformed values.")


if __name__ == "__main__":
    main()

4. 사후 검증 결과 및 운영 권고

현재 샘플 기준 예상 처리 결과는 다음과 같습니다.

• C004: churned 결측이므로 학습 데이터에서 제외
• C006: age=-5를 결측으로 변환 후 중앙값 대체
• C004: age, monthly_fee, last_login 결측을 파이프라인에서 처리
• M과 Male은 M으로 통합
• Seoul, seoul, Seoul 은 Seoul로 통합
• 최종 변환 결과에는 결측값과 무한값이 없어야 함
• 수치형 출력은 RobustScaler를 통해 이상치 영향이 완화됨
• One-Hot 결과는 float32 sparse matrix를 사용하여 메모리를 절약함

실제 운영 전에는 최소한 다음을 추가로 확인해야 합니다.

1. last_login이 예측 기준일 이후의 정보인지 여부
2. 요금제별 결측 대체가 필요한지 여부
3. 연령의 허용 범위가 실제 업무 규칙과 일치하는지 여부
4. 시간 기반 train/validation 분할
5. 결측률과 범주 분포의 일별 모니터링
6. 신규 지역·플랜 발생 시 handle_unknown="ignore" 동작 확인

현재 데이터는 7건뿐이므로 모델 성능 평가나 통계적 결론을 내리기에는 부족합니다. 이 데이터셋은 우선 데이터 품질 규칙과 파이프라인 검증용 샘플로 사용하는 것이 적절합니다.