Python 珍藏函数超详解：随机抽样，分层抽样，系统抽样方法汇总

71 阅读 0 评论 47 点赞

我是靠谱客的博主踏实毛衣，最近开发中收集的这篇文章主要介绍Python 珍藏函数超详解：随机抽样，分层抽样，系统抽样方法汇总，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

引言

抽样调查在统计学与 Python数据分析/数据挖掘/数据科学中非常常用，在实际业务中也是高频刚需，而 Python 并没有专有的抽样方法库，所以将自己以前的笔记汇总到自写库中，用到时直接调用函数即可，快速且精确。

进行统计学分析，假设检验，方差分析，单因素xx，t 检验，xx 检验的时候，样本量数据如果过大，则的出来的统计检验力如 p 值等就不可信。即假如我们的整体数据有 10 万，进行假设检验的时候只需要根据数据分布情况分层抽样一小部分就行了，所以分层抽样用得也是最多的，但很可惜 Python 并没有这样的库，只能自己写，以往的学习中我也不记得这个库是从哪里找到的了，不过一直都非常非常好用，效率 10 倍 max。

行文思路

全部源代码(根据填入参数选择不同的抽样方法) --》对每一方法进行单独使用并附上效果图

以某电信客户数据为例

数据预览

在这里插入图片描述

随机抽样

在这里插入图片描述

两种方法实现分层抽样

方法一：每层相同的数量
在这里插入图片描述
方法二：每层相同的百分比 20%

系统抽样

系统抽样用得太少，我不会

全部源代码（含注释）

# ----------------------  抽样函数  --------------------------
def get_sample(df, sampling="simple_random", k=1, stratified_col=None):
    """
    对输入的 dataframe 进行抽样的函数

    参数:
        - df: 输入的数据框 pandas.dataframe 对象

        - sampling:抽样方法 str
            可选值有 ["simple_random", "stratified", "systematic"]
            按顺序分别为: 简单随机抽样、分层抽样、系统抽样

        - k: 抽样个数或抽样比例 int or float
            (int, 则必须大于0; float, 则必须在区间(0,1)中)
            如果 0 < k < 1 , 则 k 表示抽样对于总体的比例
            如果 k >= 1 , 则 k 表示抽样的个数；当为分层抽样时，代表每层的样本量

        - stratified_col: 需要分层的列名的列表 list
            只有在分层抽样时才生效

    返回值:
        pandas.dataframe 对象, 抽样结果
    """
    import random
    import pandas as pd
    from functools import reduce
    import numpy as np
    import math
    
    len_df = len(df)
    if k <= 0:
        raise AssertionError("k不能为负数")
    elif k >= 1:
        assert isinstance(k, int), "选择抽样个数时, k必须为正整数"
        sample_by_n=True
        if sampling is "stratified":
            alln=k*df.groupby(by=stratified_col)[stratified_col[0]].count().count() # 有问题的
            #alln=k*df[stratified_col].value_counts().count() 
            if alln >= len_df:
                raise AssertionError("请确认k乘以层数不能超过总样本量")
    else:
        sample_by_n=False
        if sampling in ("simple_random", "systematic"):
            k = math.ceil(len_df * k)
        
    #print(k)

    if sampling is "simple_random":
        print("使用简单随机抽样")
        idx = random.sample(range(len_df), k)
        res_df = df.iloc[idx,:].copy()
        return res_df

    elif sampling is "systematic":
        print("使用系统抽样")
        step = len_df // k+1          #step=len_df//k-1
        start = 0                  #start=0
        idx = range(len_df)[start::step]  #idx=range(len_df+1)[start::step]
        res_df = df.iloc[idx,:].copy()
        #print("k=%d,step=%d,idx=%d"%(k,step,len(idx)))
        return res_df

    elif sampling is "stratified":
        # assert 设置断言
        assert stratified_col is not None, "请传入包含需要分层的列名的列表"
        assert all(np.in1d(stratified_col, df.columns)), "请检查输入的列名"
        
        grouped = df.groupby(by=stratified_col)[stratified_col[0]].count()
        if sample_by_n==True:
            group_k = grouped.map(lambda x:k)
        else:
            group_k = grouped.map(lambda x: math.ceil(x * k))
        
        res_df = pd.DataFrame(columns=df.columns)
        for df_idx in group_k.index:
            df1=df
            if len(stratified_col)==1:
                df1=df1[df1[stratified_col[0]]==df_idx]
            else:
                for i in range(len(df_idx)):
                    df1=df1[df1[stratified_col[i]]==df_idx[i]]
            idx = random.sample(range(len(df1)), group_k[df_idx])
            group_df = df1.iloc[idx,:].copy()
            res_df = res_df.append(group_df)
        return res_df

    else:
        raise AssertionError("sampling is illegal")