在 Pandas groupby().agg() 中实现带权平均计算的进阶指南

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

在 Pandas groupby().agg() 中实现带权平均计算的进阶指南

2025-11-06

浏览次数：次

返回列表

在 Pandas groupby().agg() 中实现带权平均计算的进阶指南

本教程旨在解决在 pandas `groupby().agg()` 操作中，当自定义聚合函数需要访问原始 dataframe 中的其他列（例如进行加权平均）时遇到的 `nameerror` 问题。文章将详细阐述 `groupby` 的工作机制，并提供一种利用 python 闭包（closure）的优雅解决方案，确保自定义函数能够正确获取并使用所需的上下文数据，从而实现复杂的数据聚合逻辑。

理解 Pandas groupby().agg() 的工作原理与常见挑战

在数据分析中，我们经常需要对 DataFrame 进行分组聚合操作。Pandas 提供了强大的 groupby().agg() 方法来实现这一目标。然而，当聚合逻辑变得复杂，特别是自定义聚合函数需要访问分组 Series 之外的原始 DataFrame 的其他列时，会遇到作用域（scope）问题，导致 NameError。

考虑以下场景：我们需要计算 other_col 列的加权平均值，权重由 amount 列提供。如果直接在 agg 中调用一个外部函数 weighted_mean，并尝试在该函数内部引用原始 DataFrame df1，Pandas 会抛出 NameError，因为 weighted_mean 函数在执行时无法直接访问到 df1 这个变量。groupby().agg() 机制会将分组后的 Series 传递给自定义函数，而不是整个 DataFrame。

原始问题代码示例：

import pandas as pd
import numpy as np

def weighted_mean(x):
    # 这里会发生 NameError，因为 df1 在此作用域中未定义
    try: 
        return np.*erage(x, weights=df1.loc[x.index, 'amount']) > 0.5
    except ZeroDivisionError:
        return 0

def some_function(df1=None):
    df1 = df1.groupby('id').agg(xx=('amount', lambda x: x.sum() > 100),
                                yy=('other_col', weighted_mean)).reset_index()
    return df1

df2 = pd.DataFrame({'id':[1,1,2,2,3], 'amount':[10, 200, 1, 10, 150], 'other_col':[0.1, 0.6, 0.7, 0.2, 0.4]})
# 调用 some_function 会导致 NameError
# df2 = some_function(df1=df2)

上述代码中，当 groupby().agg() 调用 weighted_mean 函数时，x 将是 other_col 列的一个 Series 子集。然而，weighted_mean 内部尝试通过 df1.loc[x.index, 'amount'] 来获取权重，此时 df1 并不在 weighted_mean 的局部或全局作用域中，从而引发 NameError。

解决方案：利用 Python 闭包（Closure）传递上下文

为了解决这个问题，我们可以利用 Python 的闭包特性。闭包是指一个函数记住其被创建时的环境，即使该环境（作用域）已经不存在，它仍然可以访问该环境中的变量。在这里，我们可以创建一个外部函数，它接收整个 DataFrame 作为参数，然后返回一个内部函数。这个内部函数就是我们实际用于聚合的函数，它能够“捕获”并访问外部函数传入的 DataFrame。

闭包的工作原理：

刺鸟创客

一款专业高效稳定的AI内容创作平台

110 查看详情刺鸟创客

外部函数（Outer Function）：接收原始 DataFrame 作为参数。
内部函数（Inner Function）：这是实际用于 agg 的函数。它定义在外部函数内部，因此可以访问外部函数参数中的 DataFrame。
返回内部函数：外部函数返回这个内部函数。当外部函数执行完毕后，内部函数仍然保留了对外部函数作用域中 DataFrame 的引用。

实现步骤与代码示例

我们将 weighted_mean 函数重构为一个接受 DataFrame 参数的外部函数，它返回一个内部函数 inner_weighted_mean。

定义外部 weighted_mean 函数： 这个函数现在接收 df1 作为参数。它的作用是创建一个并返回一个专门用于聚合的内部函数。

