WOE và IV: Nghệ thuật 'gạn đục khơi trong' cho dữ liệu tín dụng

import numpy as np
import pandas as pd
import scorecardpy as sc

# --- Cách 1: scorecardpy — chuẩn production ---
# Tự động binning + enforce monotonicity
bins = sc.woebin(
    df, y='bad_flag',
    x=['monthly_income', 'dpd_max_6m', 'job_tenure_months'],
    bin_num_limit=6,
    # monotonic_binning=True  # bật nếu muốn enforce
)

# Tóm tắt IV của tất cả biến
iv_table = pd.DataFrame({
    col: {'IV': bins[col]['total_iv'].iloc[0]}
    for col in bins
}).T.sort_values('IV', ascending=False)
print(iv_table)
# monthly_income    0.391
# dpd_max_6m        0.312
# job_tenure_months 0.087

# Transform sang WOE values để đưa vào Logistic Regression
train_woe = sc.woebin_ply(train_df, bins)
# ⚠️  QUAN TRỌNG: lock bảng bins này khi deploy
# Không recalculate lại trên data mới — WOE map phải được freeze

# --- Cách 2: optbinning — mạnh hơn, hỗ trợ nhiều constraint ---
from optbinning import BinningProcess

binning_process = BinningProcess(
    variable_names=['monthly_income', 'dpd_max_6m'],
    categorical_variables=[],
    max_n_bins=6,
    min_bin_size=0.05,   # mỗi bin tối thiểu 5% population
    monotonic_trend='auto'
)
binning_process.fit(X_train, y_train)
print(binning_process.summary()[['name', 'iv', 'js']].sort_values('iv', ascending=False))

import pandas as pd
import numpy as np

def woe_iv(df: pd.DataFrame, feature: str, target: str,
           bins: int = 10, min_bin_pct: float = 0.05) -> tuple[pd.DataFrame, float]:
    """
    Compute WOE and IV for a numeric feature.
    target: 1 = Bad (default), 0 = Good (non-default)
    """
    df = df[[feature, target]].copy()

    # Handle missing values as a separate bin
    missing_mask = df[feature].isna()
    df_non_missing = df[~missing_mask].copy()
    df_missing = df[missing_mask].copy()

    # Bin non-missing values using quantile cuts
    df_non_missing['bin'] = pd.qcut(
        df_non_missing[feature], q=bins, duplicates='drop'
    )

    def compute_stats(subset):
        agg = subset.groupby('bin', observed=True)[target].agg(['sum', 'count'])
        agg.columns = ['bad', 'total']
        agg['good'] = agg['total'] - agg['bad']
        return agg

    stats = compute_stats(df_non_missing)

    # Add missing bin if it exists
    if len(df_missing) > 0:
        missing_stats = pd.DataFrame({
            'bad': [df_missing[target].sum()],
            'total': [len(df_missing)],
            'good': [len(df_missing) - df_missing[target].sum()],
        }, index=['MISSING'])
        stats = pd.concat([stats, missing_stats])

    total_good = stats['good'].sum()
    total_bad = stats['bad'].sum()

    stats['pct_good'] = stats['good'] / total_good
    stats['pct_bad']  = stats['bad']  / total_bad

    # Avoid log(0)
    stats['pct_good'] = stats['pct_good'].clip(lower=1e-4)
    stats['pct_bad']  = stats['pct_bad'].clip(lower=1e-4)

    stats['woe']    = np.log(stats['pct_good'] / stats['pct_bad'])
    stats['iv_bin'] = (stats['pct_good'] - stats['pct_bad']) * stats['woe']

    iv_total = stats['iv_bin'].sum()
    return stats, round(iv_total, 4)

# Usage
woe_table, iv = woe_iv(df, feature='monthly_income', target='is_default')
print(f"IV = {iv}")
print(woe_table[['bad', 'good', 'pct_bad', 'pct_good', 'woe', 'iv_bin']])

# ĐỪNG làm thế này
df['monthly_income'].fillna(df['monthly_income'].median(), inplace=True)

# HÃY làm thế này
# Để WOE/IV tính separate bin cho missing
# Sau đó quyết định dựa trên WOE của MISSING bin

# Nếu WOE(MISSING) = -1.5 → khách hàng thiếu income data = high risk
# → Missing mang thông tin quan trọng, không được fill trước khi WOE

# Nếu WOE(MISSING) ≈ 0 → missing ngẫu nhiên, có thể fill sau

def woe_stability_check(dev_woe_table, prod_df, feature, target, bins=10):
    """Check if WOE table from dev period is still valid on current data."""
    # Apply dev bins to prod data
    prod_woe_table, _ = woe_iv(prod_df, feature, target, bins)

    # Compare WOE values per bin
    comparison = dev_woe_table[['woe']].rename(columns={'woe': 'woe_dev'}).join(
        prod_woe_table[['woe']].rename(columns={'woe': 'woe_prod'}),
        how='outer'
    )
    comparison['woe_shift'] = abs(comparison['woe_dev'] - comparison['woe_prod'])
    return comparison

# Kiểm tra leakage đơn giản: correlation với target
suspicious = df.corrwith(df['is_default']).abs()
high_corr = suspicious[suspicious > 0.5]
print("Suspicious high-correlation features:")
print(high_corr.sort_values(ascending=False))