1. 项目概述：为什么多维聚合不是“高级技巧”，而是日常分析的呼吸本身

你有没有过这种经历：凌晨两点，报表系统突然报错，下游BI看板一片空白，而业务方的钉钉消息已经炸了——“客户分群模型今天没跑出来，风控策略全停了！”你点开日志，发现核心问题竟然是： groupby().agg() 返回了一个带双层索引的 DataFrame，下游代码直接 .to_dict() 报 KeyError ；或者更糟，滚动窗口计算时忘了 .reset_index(level=0, drop=True) ，导致时间序列对齐全乱，欺诈识别阈值偏移37%。这不是玄学故障，这是多维聚合在真实生产环境里最朴素的“落地震颤”。

我干了十年数据工程和银行分析系统搭建，从最早用 SQL 写嵌套子查询拼年同比、季环比，到后来在 Spark 上调度千万级交易流，再到如今在 Pandas 里写单机可复现的分析原型——所有这些场景里， 多维聚合从来不是“锦上添花”的炫技模块，而是整个分析链路的承重墙 。它不声不响地扛着三件事：第一，把原始流水（transaction log）变成可解释的业务语言（比如“华东区高端客群在奢侈品类目上的月均消费波动率”）；第二，让不同团队（风控、运营、财务）能用同一份底层聚合结果，各自衍生指标，避免“一个需求一套 SQL”带来的口径撕裂；第三，为后续建模提供稳定、低噪声、带时间上下文的特征输入。你看到的 dashboard 上那个“近7天客单价趋势图”，背后可能是 groupby(['customer_segment', 'date']).rolling(7).mean() 套着 unstack('customer_segment') 再接 fillna(method='ffill') 的三层嵌套；你收到的邮件里那句“高风险商户清单已更新”，源头可能是一段 agg({'amount': lambda x: x.max() - x.min(), 'count': 'sum'}) 加条件过滤。

这篇文章讲的，不是 Pandas 文档里“ agg() 方法支持字典传参”这种教科书定义，而是我在某全国性股份制银行做信用卡反欺诈模型迭代时，被业务方连续三次推翻需求后，亲手重写的第七版聚合逻辑——它要同时满足：① 按客户ID+商户类别双维度聚合；② 对每组输出均值、中位数、极差、标准差四个统计量；③ 对金额列额外计算滚动30天变异系数（标准差/均值）；④ 结果必须能一键导出 Excel，表头是“客户ID | 商户类别 | 均值 | 中位数 | 极差 | 标准差 | 30日变异系数”；⑤ 全流程运行时间不能超过12秒（因为要嵌入T+1批处理流水线）。这五个约束条件，逼我放弃了所有“优雅但慢”的写法，最终用 agg() 字典映射 + rolling().apply() 自定义函数 + unstack().reset_index() 三板斧，在保证可读性的前提下压到了9.8秒。后面我会逐行拆解这个真实案例的每一行代码为什么这么写、不那么写会掉进什么坑。

关键词里的 “Towards AI - Medium” 不是平台背书，而是提醒你：这里没有“理论正确但无法上线”的空中楼阁。每一个示例都来自银行、保险、支付机构的真实数据管道，所有参数（比如滚动窗口设为7天还是30天、极差阈值定为200还是500）都有明确的业务依据——不是拍脑袋，而是和风控总监在会议室白板上画完客户资金流向图后共同敲定的。如果你正在为“怎么把几十个指标塞进一次 groupby”发愁，或者被“滚动计算结果对不上Excel手工验算”折磨得睡不着，那你来对地方了。接下来的内容，全是硬核实操，没有一句废话。

2. 多维聚合的核心设计逻辑：为什么必须放弃“先 groupby 再 merge”的旧思维

2.1 从“三步走”到“一步到位”：性能与可维护性的双重碾压

十年前我刚接手某城商行的贷后监控系统时，分析师提交的需求是：“请输出各分行、各产品类型下，近90天逾期率、平均逾期天数、首逾客户数”。当时团队的“标准解法”是写三个独立 SQL：

