Python案例如何用Pandas做数据抽样采样

我来介绍用Pandas进行数据抽样的几种常用方法。

基本抽样方法

随机抽样 (random sample)

import pandas as pd
import numpy as np
# 创建示例数据
df = pd.DataFrame({
    '姓名': ['张三', '李四', '王五', '赵六', '孙七', '周八', '吴九', '郑十'],
    '年龄': [25, 30, 35, 28, 32, 27, 29, 31],
    '城市': ['北京', '上海', '广州', '深圳', '北京', '上海', '广州', '深圳'],
    '收入': [10000, 15000, 12000, 18000, 11000, 16000, 13000, 14000]
})
print("原始数据:")
print(df)
print()
# 1. 抽样指定数量的行
sample_3 = df.sample(n=3)
print("随机抽取3行:")
print(sample_3)
print()
# 2. 抽样指定比例的行
sample_50pct = df.sample(frac=0.5)  # 抽取50%的数据
print("随机抽取50%的行:")
print(sample_50pct)
print()
# 设置随机种子，保证结果可重现
sample_with_seed = df.sample(n=3, random_state=42)
print("设置随机种子(可重复):")
print(sample_with_seed)

按权重抽样

# 创建权重进行抽样
weights = [0.1, 0.2, 0.3, 0.4]  # 权重之和不必为1
weighted_sample = df.sample(n=3, weights=weights)
print("按权重抽样:")
print(weighted_sample)
print()
# 根据某列的值作为权重
weighted_by_age = df.sample(n=3, weights='年龄')
print("按年龄权重抽样:")
print(weighted_by_age)

分层抽样

# 按城市分层抽样
def stratified_sample(df, strata_col, sample_size, random_state=None):
    """
    分层抽样函数
    """
    strata = df[strata_col].unique()
    sample_per_stratum = sample_size // len(strata)
    samples = []
    for stratum in strata:
        stratum_data = df[df[strata_col] == stratum]
        if len(stratum_data) >= sample_per_stratum:
            sample = stratum_data.sample(n=sample_per_stratum, 
                                        random_state=random_state)
        else:
            sample = stratum_data  # 如果数据不足，取全部
        samples.append(sample)
    return pd.concat(samples)
# 执行分层抽样
stratified_result = stratified_sample(df, '城市', 4, random_state=42)
print("分层抽样结果(每层2个样本):")
print(stratified_result)
print()
# 使用sklearn的分层抽样
from sklearn.model_selection import train_test_split
# 按城市分层，抽取20%的数据
stratified_train, stratified_test = train_test_split(
    df, test_size=0.5, stratify=df['城市'], random_state=42
)
print("sklearn分层抽样(测试集):")
print(stratified_test)

系统抽样

def systematic_sample(df, step):
    """
    系统抽样函数
    """
    indices = np.arange(0, len(df), step)
    return df.iloc[indices]
# 每隔2行抽取一个样本
systematic_result = systematic_sample(df, 2)
print("系统抽样(每隔2行):")
print(systematic_result)
print()
# 随机起点系统抽样
def random_start_systematic_sample(df, step, random_state=None):
    np.random.seed(random_state)
    start = np.random.randint(0, step)
    indices = np.arange(start, len(df), step)
    return df.iloc[indices]
systematic_random = random_start_systematic_sample(df, 2, random_state=42)
print("随机起点系统抽样:")
print(systematic_random)

集群抽样

# 示例：按城市进行集群抽样
def cluster_sample(df, cluster_col, n_clusters, random_state=None):
    """
    集群抽样函数
    """
    clusters = df[cluster_col].unique()
    selected_clusters = np.random.choice(clusters, 
                                        size=min(n_clusters, len(clusters)), 
                                        replace=False)
    return df[df[cluster_col].isin(selected_clusters)]
# 抽取2个城市的全部数据
cluster_result = cluster_sample(df, '城市', 2, random_state=42)
print("集群抽样(抽取2个城市):")
print(cluster_result)

高级抽样示例

# 1. 带放回抽样 (Bootstrap)
bootstrap_sample = df.sample(n=10, replace=True)  # 允许重复抽取
print("Bootstrap抽样(10个样本，可重复):")
print(bootstrap_sample)
print()
# 2. 条件抽样
def conditional_sample(df, condition, n=None, frac=None):
    """
    条件抽样：先筛选再抽样
    """
    filtered = df[condition]
    return filtered.sample(n=n, frac=frac) if len(filtered) > 0 else pd.DataFrame()
# 从年龄大于28的数据中抽取50%
condition = df['年龄'] > 28
conditional_result = conditional_sample(df, condition, frac=0.5)
print("条件抽样(年龄>28，抽取50%):")
print(conditional_result)
print()
# 3. 按组等量抽样
def equal_sample_per_group(df, group_col, n_per_group, random_state=None):
    """
    每组抽取相同数量的样本
    """
    return df.groupby(group_col).apply(
        lambda x: x.sample(n=min(n_per_group, len(x)), 
                          random_state=random_state)
    ).reset_index(drop=True)
equal_result = equal_sample_per_group(df, '城市', 1, random_state=42)
print("每组抽取1个样本:")
print(equal_result)

实用案例分析

# 创建更大的数据集用于演示
np.random.seed(42)
large_df = pd.DataFrame({
    'ID': range(1, 1001),
    '年龄': np.random.randint(18, 65, 1000),
    '性别': np.random.choice(['男', '女'], 1000),
    '城市': np.random.choice(['北京', '上海', '广州', '深圳'], 1000),
    '收入': np.random.normal(15000, 5000, 1000)
})
print("大数据集前5行:")
print(large_df.head())
print()
# 案例1：质量控制抽样（按比例）
def quality_control_sample(df, confidence_level=0.95, margin_error=0.05):
    """
    简单质量控制抽样计算
    """
    from scipy import stats
    z_score = stats.norm.ppf(1 - (1 - confidence_level) / 2)
    p = 0.5  # 最大变异情况
    sample_size = int(np.ceil((z_score**2 * p * (1 - p)) / margin_error**2))
    return df.sample(n=min(sample_size, len(df)))
qc_sample = quality_control_sample(large_df)
print(f"质量控制抽样样本数: {len(qc_sample)}")
print(qc_sample.head())
print()
# 案例2：按比例分层抽样
def proportional_stratified_sample(df, strata_col, sample_frac):
    """
    按比例的分层抽样（每层抽取相同比例）
    """
    return df.groupby(strata_col, group_keys=False).apply(
        lambda x: x.sample(frac=sample_frac)
    )
# 按性别分层，抽取10%
prop_stratified = proportional_stratified_sample(large_df, '性别', 0.1)
print(f"按比例分层抽样样本数: {len(prop_stratified)}")
print(prop_stratified.head())
print()
# 案例3：时间序列采样（如果有时间数据）
dates = pd.date_range('2024-01-01', periods=365, freq='D')
time_series = pd.DataFrame({
    '日期': dates,
    '销售额': np.random.normal(1000, 200, 365)
})
# 周采样（每周取一天）
weekly_sample = time_series.iloc[::7, :]
print("周采样:")
print(weekly_sample.head(10))

# 主要抽样函数
df.sample(n=数量)           # 按数量抽样
df.sample(frac=比例)         # 按比例抽样
df.sample(weights=权重)      # 按权重抽样
df.sample(replace=True)      # 有放回抽样
df.sample(random_state=42)   # 设置随机种子
# 分组抽样
df.groupby('列名').apply(lambda x: x.sample(n=数量))

这些方法涵盖了数据分析中大部分抽样场景,可以根据具体需求选择合适的方法。