Python案例如何用Pandas做数据分组重平衡

wen python案例 2026-07-14 200

本文目录导读：

下采样（Undersampling）—— 减少多数类样本
上采样（Oversampling）—— 增加少数类样本
SMOTE方法（需要安装imbalanced-learn）
自定义重平衡权重
分层采样（保持比例）
实用案例：信用卡欺诈检测数据重平衡
完整的工作流程
选择建议

我来介绍几种使用Pandas进行数据分组重平衡的方法：

下采样（Undersampling）—— 减少多数类样本

import pandas as pd
import numpy as np
# 创建示例数据
np.random.seed(42)
n_samples = 1000
# 模拟不平衡数据
data = pd.DataFrame({
    'feature1': np.random.randn(n_samples),
    'feature2': np.random.randn(n_samples),
    'target': np.random.choice(['A', 'B', 'C'], 
                               size=n_samples, 
                               p=[0.7, 0.2, 0.1])  # 类别不平衡
})
print("原始数据分布：")
print(data['target'].value_counts())
# 下采样：使所有类别样本数等于最少类别
min_class_size = data['target'].value_counts().min()
def undersample_group(group):
    return group.sample(n=min_class_size, random_state=42)
balanced_data = data.groupby('target', group_keys=False).apply(undersample_group)
print("\n下采样后数据分布：")
print(balanced_data['target'].value_counts())

上采样（Oversampling）—— 增加少数类样本

# 上采样：使所有类别样本数等于最多类别
max_class_size = data['target'].value_counts().max()
def oversample_group(group):
    return group.sample(n=max_class_size, replace=True, random_state=42)
balanced_data = data.groupby('target', group_keys=False).apply(oversample_group)
print("\n上采样后数据分布：")
print(balanced_data['target'].value_counts())

SMOTE方法（需要安装imbalanced-learn）

# 安装：pip install imbalanced-learn
from imblearn.over_sampling import SMOTE
from sklearn.preprocessing import LabelEncoder
# 准备特征和标签
X = data[['feature1', 'feature2']]
y = data['target']
# 使用SMOTE进行过采样
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X, y)
print("\nSMOTE重采样后数据分布：")
print(pd.Series(y_resampled).value_counts())

自定义重平衡权重

# 计算每个类别的权重
class_weights = 1 / data['target'].value_counts()
sample_weights = data['target'].map(class_weights)
print("\n样本权重（前5个）：")
print(sample_weights.head())
# 带权重的采样
weighted_data = data.sample(n=len(data), 
                           weights=sample_weights, 
                           replace=True, 
                           random_state=42)
print("\n带权重采样后数据分布：")
print(weighted_data['target'].value_counts())

分层采样（保持比例）

def stratified_sample(df, stratify_col, sample_size):
    """
    分层采样：从各组中抽取指定比例或数量的样本
    """
    def sample_group(group, size):
        if size < len(group):
            return group.sample(n=size, random_state=42)
        return group
    # 计算每组应抽取的样本数
    group_sizes = df[stratify_col].value_counts()
    n_per_group = int(sample_size / len(group_sizes))
    return df.groupby(stratify_col, group_keys=False).apply(
        lambda x: sample_group(x, min(n_per_group, len(x)))
    )
# 从每个类别抽取相同数量的样本
stratified_data = stratified_sample(data, 'target', 100)
print("\n分层采样后数据分布：")
print(stratified_data['target'].value_counts())

实用案例：信用卡欺诈检测数据重平衡

# 模拟信用卡交易数据
np.random.seed(42)
n_transactions = 10000
fraud_data = pd.DataFrame({
    'amount': np.random.exponential(100, n_transactions),
    'time': np.random.randint(0, 24, n_transactions),
    'is_fraud': np.random.choice([0, 1], 
                                 size=n_transactions,
                                 p=[0.99, 0.01])  # 1%的欺诈交易
})
print("原始欺诈数据分布：")
print(fraud_data['is_fraud'].value_counts(normalize=True))
# 重平衡策略
def balance_fraud_data(df, target_col, ratio=0.5):
    """
    重平衡欺诈检测数据
    参数:
    - df: 数据框
    - target_col: 目标列名
    - ratio: 少数类比例（0-1之间）
    """
    # 分离多数类和少数类
    majority = df[df[target_col] == 0]
    minority = df[df[target_col] == 1]
    # 计算需要的样本数
    n_minority = len(minority)
    n_majority_needed = int(n_minority * (1 - ratio) / ratio)
    # 对多数类进行下采样
    if n_majority_needed < len(majority):
        majority_sampled = majority.sample(n=n_majority_needed, random_state=42)
    else:
        majority_sampled = majority.sample(n=n_majority_needed, 
                                          replace=True, 
                                          random_state=42)
    # 合并数据
    balanced = pd.concat([majority_sampled, minority])
    return balanced.sample(frac=1, random_state=42)  # 打乱顺序
# 应用重平衡
balanced_fraud = balance_fraud_data(fraud_data, 'is_fraud', ratio=0.5)
print("\n重平衡后欺诈数据分布：")
print(balanced_fraud['is_fraud'].value_counts())
print(f"\n重平衡前样本数: {len(fraud_data)}")
print(f"重平衡后样本数: {len(balanced_fraud)}")

完整的工作流程

class DataBalancer:
    """
    数据重平衡器
    """
    def __init__(self, method='undersample'):
        self.method = method
        self.balanced_data = None
    def balance(self, data, target_col, random_state=42):
        """
        执行数据重平衡
        """
        if self.method == 'undersample':
            min_size = data[target_col].value_counts().min()
            self.balanced_data = data.groupby(target_col, group_keys=False).apply(
                lambda x: x.sample(n=min_size, random_state=random_state)
            )
        elif self.method == 'oversample':
            max_size = data[target_col].value_counts().max()
            self.balanced_data = data.groupby(target_col, group_keys=False).apply(
                lambda x: x.sample(n=max_size, replace=True, random_state=random_state)
            )
        return self.balanced_data
    def get_balance_ratio(self):
        """检查重平衡后的比例"""
        return self.balanced_data['target'].value_counts(normalize=True)
# 使用示例
balancer = DataBalancer(method='oversample')
balanced = balancer.balance(data, 'target')
print("\n重平衡比例：")
print(balancer.get_balance_ratio())