-- 查询1：各分行各产品逾期率
SELECT branch, product, COUNT(CASE WHEN overdue_days > 0 THEN 1 END) * 1.0 / COUNT(*) as overdue_rate
FROM loan_table WHERE report_date >= DATE_SUB(CURDATE(), INTERVAL 90 DAY)
GROUP BY branch, product;

-- 查询2：各分行各产品平均逾期天数  
SELECT branch, product, AVG(overdue_days) as avg_overdue_days
FROM loan_table WHERE report_date >= DATE_SUB(CURDATE(), INTERVAL 90 DAY) AND overdue_days > 0
GROUP BY branch, product;

-- 查询3：各分行各产品首逾客户数
SELECT branch, product, COUNT(DISTINCT customer_id) as first_overdue_cnt
FROM loan_table WHERE report_date >= DATE_SUB(CURDATE(), INTERVAL 90 DAY) 
  AND overdue_days > 0 AND first_overdue_flag = 1
GROUP BY branch, product;

然后用 Python 把三张结果表按 (branch, product) 主键 merge 起来。这套方案的问题在半年后集中爆发：当贷款余额突破千亿，单日新增记录达200万条时，三个查询总耗时从18秒飙升到217秒，且每次需求变更（比如新增“逾期金额占比”指标），都要改三处SQL、调三遍Python、验三次数据一致性。根本原因在于： 重复扫描同一张大表三次，且每次扫描都带不同过滤条件和聚合逻辑，磁盘IO和CPU成了瓶颈 。

Pandas 的 agg() 字典映射彻底终结了这种低效。回到上面的需求，一行代码搞定：

# 假设 df_loan 是已加载的贷款明细DataFrame
result = df_loan[
    df_loan['report_date'] >= (pd.Timestamp.today() - pd.DateOffset(days=90))
].groupby(['branch', 'product']).agg({
    'overdue_days': [
        ('overdue_rate', lambda x: (x > 0).mean()),
        ('avg_overdue_days', lambda x: x[x > 0].mean()),
        ('first_overdue_cnt', lambda x: df_loan.loc[x.index, 'first_overdue_flag'].sum())
    ],
    'loan_amount': [
        ('overdue_amt_ratio', lambda x: df_loan.loc[x.index, 'overdue_amount'].sum() / x.sum())
    ]
})

提示：这里 lambda x: (x > 0).mean() 计算逾期率，本质是布尔数组均值，比 COUNT(CASE...) / COUNT(*) 更简洁； df_loan.loc[x.index, 'first_overdue_flag'].sum() 利用 groupby 分组后 x.index 就是该组原始行索引的特性，精准定位到同组的 first_overdue_flag 字段求和，避免了SQL中复杂的关联子查询。

为什么这行代码能快？因为它只对原始数据做 一次扫描、一次分组、一次内存遍历 。Pandas 在底层用 Cython 实现了高效的分组引擎，当 agg() 接收字典时，它会为每个键值对预编译聚合路径，所有计算共享同一个分组上下文。实测对比：对1000万行贷款数据，传统三SQL方案（通过 SQLAlchemy 执行）平均耗时142秒；而 Pandas 单次 agg() 仅需6.3秒，性能提升22倍。更重要的是可维护性——新增一个指标，只需在字典里加一行键值对，无需动其他逻辑。

2.2 层级列名（Hierarchical Columns）：不是bug，是设计精妙的接口契约

新手看到 agg() 输出的双层列名常一脸懵：“ transaction_amount 下面怎么还套着 mean 、 median ？我要取‘餐饮类均值’还得写 result[('transaction_amount', 'mean')] ？太反人类了！” 这其实是 Pandas 故意为之的 强类型接口设计 。想象一下，如果所有聚合结果都扁平化成 'transaction_amount_mean' 、 'transaction_amount_median' 这种字符串，当你需要动态生成指标（比如根据配置文件决定计算哪些统计量）时，就得用 getattr() 或 eval() 拼接字符串，极易出错且无法被IDE智能提示。

层级列名强制你用元组索引，这带来了两个关键好处：第一， 语义清晰不可歧义 。 ('amount', 'std') 明确表示“金额列的标准差”，不会和 'fee_std' （手续费标准差）混淆；第二， 支持批量操作 。你可以用 result['amount'] 直接选取所有 amount 相关的聚合列（ mean 、 median 、 std ），再统一 .round(2) ；或者用 result.xs('mean', level=1, axis=1) 快速提取所有指标的均值列。这在生成日报表时极其高效——比如风控日报需要所有指标的均值和标准差，运营日报需要均值和极差，你只需用 xs() 切片即可，不用重写聚合逻辑。

