Pandas基于阈值的条件式数据分组策略

在数据分析实践中，我们经常需要对数据集进行多维度的分组聚合。然而，有时我们希望对那些规模过小的分组停止进一步的细化，将其作为一个整体进行统计，以避免过度分散的数据或保护隐私。例如，在一个包含多个层级（如省份、城市、区县）的数据集中，如果某个城市的数据量低于特定阈值，我们可能就不再关心该城市内部的区县分布，而是将其所有数据合并到城市层面进行报告。

问题场景与挑战

假设我们有一个包含多列（如a, b, c）的DataFrame，我们希望按照这些列的顺序进行分组。核心需求是：对于任意一个分组层级，如果当前分组的行数小于预设的阈值，则停止对该分组进行后续列的细分，将其作为一个最终的聚合单元。而对于那些行数超过阈值的分组，则继续按照下一列进行细分。

考虑以下示例数据：

import pandas as pd
import numpy as np

df = pd.DataFrame({'a':[1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                 'b':[1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                 'c':[1, 1, 1, 2, 2, 2, 3, 4, 4, 2, 2, 2, 2, 2, 2, 2, 2, 2]
                })

我们期望的输出结果，假设阈值为3：

   a  b    c  count
0  1  1  1.0      3  # (1,1,1) 组大小为3，达到阈值，不再细分
1  1  1  2.0      3  # (1,1,2) 组大小为3，达到阈值，不再细分
2  2  2  2.0      9  # (2,2,2) 组大小为9，达到阈值，不再细分
0  1  2  NaN      3  # (1,2,3)大小1，(1,2,4)大小2，总计(1,2)组大小3，达到阈值，c列聚合为NaN

传统的groupby()方法难以直接实现这种“条件式停止”的逻辑，因为它通常会一次性按照所有指定列进行分组，或者需要复杂的迭代和合并操作。

硅基智能

基于Web3.0的元宇宙，去中心化的互联网，高质量、沉浸式元宇宙直播平台，用数字化重新定义直播

62

查看详情

解决方案：迭代聚合与条件筛选

解决此问题的一种高效方法是利用Pandas的value_counts()函数结合迭代的groupby(level=...)操作。核心思想是从最细粒度的分组开始，逐步向上聚合，并在每个聚合层级检查分组大小，将符合阈值条件的组“锁定”并收集起来，而将不符合条件的组继续向上聚合。

算法步骤

1. 获取所有最细粒度分组的计数： 使用df.value_counts()一次性计算所有列组合的频次。这比先groupby().size()更高效。
2. 初始化： 定义一个空列表用于存储最终结果，并获取所有参与分组的列名列表。
3. 迭代聚合： 从最详细的列组合开始，逐步减少分组的列数（即从右向左移除列）。
   • 在每次迭代中，将当前待处理的Series（包含各种组合的计数）按照当前剩余的列进行groupby(level=cols).sum()，从而实现向上聚合。
   • 识别出聚合后计数达到或超过阈值的组。这些组是“最终”的，将其添加到结果列表中。
   • 识别出聚合后计数仍低于阈值的组。这些组需要进一步向上聚合，因此将其保留到下一次迭代。
   • 从列名列表中移除最右边的列，准备进行更粗粒度的聚合。
4. 收集剩余组： 循环结束后，如果仍有任何组未能达到阈值（即它们在最粗粒度分组时也未达到阈值），将其添加到结果列表中。
5. 整合结果： 将所有收集到的Series（每个都是一个Pandas Series，索引是多级索引）通过reset_index()转换为DataFrame，并使用pd.concat()合并成最终结果。

示例代码

import pandas as pd
import numpy as np

# 示例数据
df = pd.DataFrame({'a':[1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                 'b':[1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                 'c':[1, 1, 1, 2, 2, 2, 3, 4, 4, 2, 2, 2, 2, 2, 2, 2, 2, 2]
                })

# 定义阈值
threshold = 3

# 获取所有列名
cols = list(df.columns)

# 第一步：获取所有最细粒度分组的计数
# value_counts() 返回一个Series，索引是多级索引，值是计数
s = df.value_counts()

# 存储最终结果的列表
out = []

# 第二步：迭代聚合
# 循环条件：还有列可以用来分组，并且还有数据需要处理
while cols and len(s) > 0:
    # 向上聚合：根据当前剩余的列进行分组求和
    # 例如，如果cols是['a', 'b', 'c']，则按a,b,c聚合；如果cols是['a', 'b']，则按a,b聚合
    s_grouped = s.groupby(level=cols).sum()

    # 找出计数小于阈值的组（需要继续向上聚合）
    mask_below_threshold = s_grouped < threshold

    # 将计数达到或超过阈值的组添加到结果列表
    # `~mask_below_threshold` 表示计数 >= threshold 的组
    out.append(s_grouped[~mask_below_threshold])

    # 更新待处理的Series，只保留计数小于阈值的组
    s = s_grouped[mask_below_threshold]

    # 移除最右边的列，以便在下一次迭代中进行更粗粒度的分组
    cols.pop()

# 第三步：将最后剩余的组（即使未达到阈值，也作为最终结果）添加到结果列表
# 这通常发生在s为空或者cols为空，但s仍有数据时（即最粗粒度分组后仍有小于阈值的组）
if len(s) > 0:
    out.append(s)

# 第四步：整合所有结果
# 将每个Series重置索引，然后合并成一个DataFrame
final_result_df = pd.concat([x.reset_index() for x in out])

# 重新命名计数列，使其更具可读性
final_result_df = final_result_df.rename(columns={0: 'group_size'})

# 填充NaN值，使输出更符合预期（聚合时未使用的列会变为NaN）
# 对于数值列，Pandas会自动填充NaN，这里主要是为了明确展示
# 例如，如果原始分组是(a,b,c)，聚合到(a,b)时，c列会变成NaN
print(final_result_df)

代码解析

1. s = df.value_counts(): 这是整个过程的起点。它创建了一个Pandas Series，其索引是DataFrame中所有列的唯一组合（多级索引），值是这些组合出现的次数。这提供了最细粒度的原始计数。
2. while cols and len(s) > 0:: 循环条件确保我们有列可以用来分组，并且还有数据需要处理。当s变为空Series时，表示所有组都已处理完毕。
3. s_grouped = s.groupby(level=cols).sum(): 这是核心的聚合步骤。groupby(level=cols)会根据当前cols列表中的列名进行分组。由于s的索引是多级索引，level参数允许我们指定按哪些级别的索引进行分组。.sum()则将相同分组下的计数相加，从而实现向上聚合。
   • 例如，如果s中有(1,2,3):1和(1,2,4):2，而cols是['a','b']，则s.groupby(level=['a','b']).sum()会得到(1,2):3。
4. mask_below_threshold = s_grouped < threshold: 创建一个布尔掩码，标记出那些聚合后计数仍然小于阈值的组。
5. out.append(s_grouped[~mask_below_threshold]): 将计数大于或等于阈值的组（即不需要再细分的组）添加到out列表中。这些组已经“完成”了它们的聚合。
6. s = s_grouped[mask_below_threshold]: 更新s，使其只包含那些计数仍然小于阈值的组。这些组将在下一次迭代中进行更粗粒度的聚合。
7. cols.pop(): 移除cols列表中的最后一个元素。这使得下一次迭代的groupby(level=cols)操作会少一个分组维度，从而实现更粗粒度的聚合。
8. if len(s) > 0: out.append(s): 循环结束后，如果s中仍然有数据（即即使在最粗粒度分组时也未达到阈值的组），将其作为最终结果添加到out列表中。
9. final_result_df = pd.concat([x.reset_index() for x in out]): out列表中的每个元素都是一个Pandas Series。x.reset_index()将Series的索引转换为列，并默认将值列命名为0。pd.concat()将所有这些DataFrame合并成一个最终的DataFrame。
10. final_result_df = final_result_df.rename(columns={0: 'group_size'}): 将默认的计数列名0改为更具描述性的group_size。

运行结果

   a  b    c  group_size
0  1  1  1.0           3
1  1  1  2.0           3
2  2  2  2.0           9
0  1  2  NaN           3

这个输出与我们期望的结果完全一致。对于a=1, b=2的组，因为其原始细分(1,2,3)和(1,2,4)的计数都小于3，它们被向上聚合到(1,2)层面，总计数为3，达到了阈值，因此在c列显示为NaN，表示该层级已停止细分。

注意事项与总结

• 效率： 使用df.value_counts()作为起点比多次groupby().size()或groupby().agg('size')通常更高效，因为它一次性计算了所有唯一组合的频次。
• 列顺序： 解决方案依赖于cols.pop()来逐步减少分组维度，这意味着它会从列列表的末尾开始聚合。因此，cols列表的顺序决定了聚合的优先级（从最细粒度到最粗粒度）。如果你的分组顺序有特定要求，请确保cols列表的顺序正确。
• NaN值： 当数据向上聚合时，那些不再用于分组的列（即被pop()掉的列）在最终结果中会显示为NaN。这是预期的行为，表示这些维度已被聚合。
• 灵活性： 这种迭代聚合的方法非常灵活，可以适应各种复杂的条件分组需求。你可以在mask_below_threshold的判断逻辑中加入更复杂的条件。

通过这种迭代聚合的策略，我们能够优雅地处理Pandas中基于阈值的条件式分组问题，使得数据分析结果既能保留必要的细节，又能对规模较小的分组进行有效的汇总，从而提升数据报告的质量和洞察力。

以上就是Pandas基于阈值的条件式数据分组策略的详细内容，更多请关注php中文网其它相关文章！