def weighted_mean_factory(df_context): # 外部函数接收整个 DataFrame
    def inner_weighted_mean(x): # 内部函数，用于 agg
        try: 
            # inner_weighted_mean 闭包捕获了 df_context
            return np.*erage(x, weights=df_context.loc[x.index, 'amount']) > 0.5
        except ZeroDivisionError:
            return 0
    return inner_weighted_mean # 返回内部函数

修改 some_function 以使用闭包： 在 some_function 内部，我们首先调用 weighted_mean_factory 并传入当前的 df1。这将返回一个“预配置”的 inner_weighted_mean 函数，该函数已经知道如何访问 df1。然后，我们将这个返回的函数传递给 agg 方法。

def some_function(df1=None):
    # 创建一个针对当前 df1 的闭包实例
    weighted_mean_for_df1 = weighted_mean_factory(df1) 

    df1 = df1.groupby('id').agg(
        xx=('amount', lambda x: x.sum() > 100),
        yy=('other_col', weighted_mean_for_df1) # 使用闭包返回的函数
    ).reset_index()
    return df1

完整代码示例：

import pandas as pd
import numpy as np

# 1. 定义闭包工厂函数
def weighted_mean_factory(df_context):
    """
    创建一个闭包，用于计算加权平均。
    df_context: 原始 DataFrame，提供权重列。
    """
    def inner_weighted_mean(x):
        """
        实际用于 agg 的函数，计算 Series x 的加权平均。
        权重从 df_context 中获取。
        """
        try: 
            # 使用闭包捕获的 df_context 来获取权重
            weights = df_context.loc[x.index, 'amount']
            # 避免空 Series 或全零权重导致 numpy 警告或错误
            if weights.sum() == 0 and not x.empty:
                return 0 # 或者其他逻辑，例如返回 False
            return np.*erage(x, weights=weights) > 0.5
        except ZeroDivisionError:
            # 当权重总和为零时，np.*erage 可能抛出 ZeroDivisionError
            return 0
    return inner_weighted_mean

# 2. 修改主函数以使用闭包工厂
def some_function(df_input=None):
    """
    对 DataFrame 进行分组聚合，其中包含一个使用闭包的加权平均。
    df_input: 输入的 Pandas DataFrame。
    """
    if df_input is None:
        raise ValueError("Input DataFrame cannot be None.")

    # 在 groupby 之前创建闭包实例，确保它能访问原始 df_input
    weighted_mean_aggregator = weighted_mean_factory(df_input)

    # 执行分组聚合
    result_df = df_input.groupby('id').agg(
        xx=('amount', lambda x: x.sum() > 100),
        yy=('other_col', weighted_mean_aggregator) # 将闭包返回的函数传递给 agg
    ).reset_index()
    return result_df

# 3. 示例数据与执行
df_original = pd.DataFrame({
    'id': [1, 1, 2, 2, 3], 
    'amount': [10, 200, 1, 10, 150], 
    'other_col': [0.1, 0.6, 0.7, 0.2, 0.4]
})

df_processed = some_function(df_input=df_original)
print(df_processed)

输出结果：

   id     xx     yy
0   1   True   True
1   2  False  False
2   3   True  False

注意事项与总结

闭包的优势： 闭包提供了一种优雅且 Pythonic 的方式，允许内部函数在需要时访问其定义时的外部环境。这在处理 Pandas groupby().agg() 等需要上下文信息的场景中非常有用，避免了使用全局变量或传递额外参数的复杂性。
性能考量： 对于非常大的 DataFrame，df_context.loc[x.index, 'amount'] 在每次分组聚合时都会执行索引查找。虽然 Pandas 的 loc 查找效率很高，但如果性能是关键因素，可以考虑在 groupby 之前预先计算好权重，或者在某些特定情况下，使用 df.apply() 可能更直接（但通常 agg 配合闭包性能更优）。
错误处理： 在计算加权平均时，需要考虑权重总和为零的情况，这可能导致 ZeroDivisionError。在 inner_weighted_mean 函数中添加 try-except 块是一个良好的实践。此外，如果 weights Series 为空，np.*erage 也可能报错，因此在实际应用中可能需要更全面的空值和零值检查。
清晰性与可维护性： 使用闭包模式可以使代码结构更清晰，将获取上下文的逻辑与实际的聚合逻辑分离，提高了代码的可读性和可维护性。