注意：生产环境中务必处理层级列名的导出兼容性。Excel 和 BI 工具通常不认双层索引，所以最后一步一定是 result.columns = ['_'.join(col).strip() for col in result.columns.values] 扁平化，或用 result.stack(0).unstack(1) 调整结构。我见过太多团队因忽略这点，导致自动化报表导出后列名显示为 ("amount", "mean") 这种JSON格式，业务方直接崩溃。

2.3 方案选型的底层逻辑：什么时候该用 agg() ，什么时候该切回 apply() ？

agg() 和 apply() 常被混用，但它们有本质分工： agg() 是 向量化聚合函数的容器 ，适合对每列独立施加统计运算（sum、mean、自定义lambda）； apply() 是 行/列级函数执行器 ，适合需要跨列计算或复杂状态管理的场景。选错会导致性能断崖式下跌。

举个典型反例：某支付公司要做“单日交易欺诈风险评分”，规则是：若单日同一设备ID的交易笔数 > 5 且总金额 > 10000，则风险分=100；否则按 log(金额+1) * 笔数 计算。错误写法：

# ❌ 错误：在 agg() 里强行跨列计算，触发 Python 循环
df.groupby('device_id').agg({
    'amount': 'sum',
    'count': 'count'
}).apply(lambda row: 100 if (row['count'] > 5 and row['amount'] > 10000) else np.log(row['amount']+1)*row['count'], axis=1)

这会让 agg() 先完成分组聚合，再用 apply() 对结果 DataFrame 做逐行计算，看似合理，但 apply() 的 axis=1 是纯Python循环，10万组数据就要执行10万次Python函数调用，速度极慢。

正确解法是 apply() 直接作用于原始分组，利用 x['amount'] 和 x['count'] 获取当前组的Series ：

# ✅ 正确：在 apply 中直接访问分组内原始数据，保持向量化
def fraud_score(group):
    total_amt = group['amount'].sum()
    total_cnt = len(group)
    if total_cnt > 5 and total_amt > 10000:
        return pd.Series({'risk_score': 100})
    else:
        # 注意：这里用 group['amount'] 是整个组的金额Series，可直接向量化计算
        return pd.Series({'risk_score': (np.log(group['amount'] + 1) * total_cnt).mean()})

result = df.groupby('device_id').apply(fraud_score).reset_index()

核心判断原则： 如果聚合逻辑只依赖单列数据（如计算金额均值、手续费极差），用 agg() ；如果必须同时读取多列原始值、或需要组内行间关系（如“首笔交易时间”、“最大单笔占比”），则必须用 apply() 。这个原则帮我避开了至少二十次线上性能事故。

3. 四大核心聚合模式的深度实操解析

3.1 多列多函数聚合：如何用字典映射榨干一次分组的全部价值

3.1.1 基础语法与业务映射表

agg() 的字典语法看似简单，但其键值对设计直指业务本质。以银行信用卡分析为例，不同团队关注点截然不同：

团队	关注指标	数据来源列	聚合函数	业务含义
风控部	交易金额极差、标准差、95分位数	amount	lambda x: x.max()-x.min() 、 'std' 、 lambda x: x.quantile(0.95)	衡量客户交易波动性，波动越大欺诈风险越高
运营部	日均交易笔数、手续费收入范围	count , fee	'mean' , lambda x: (x.min(), x.max())	评估商户合作质量，手续费范围异常提示结算问题
财务部	月度总交易额、手续费总收入	amount , fee	'sum'	直接生成损益表核心数据

将这些需求翻译成 agg() 字典，就是：

# 真实银行生产代码片段（已脱敏）
risk_ops_finance_agg = {
    'amount': [
        ('amt_range', lambda x: x.max() - x.min()),
        ('amt_std', 'std'),
        ('amt_p95', lambda x: x.quantile(0.95)),
        ('amt_sum', 'sum')
    ],
    'count': [
        ('daily_txn_cnt', 'mean')
    ],
    'fee': [
        ('fee_minmax', lambda x: f"{x.min():.2f}-{x.max():.2f}"),
        ('fee_sum', 'sum')
    ]
}

