Pandas高级GroupBy技术详解

news/2025/12/4 6:02:36/文章来源:https://www.cnblogs.com/codeshare1135/p/19304704

引言

虽然 groupby().sum() 和 groupby().mean() 用于快速检查没有问题，但生产级别的指标需要更健壮的解决方案。现实世界的表格通常涉及多个键、时间序列数据、权重以及各种条件，如促销、退货或异常值。

这意味着你经常需要计算总额和比率、在每个细分市场中对项目进行排名、按日历时段汇总数据，然后将分组统计数据合并回原始行以进行建模。本文将指导你使用Pandas库中的高级分组技术来有效处理这些复杂场景。

选择正确的模式

使用 agg 将组缩减为一行

当需要每组一条记录（例如总计、均值、中位数、最小值/最大值和自定义向量化聚合）时，使用 agg。

out = (df.groupby(['store', 'cat'], as_index=False, sort=False).agg(sales=('rev', 'sum'),orders=('order_id', 'nunique'),avg_price=('price', 'mean'))
)

这适用于关键绩效指标（KPI）表、每周汇总和多指标摘要。

使用 transform 将统计量广播回各行

transform 方法返回一个与输入形状相同的结果。它非常适合创建每行需要的特征，例如Z分数、组内份额或分组填充。

g = df.groupby('store')['rev']
df['rev_z'] = (df['rev'] - g.transform('mean')) / g.transform('std')
df['rev_share'] = df['rev'] / g.transform('sum')

这适用于建模特征、质量保证比率和插补。

使用 apply 实现自定义的组内逻辑

仅当所需逻辑无法用内置函数表达时，才使用 apply。它速度较慢且难以优化，因此应首先尝试使用 agg 或 transform。

def capped_mean(s):q1, q3 = s.quantile([.25, .75])return s.clip(q1, q3).mean()df.groupby('store')['rev'].apply(capped_mean)

这适用于定制规则和小型分组。

使用 filter 保留或丢弃整个组

filter 方法允许整个组通过或失败某个条件。这对于数据质量规则和阈值设定非常方便。

big = df.groupby('store').filter(lambda g: g['order_id'].nunique() >= 100)

这适用于最小规模队列，以及在聚合前移除稀疏类别。

多键分组和命名聚合

按多个键分组

你可以控制输出形状和顺序，以便结果可以直接放入商业智能工具。

g = df.groupby(['store', 'cat'], as_index=False, sort=False, observed=True)

as_index=False 返回一个扁平的DataFrame，更容易连接和导出
sort=False 避免重新排序分组，在顺序无关紧要时可以节省工作量
observed=True（与分类列一起使用）会丢弃未使用的类别组合

使用命名聚合

命名聚合会产生可读的、类似SQL的列名。

