在第 13 章掌握了 groupby 基础后，本篇进入进阶：分组+滚动窗口、分组内 shift/diff、自定义聚合对象、named agg 高级用法、链式流水线、分组时间序列重采样。这些是数据分析实战中真正高频的高阶技巧。
目录
1. 进阶总览
2. 分组 + 滚动窗口
3. 分组内 shift / diff / 排名
4. NamedAgg 与多列命名聚合
5. 自定义聚合函数与聚合对象
6. 分组链式流水线
7. 分组 + 时间序列 resample
8. groupby 与 apply 的返回值规则
9. 按函数/索引层级分组
10. 分组性能优化
11. 常见坑与注意事项
12. 本章小结与练习
1. 进阶总览
| 技巧 | 一句话说明 | 典型场景 |
|------|-----------|----------|
| 分组滚动 | groupby(...).rolling(n) | 每个部门/商品算 7 日均值 |
| 分组 shift | groupby(...)[col].shift(1) | 每个客户的环比 |
| NamedAgg | 一行写多个命名聚合 | 生成规范统计表 |
| 自定义聚合对象 | Aggregate 类 | 复杂统计逻辑复用 |
| 链式流水线 | .groupby().agg().reset_index().sort_values() | 一行完成整套统计 |
| 分组重采样 | groupby(...).resample(...) | 每个分组的时间序列 |
| 函数分组 | groupby(callable) | 按自定义规则分组 |
准备数据：
import pandas as pd
import numpy as np
np.random.seed(42)
df = pd.DataFrame({
"部门": np.random.choice(["技术", "销售", "市场"], 200),
"员工": [f"E{i}" for i in range(200)],
"工资": np.random.randint(5000, 20000, 200),
"日期": pd.date_range("2025-01-01", periods=200, freq="D"),
})
print(df.head())
2. 分组 + 滚动窗口
groupby(...).rolling(window) 对每个分组内部做滚动计算：
# 每个部门内部按日期排序后做 7 天滚动均值
df_sorted = df.sort_values(["部门", "日期"])
df_sorted["工资_7日均值"] = (
df_sorted.groupby("部门")["工资"]
.rolling(7, min_periods=1)
.mean()
.reset_index(level=0, drop=True) # 对齐回原索引
)
print(df_sorted.head(10))
部门 员工 工资 日期 工资_7日均值
0 市场 E0 11851 2025-01-01 11851.000000
1 市场 E1 5800 2025-01-02 8825.500000
2 市场 E2 14370 2025-01-03 10673.666667
...
关键：rolling的结果索引是 MultiIndex（部门, 原索引），用reset_index(level=0, drop=True)去掉部门层，才能按原索引对齐回df_sorted。
# 分组滚动 + 多窗口
df_sorted["工资_30日均值"] = (
df_sorted.groupby("部门")["工资"]
.rolling(30, min_periods=5)
.mean()
.reset_index(level=0, drop=True)
)
# 分组滚动排名/分位
df_sorted["部门内排名"] = df_sorted.groupby("部门")["工资"].rank()
df_sorted["部门内分位"] = df_sorted.groupby("部门")["工资"].rank(pct=True)
3. 分组内 shift / diff / 排名
# 每个部门内部：上一名员工的工资（分组 shift）
df_sorted["上一人工资"] = df_sorted.groupby("部门")["工资"].shift(1)
# 分组内环比（本员工 vs 组内上一人）
df_sorted["环比变化"] = df_sorted["工资"] - df_sorted.groupby("部门")["工资"].shift(1)
df_sorted["环比变化率"] = df_sorted.groupby("部门")["工资"].pct_change()
# 分组内累计
df_sorted["组内累计"] = df_sorted.groupby("部门")["工资"].cumsum()
# 分组内 diff
df_sorted["组内差分"] = df_sorted.groupby("部门")["工资"].diff()
# 分组内分位数排名
df_sorted["工资百分位"] = df_sorted.groupby("部门")["工资"].rank(pct=True)
print(df_sorted[df_sorted["部门"] == "市场"].head(6))
| 方法 | 作用 |
|------|------|
| groupby(...)[col].shift(n) | 组内滞后 n 行 |
| groupby(...)[col].diff() | 组内差分 |
| groupby(...)[col].pct_change() | 组内环比变化率 |
| groupby(...)[col].cumsum() | 组内累计和 |
| groupby(...)[col].rank(pct=True) | 组内排名（百分比） |
4. NamedAgg 与多列命名聚合
NamedAgg 让"多列 × 多聚合"的结果列名完全可控：
# 方式 1：named agg 元组语法
result = df.groupby("部门").agg(
平均工资=("工资", "mean"),
最高工资=("工资", "max"),
最低工资=("工资", "min"),
人数=("员工", "count"),
工资总和=("工资", "sum"),
)
print(result)
平均工资 最高工资 最低工资 人数 工资总和
部门
市场 11687.3 19793 5301 69 806423
技术 12149.8 19985 5009 66 801889
销售 12255.3 19989 5009 65 796594
# 方式 2：NamedAgg 类（等价）
from pandas import NamedAgg
result2 = df.groupby("部门").agg(
平均工资=NamedAgg(column="工资", aggfunc="mean"),
人数=NamedAgg(column="员工", aggfunc="count"),
)
# 方式 3：自定义函数 + 重命名
def spread(x):
return x.max() - x.min()
result3 = df.groupby("部门")["工资"].agg(
diff=spread,
std=lambda x: round(x.std(), 2),
)
print(result3)
# 复杂组合：不同列不同聚合
result4 = df.groupby("部门").agg({
"工资": ["mean", "std"],
"员工": "count",
})
print(result4.columns) # MultiIndex 列
推荐方式 1（named agg 元组）：列名扁平、无 MultiIndex、可读性好。
5. 自定义聚合函数与聚合对象
5.1 多个自定义函数
def q25(x):
return x.quantile(0.25)
def q75(x):
return x.quantile(0.75)
def iqr(x):
return q75(x) - q25(x)
result = df.groupby("部门")["工资"].agg([q25, q75, iqr])
print(result)
5.2 聚合对象（Aggregate 类）——逻辑复用
from pandas.core.base import PandasObject
# 自定义聚合类：一次性返回多个统计量
class RobustStats(PandasObject):
def __init__(self, col):
self.col = col
def __call__(self, group):
s = group[self.col]
return pd.Series({
"中位数": s.median(),
"IQR": s.quantile(0.75) - s.quantile(0.25),
"均值": s.mean(),
})
result = df.groupby("部门").apply(RobustStats("工资"))
print(result)
中位数 IQR 均值
部门
市场 11637.0 5374.25 11687.304348
技术 12543.0 5634.50 12149.787879
销售 12795.0 5858.00 12255.307692
5.3 分组聚合函数注意
# 聚合函数只接收"当前组的列值"（一维），返回标量
def my_agg(x):
# x 是 Series（当前组这一列的所有值）
return x.sum() / len(x) * 2 # 标量返回
df.groupby("部门")["工资"].agg(my_agg)
6. 分组链式流水线
用 .pipe 和链式调用把分组统计写成一行流水线（参考第 10 章）：
result = (
df
.groupby("部门")["工资"]
.agg(平均="mean", 最高="max", 最低="min")
.reset_index() # 部门变回普通列
.sort_values("平均", ascending=False)
.assign(极差=lambda d: d["最高"] - d["最低"])
.rename(columns={"平均": "平均工资"})
)
print(result)
部门 平均工资 最高 最低 极差
0 销售 12255.307692 19989 5009 14980
1 技术 12149.787879 19985 5009 14976
2 市场 11687.304348 19793 5301 14492
常见流水线结尾：
# 分组 -> 多聚合 -> 宽表 -> 排名
top = (
df.groupby("部门")["工资"]
.agg(["mean", "sum"])
.sort_values("sum", ascending=False)
.assign(rank=lambda d: d["sum"].rank(ascending=False))
)
print(top)
7. 分组 + 时间序列 resample
groupby(...).resample(...) 先分组、再对每个分组做时间重采样：
# 给每个部门构造跨日期的序列
np.random.seed(1)
ts = pd.DataFrame({
"部门": np.random.choice(["A", "B"], 100),
"日期": pd.date_range("2025-01-01", periods=100, freq="D"),
"销量": np.random.randint(10, 100, 100),
})
# 按部门+周 汇总销量
weekly = (
ts.set_index("日期")
.groupby("部门")
.resample("W")
.agg({"销量": "sum"})
)
print(weekly.head())
销量
部门 日期
A 2025-01-05 211
2025-01-12 260
2025-01-19 311
...
# 每部门每月均值（透视输出）
monthly = (
ts.set_index("日期")
.groupby("部门")
.resample("M")["销量"]
.mean()
.unstack(level=0)
)
print(monthly)
日期 A B
2025-01-31 46.806452 52.193548
2025-02-28 50.107143 48.535714
2025-03-31 55.580645 51.322581
2025-04-10 52.750000 53.250000
groupby("部门").resample("W")的等价写法：ts.groupby(["部门", pd.Grouper(freq="W")])。
8. groupby 与 apply 的返回值规则
apply 的分组函数可以返回标量、Series、DataFrame，pandas 会自动"拼装"：
# 返回标量 -> 一列
df.groupby("部门").apply(lambda g: g["工资"].mean())
# 返回 Series -> 多列
df.groupby("部门").apply(lambda g: pd.Series({
"均值": g["工资"].mean(),
"人数": len(g),
}))
# 返回 DataFrame -> 分组前缀索引
df.groupby("部门").apply(lambda g: g.nlargest(2, "工资"))
重要规则：
| 返回值 | 结果结构 | 注意 |
|--------|----------|------|
| 标量 | Series（分组键为索引） | 与 agg 类似 |
| Series | DataFrame（分组键 + Series 索引为行） | 索引变成 MultiIndex |
| DataFrame | DataFrame | 分组键作为最外层索引 |
| 标量（include_groups=False） | Series | 3.x 新行为 |
# pandas 2.x 建议显式声明
df.groupby("部门").apply(lambda g: g["工资"].mean(), include_groups=False)
能不用 apply 就不用 apply；apply 在复杂逻辑（如每组取 TopN、每组构建子模型）时才值得。
9. 按函数/索引层级分组
# 按函数分组：工资是否过万
df["高薪"] = df["工资"] >= 10000
print(df.groupby("高薪")["工资"].mean())
# 按自定义函数（对索引应用）
df_idx = df.set_index("部门")
print(df_idx.groupby(lambda x: x).size()) # 等价按部门
# 按索引层级分组（MultiIndex 时）
df_mi = df.set_index(["部门", "员工"])
print(df_mi.groupby(level=0)["工资"].sum()) # 按第一层
print(df_mi.groupby(level="部门")["工资"].sum()) # 按名称
print(df_mi.groupby(level=[0, 1])["工资"].sum()) # 按多层
# 用 pd.cut 分箱分组
df["工资档"] = pd.cut(df["工资"], bins=[0, 8000, 12000, 20000],
labels=["低", "中", "高"])
print(df.groupby("工资档", observed=False)["工资"].agg(["count", "mean"]))
10. 分组性能优化
| 问题 | 优化 |
|------|------|
| apply 太慢 | 尽量用 agg/transform 内置函数 |
| 分组键是 object | 转 category 可提速 |
| 多次 groupby 同一键 | 先 g = df.groupby("部门") 复用 |
| sort=False | 不需要排序结果时 groupby(..., sort=False) |
| 大表多次聚合 | 一次 agg 代替多次 groupby |
# 复用分组对象
g = df.groupby("部门", sort=False)
mean_salary = g["工资"].mean()
count_emp = g["员工"].count()
# 性能对比示例（%timeit 在 Jupyter 中使用）
# %timeit df.groupby("部门")["工资"].mean()
11. 常见坑与注意事项
| 坑 | 现象 | 解决办法 |
|----|------|----------|
| rolling 结果索引错位 | 对齐回原表出错 | reset_index(level=0, drop=True) |
| named agg 语法错误 | 报错 | 使用 列=("字段", "函数") 元组 |
| apply 返回形状不定 | 结果难预料 | 返回固定结构的 Series/DataFrame |
| groupby 后忘 reset_index | 索引多层 | reset_index() 或 as_index=False |
| resample 分组键混淆 | 重采样结果错误 | 先 set_index("日期") 再分组 |
| apply 里引用外部变量 | 结果错误 | 用闭包/partial 正确传参 |
12. 本章小结与练习
小结
- 分组滚动：
groupby(...).rolling(n)+ 索引对齐； - 组内 shift/diff/rank/cumsum 是环比分析的利器；
- NamedAgg 让聚合列名完全可控；
- 链式流水线让统计一步到位；
groupby + resample处理分组时间序列；- apply 能解决一切，但优先用内置聚合。
练习题
1. 对销售数据按"区域"分组计算 7 天滚动均值并正确对齐。
2. 用 NamedAgg 生成"区域×城市"的平均金额和订单数。
3. 用链式写法：分组 → 聚合 → 排序 → 加排名列。
4. 对"部门×日期"数据做按周的 resample 汇总。
5. 用 pd.cut 把金额分档后分组统计。
下一篇预告：第 20 章 时间序列高级 —— 工作日历、节假日、时区、Period 与频率转换。
文章回复
0 条公开回复