result = df_txn.groupby(['customer_segment', 'merchant_category']).agg(risk_ops_finance_agg)

注意 fee_minmax 的写法：用 lambda 返回格式化字符串，而非数值。这体现了 agg() 的灵活性——它不强制输出数值，只要你的函数能对Series返回标量即可。但要警惕：返回字符串会污染后续数值计算，所以这类字段应明确命名含 _str 后缀，并在导出前与其他数值列分离。

3.1.2 高阶技巧：用 namedtuple 实现原子化指标封装

当某个业务指标逻辑复杂（如“有效交易率”=非拒付交易数/总交易数），且需在多个分析中复用时，用匿名 lambda 会降低可读性。此时应定义命名元组（ namedtuple ）封装计算逻辑：

from collections import namedtuple

# 定义“交易健康度”指标，包含三个原子值
TransactionHealth = namedtuple('TransactionHealth', ['valid_rate', 'avg_delay_hours', 'retry_ratio'])

def calc_transaction_health(group):
    """计算单组交易健康度，返回命名元组便于解包"""
    total = len(group)
    valid = group[group['status'] == 'success'].shape[0]
    delay_hours = (group['settle_time'] - group['txn_time']).dt.total_seconds().mean() / 3600
    retry_cnt = group[group['retry_flag'] == True].shape[0]
    
    return TransactionHealth(
        valid_rate=valid / total if total > 0 else 0,
        avg_delay_hours=delay_hours,
        retry_ratio=retry_cnt / total if total > 0 else 0
    )

# 在 agg 中使用
result = df_txn.groupby('channel').agg({
    'amount': [('health_metrics', calc_transaction_health)]
})

# 解包结果（自动展开为三列）
health_df = pd.DataFrame(result['health_metrics'].tolist(),
                        index=result.index,
                        columns=['valid_rate', 'avg_delay_hours', 'retry_ratio'])

这样做的好处是：业务逻辑完全隔离在 calc_transaction_health 函数内，单元测试可直接对该函数打桩验证；结果解包后是标准 DataFrame，无缝接入下游分析。我在某第三方支付公司的清算对账系统中，用此模式封装了“资金到账及时率”、“通道成功率”等8个核心健康度指标，使月度对账报告生成代码从300行缩减到80行。

3.2 自定义聚合函数：从“计算”到“业务决策”的最后一公里

3.2.1 Lambda 的边界与陷阱

lambda 适合单行简单逻辑，如 lambda x: x.max() - x.min() 。但一旦涉及条件分支、异常处理或多次计算，就必须升格为命名函数。常见陷阱：

• 陷阱1：lambda 中引用外部变量导致闭包问题

  # ❌ 危险：threshold 在循环中变化，所有lambda共享最后一个值
  thresholds = {'Retail': 200, 'Dining': 150}
  for cat, th in thresholds.items():
      agg_dict[cat] = lambda x, t=th: (x > t).sum()  # 必须用 t=th 绑定当前值
  
  # ✅ 正确：用默认参数捕获当前循环变量值
  

• 陷阱2：lambda 返回None引发聚合中断

  # ❌ 若series为空，min()/max()抛ValueError，lambda返回None，agg失败
  lambda x: x.min() if len(x) > 0 else 0
  
  # ✅ 正确：显式处理空Series
  lambda x: x.min() if not x.empty else 0
  

3.2.2 命名函数的工业级写法：文档即契约

生产环境的自定义函数必须自带文档字符串，且文档要说明 业务意图 ，而非技术实现：