out = (df.groupby(['store', 'cat']).agg(sales=('rev', 'sum'),orders=('order_id', 'nunique'),    # 在这里使用你的ID列avg_price=('price', 'mean'))
)

整理列

如果堆叠多个聚合，你将得到一个多级索引（MultiIndex）。将其扁平化一次并标准化列的顺序。

out = out.reset_index()
out.columns = ['_'.join(c) if isinstance(c, tuple) else cfor c in out.columns
]
# 可选：确保业务友好的列顺序
cols = ['store', 'cat', 'orders', 'sales', 'avg_price']
out = out[cols]

无需 apply 的条件聚合

在 agg 中使用布尔掩码数学

当掩码依赖于其他列时，通过其索引对齐数据。

# 按 (store, cat) 的促销销售额和促销率
cond = df['is_promo']
out = df.groupby(['store', 'cat']).agg(promo_sales=('rev', lambda s: s[cond.loc[s.index]].sum()),promo_rate=('is_promo', 'mean')  # 促销行的比例
)

计算比率和比例

比率简单地等于 sum(mask) / size，这等价于布尔列的平均值。

df['is_return'] = df['status'].eq('returned')
rates = df.groupby('store').agg(return_rate=('is_return', 'mean'))

创建队列式窗口

首先，使用日期边界预计算掩码，然后聚合数据。

# 示例：每位客户队列首次购买后30天内的重复购买
first_ts = df.groupby('customer_id')['ts'].transform('min')
within_30 = (df['ts'] <= first_ts + pd.Timedelta('30D')) & (df['ts'] > first_ts)# 客户队列 = 首次购买的月份
df['cohort'] = first_ts.dt.to_period('M').astype(str)repeat_30_rate = (df.groupby('cohort').agg(repeat_30_rate=('within_30', 'mean')).rename_axis(None)
)

每组的加权指标

实现加权平均模式

向量化数学计算，并防范零权重的除法。

import numpy as nptmp = df.assign(wx=df['price'] * df['qty'])
agg = tmp.groupby(['store', 'cat']).agg(wx=('wx', 'sum'), w=('qty', 'sum'))# 按 (store, cat) 的加权平均价格
agg['wavg_price'] = np.where(agg['w'] > 0, agg['wx'] / agg['w'], np.nan)

安全地处理 NaN 值

决定为空组或全NaN值返回什么。两种常见选择是：

# 1) 返回 NaN（透明，对于下游统计最安全）
agg['wavg_price'] = np.where(agg['w'] > 0, agg['wx'] / agg['w'], np.nan)# 2) 如果所有权重为零，则回退到未加权平均值（明确的策略）
mean_price = df.groupby(['store', 'cat'])['price'].mean()
agg['wavg_price_safe'] = np.where(agg['w'] > 0, agg['wx'] / agg['w'], mean_price.reindex(agg.index).to_numpy()
)

时间感知分组

使用带频率的 pd.Grouper

通过将时间序列数据分组到特定间隔来尊重KPI的日历边界。

weekly = df.groupby(['store', pd.Grouper(key='ts', freq='W')], observed=True).agg(sales=('rev', 'sum'), orders=('order_id', 'nunique')
)

对每个分组应用滚动/扩展窗口

始终先对数据进行排序，并基于时间戳列对齐。

df = df.sort_values(['customer_id', 'ts'])
df['rev_30d_mean'] = (df.groupby('customer_id').rolling('30D', on='ts')['rev'].mean().reset_index(level=0, drop=True)
)

避免数据泄漏

保持时间顺序，并确保窗口只能“看到”过去的数据。不要打乱时间序列数据，也不要在为训练和测试分割数据集之前计算全数据集的组统计量。

组内排名和前N名

查找每组的前k行

以下是选择每组前N行的两种实用选项。

# 排序 + head
top3 = (df.sort_values(['cat', 'rev'], ascending=[True, False]).groupby('cat').head(3))# 对单一指标进行组内 nlargest
top3_alt = (df.groupby('cat', group_keys=False).apply(lambda g: g.nlargest(3, 'rev')))

使用辅助函数

Pandas提供了几个用于排名和选择的辅助函数。

rank — 控制如何处理并列情况（例如，method='dense' 或 'first'），并可以使用 pct=True 计算百分位排名。
```
df['rev_rank_in_cat'] = df.groupby('cat')['rev'].rank(method='dense', ascending=False)
```
cumcount — 提供每行在其组内的基于0的位置。
```
df['pos_in_store'] = df.groupby('store').cumcount()
```

nth — 在不排序整个DataFrame的情况下，选取每组中的第k行。

second_row = df.groupby('store').nth(1)  # 每家店的第二行（按出现顺序）

使用 transform 广播特征

执行组内标准化

在每个组内标准化一个指标，使得不同组间的行具有可比性。

g = df.groupby('store')['rev']
df['rev_z'] = (df['rev'] - g.transform('mean')) / g.transform('std')

插补缺失值

用组统计量填充缺失值。这通常比使用全局填充值更能保持接近现实的分布。

df['price'] = df['price'].fillna(df.groupby('cat')['price'].transform('median'))

创建组内份额特征

将原始数字转换为组内比例，以便进行更清晰的比较。

df['rev_share_in_store'] = df['rev'] / df.groupby('store')['rev'].transform('sum')

处理类别、空组和缺失数据

使用分类类型提高速度

如果你的键来自一个固定的集合（例如，商店、地区、产品类别），将它们转换为分类类型一次。这会使GroupBy操作更快、更节省内存。

from pandas.api.types import CategoricalDtypestore_type = CategoricalDtype(categories=sorted(df['store'].dropna().unique()), ordered=False)
df['store'] = df['store'].astype(store_type)cat_type = CategoricalDtype(categories=['Grocery', 'Electronics', 'Home', 'Clothing', 'Sports'])
df['cat'] = df['cat'].astype(cat_type)

丢弃未使用的组合

在分类列上进行分组时，设置 observed=True 会排除数据中实际未出现的类别组合，从而得到更干净、干扰更少的输出。

out = df.groupby(['store', 'cat'], observed=True).size().reset_index(name='n')

使用 NaN 键进行分组

明确说明如何处理缺失键。默认情况下，Pandas会删除NaN组；仅当它有助于你的质量保证过程时才保留它们。

# 默认：NaN 键被删除
by_default = df.groupby('region').size()# 当你需要审计缺失键时，将 NaN 保留为其自己的组
kept = df.groupby('region', dropna=False).size()

快速备忘单

计算每组的条件比率

# 布尔值的平均值就是比率
df.groupby(keys).agg(rate=('flag', 'mean'))
# 或显式地：sum(mask)/size
df.groupby(keys).agg(rate=('flag', lambda s: s.sum() / s.size))

计算加权平均值

df.assign(wx=df[x] * df[w]).groupby(keys).apply(lambda g: g['wx'].sum() / g[w].sum() if g[w].sum() else np.nan).rename('wavg')

查找每组的前k名

(df.sort_values([key, metric], ascending=[True, False]).groupby(key).head(k))
# 或
df.groupby(key, group_keys=False).apply(lambda g: g.nlargest(k, metric))

计算每周指标

df.groupby([key, pd.Grouper(key='ts', freq='W')], observed=True).agg(...)

执行组内填充

df[col] = df[col].fillna(df.groupby(keys)[col].transform('median'))

计算组内份额

df['share'] = df[val] / df.groupby(keys)[val].transform('sum')

总结

首先，为你的任务选择正确的模式：使用 agg 进行缩减，使用 transform 进行广播，并仅在无法向量化时才保留使用 apply。依靠 pd.Grouper 处理基于时间的时段，依靠排名辅助函数进行前N名选择。通过倾向于清晰、向量化的模式，你可以保持输出扁平化、命名清晰且易于测试，确保你的指标保持正确，你的笔记本运行快速。
更多精彩内容请关注我的个人公众号公众号（办公AI智能小助手）或者我的个人博客 https://blog.qife122.com/
对网络安全、黑客技术感兴趣的朋友可以关注我的安全公众号（网络安全技术点滴分享）

公众号二维码

公众号二维码