def weighted_txn_avg(series):
    """
    计算加权交易均值：近期交易权重更高，反映客户最新消费能力。
    
    【业务依据】风控模型验证显示，过去30天交易对信用额度预测贡献度达68%，
    因此采用线性递增权重（第1天权重0.5，最后1天权重1.5）。
    
    Parameters:
    -----------
    series : pd.Series
        交易金额序列，索引为datetime（已按时间排序）
        
    Returns:
    --------
    float
        加权平均金额，保留2位小数
        
    Examples:
    ---------
    >>> s = pd.Series([100, 200, 300], index=pd.date_range('2024-01-01', periods=3))
    >>> weighted_txn_avg(s)
    233.33
    """
    if len(series) < 2:
        return float(series.mean())
    
    # 权重向量：从0.5线性增至1.5
    weights = np.linspace(0.5, 1.5, len(series))
    weighted_avg = np.average(series, weights=weights)
    return round(weighted_avg, 2)

这段文档的价值在于：六个月后新来的分析师看到 weighted_txn_avg ，不用翻历史会议纪要，就能从docstring里读懂“为什么用线性权重”、“为什么起始0.5结束1.5”、“少于2笔交易如何兜底”。这才是真正的可审计、可传承。

3.2.3 复杂状态聚合：用 apply() 实现跨行逻辑

有些指标必须知道组内行序，比如“首笔交易金额占比”（首笔金额 / 总金额）。这无法用 agg() 实现，因为 agg() 的每个函数只看到一列数据，不知道哪行是第一行：

def first_txn_ratio(group):
    """计算首笔交易金额占总金额比例"""
    # group 按时间排序后取第一行
    first_amount = group.nsmallest(1, 'txn_time')['amount'].iloc[0]
    total_amount = group['amount'].sum()
    return first_amount / total_amount if total_amount > 0 else 0

# ✅ 正确：apply 作用于整个分组DataFrame
result = df_txn.groupby('customer_id').apply(first_txn_ratio)

更进一步，“最近N笔交易的金额标准差”也需 apply() ：

def recent_n_std(group, n=5):
    """计算最近n笔交易金额的标准差"""
    # 按时间倒序取前n笔
    recent_n = group.nlargest(n, 'txn_time')['amount']
    return recent_n.std() if len(recent_n) >= 2 else 0

result = df_txn.groupby('customer_id').apply(recent_n_std, n=5)

实操心得： apply() 的 group 参数是原始分组的 DataFrame（非Series），所以你可以自由访问任意列、任意行操作。但代价是性能—— apply() 默认是Python循环，大数据量时务必用 engine='numba' （需安装 numba）或提前用 sort_values().tail(n) 优化。

3.3 滚动窗口聚合：时间维度上的“动态快照”

3.3.1 滚动窗口的本质：滑动子集上的重复聚合

滚动窗口（Rolling Window）不是魔法，它的数学本质是：对时间序列 S[t] ，在每个时间点 t ，取区间 [t-w+1, t] （w为窗口大小）内的子序列 S[t-w+1:t+1] ，对其应用聚合函数。Pandas 的 rolling() 方法正是这一过程的封装。

关键参数解析：

• window=7 ：固定长度7天窗口。若某天无数据，窗口自动向前补足（需配合 min_periods ）。
• min_periods=3 ：窗口内至少3个有效值才计算，否则返回NaN。这是风控系统的生命线——若7天内只有2天有交易，均值无意义，必须标记缺失。
• center=False （默认）：窗口右对齐，即 t 时刻的值基于 [t-6,t] ；设为 True 则中心对齐，基于 [t-3,t+3] ，适用于平滑但不适用于实时监控（未来数据不可知）。

3.3.2 生产级滚动计算：处理时间对齐与缺失值

真实交易数据充满缺失：周末无交易、系统故障丢数据、商户临时歇业。直接 rolling(7).mean() 会因缺失值导致大量NaN，进而污染下游。必须主动治理：

# 正确的滚动均值计算流程
def robust_rolling_mean(df, window=7, min_periods=3):
    """
    健壮的滚动均值：先按日补全，再滚动，最后填充
    """
    # 步骤1：确保时间索引连续（按日）
    full_date_range = pd.date_range(df.index.min(), df.index.max(), freq='D')
    df_full = df.reindex(full_date_range)
    
    # 步骤2：用前向填充补交易数据（假设周末无交易，用周五值代替）
    df_filled = df_full.fillna(method='ffill')
    
    # 步骤3：计算滚动均值，要求至少min_periods个非空值
    rolling_result = df_filled.rolling(window=window, min_periods=min_periods).mean()
    
    # 步骤4：对仍为NaN的位置，用全局均值填充（业务兜底）
    global_mean = df['amount'].mean()
    final_result = rolling_result.fillna(global_mean)
    
    return final_result

# 应用到分组数据
df_sorted = df_txn.sort_values('txn_time').set_index('txn_time')
rolling_by_customer = df_sorted.groupby('customer_id')['amount'].apply(
    lambda x: robust_rolling_mean(x.to_frame(), window=7)
).reset_index(name='rolling_7day_avg')

这个 robust_rolling_mean 函数解决了三个生产痛点：① 时间索引不连续导致窗口计算跳变；② 缺失值过多使 min_periods 失效；③ NaN 传播至下游告警系统。我在某基金公司的申赎流量监控中，用此模式将滚动计算的准确率从72%提升至99.8%。

3.3.3 滚动窗口的业务变形：滚动分位数与变异系数

风控场景常需“滚动变异系数”（标准差/均值），衡量客户近期交易稳定性：

def rolling_cv(series, window=30):
    """计算滚动30天变异系数，规避除零错误"""
    rolling_std = series.rolling(window=window).std()
    rolling_mean = series.rolling(window=window).mean()
    # 防止均值为0导致除零
    cv = rolling_std / rolling_mean.replace(0, np.nan)
    return cv

# 应用
df_txn['cv_30day'] = df_txn.groupby('customer_id')['amount'].apply(rolling_cv)

更高级的“滚动分位数”用于动态设定阈值：

# 动态欺诈阈值：滚动90天金额95分位数的1.2倍
df_txn['fraud_threshold'] = (
    df_txn.groupby('customer_id')['amount']
    .rolling(90).quantile(0.95).values * 1.2
)

注意： rolling().quantile() 返回的是与原索引对齐的Series，但 .values 取出的是numpy数组，需确保长度一致。生产中建议用 .reset_index(drop=True) 显式对齐。

3.4 扩展窗口与多级分组：构建决策矩阵的终极形态

3.4.1 扩展窗口（Expanding）：从“回顾历史”到“累积认知”

扩展窗口与滚动窗口是镜像关系：滚动窗口大小恒定，扩展窗口从起点持续增长。它的核心价值在于 累积性指标 ，如“客户生命周期总消费”、“渠道累计转化率”。但要注意：扩展窗口默认从第一个值开始，若数据有缺失，需指定 min_periods ：

# 正确：设置 min_periods=1，确保首行就有值
df_txn['cumulative_spend'] = (
    df_txn.groupby('customer_id')['amount']
    .expanding(min_periods=1).sum().values
)

# 错误：不设 min_periods，首行返回NaN
# .expanding().sum() # 首行NaN

扩展窗口的隐藏威力在于 组合聚合 。例如“滚动年化收益率”需先算累计收益，再折算年化：

def expanding_annualized_return(series, annual_days=365):
    """计算扩展窗口年化收益率：(累计收益/初始本金)^(365/天数) - 1"""
    # 假设series是每日收益（小数形式，如0.01表示1%）
    cumulative_return = series.expanding().sum()
    # 计算从首日到当前日的天数
    days_elapsed = np.arange(1, len(series)+1)
    # 年化：(1+累计收益)^(365/天数) - 1
    annualized = (1 + cumulative_return) ** (annual_days / days_elapsed) - 1
    return annualized

df_daily_ret['annualized_ret'] = df_daily_ret.groupby('fund_id')['daily_return'].apply(expanding_annualized_return)

3.4.2 多级分组 + unstack：把“立方体”摊成“平面报表”

多级分组（ groupby(['region','product']) ）生成的是 MultiIndex Series，视觉上是树状结构。 unstack() 的作用是将指定层级的索引“旋转”为列，把树状结构压平成二维表格，这正是业务人员最熟悉的交叉分析视图。

关键技巧：

• unstack(level=0) ：将最外层索引（如 region ）转为列；
• unstack(level=1) ：将内层索引（如 product ）转为列；
• fill_value=0 ：用0填充缺失组合（如“西北区无某产品销售”），避免NaN干扰Excel透视；
• unstack().T ：转置后，行变列、列变行，适配不同BI工具需求。

实战案例：某电商大促期间，需实时监控“各省份-各品类”的GMV达成率。原始数据：

sales_data = {
    'province': ['广东','浙江','广东','江苏','浙江','广东'],
    'category': ['手机','电脑','家电','手机','电脑','家电'],
    'gmv': [120000, 85000, 95000, 110000, 78000, 88000],
    'target': [150000, 100000, 120000, 130000, 90000, 100000]
}
df = pd.DataFrame(sales_data)

# 多级分组计算达成率
result = df.groupby(['province','category']).apply(
    lambda x: (x['gmv'].sum() / x['target'].sum() * 100)
).unstack(fill_value=0).round(1)

# 输出：行是省份，列是品类，值是达成率%
# province   手机   电脑   家电
# 广东     102.0  0.0  88.0
# 浙江       0.0  86.7  0.0
# 江苏      84.6  0.0  0.0

提示： unstack() 后若列名是元组（如 ('category', '手机') ），用 result.columns = result.columns.get_level_values(1) 提取内层品类名作为列名，更简洁。

3.4.3 高级变形：用 pivot_table() 替代 groupby().unstack()

当分组逻辑复杂（如需 aggfunc 同时含多种函数）， pivot_table() 更直观：

# 等价于前面的多级分组+unstack，但更声明式
pivot_result = df.pivot_table(
    values=['gmv','target'],
    index='province',
    columns='category',
    aggfunc={'gmv': 'sum', 'target': 'sum'}
)

# 再计算达成率
pivot_result['achieve_rate'] = (pivot_result[('gmv','手机')] / pivot_result[('target','手机')] * 100).round(1)

pivot_table() 的优势是：参数语义清晰（ index / columns / values ），且原生支持多值列聚合，避免手动 unstack() 后的列名处理。但在超大数据集上， groupby().unstack() 内存效率略高。

4. 真实生产环境中的问题排查与避坑指南

4.1 常见问题速查表

问题现象	根本原因	解决方案	我踩过的坑
agg() 输出列名是元组，下游系统报错	BI工具/Excel不支持MultiIndex	result.columns = ['_'.join(map(str, col)) for col in result.columns.values] 扁平化	某次上线前未扁平化，导致Power BI连接后所有字段显示为 ("amount","mean") ，业务方以为系统崩了，紧急回滚
滚动窗口计算结果全为NaN	min_periods 设过大，或数据未按时间排序	① df = df.sort_values('date').set_index('date') ；② min_periods=1 ；③ 检查时间索引是否连续	在某券商行情系统中，因未排序， rolling(5).mean() 对乱序数据返回随机NaN，花了3小时定位
unstack() 报 Index contains duplicate entries	分组键组合存在重复（如同一省同一品类有多条记录未聚合）	先 groupby().agg() 聚合，再 unstack() ；或用 pivot_table() 自动去重	某零售客户数据中， province-category 组合因脏数据重复， unstack() 直接崩溃，改用 pivot_table(values='gmv', index='p', columns='c', aggfunc='sum') 一行解决
自定义函数在 agg() 中报 TypeError: cannot convert the series to <class 'float'>	函数返回了Series或DataFrame，而非标量	确保函数末尾 return 一个标量（int/float/str）；用 print(type(result)) 调试	为计算“交易集中度”写了 lambda x: x.value_counts(normalize=True).max() ，但 value_counts() 返回Series，应改为 .max() 后再 float() 强转
expanding().sum() 结果长度与原DataFrame不一致	expanding() 返回的是MultiIndex Series，索引未对齐	用 .values 取值数组，或 .reset_index(drop=True) 对齐索引	某次在Spark UDF中调用Pandas expanding ，因索引错位导致特征列全乱，模型AUC暴跌0.15

4.2 性能优化黄金法则

4.2.1 内存与速度的平衡术

Pandas 聚合的瓶颈常在内存。当数据超500万行，以下技巧可提速3-10倍：

• 预过滤再聚合 ：永远先 df.query("date >= '2024-01-01'") 再 groupby() ，而非在 agg() 里用 lambda 过滤；
• 选择合适的数据类型 ： category 类型比 object 内存省80%， int32 比 `int6