多维聚合实战：滚动计算、自定义函数与unstack重塑

1. 项目概述：为什么多维聚合不是“高级技巧”，而是日常生存技能

你有没有遇到过这样的场景：凌晨两点，风控系统突然告警，说某类商户的交易波动率飙升了300%；你打开Jupyter Notebook，手忙脚乱地写groupby('merchant')、再groupby('region')、再groupby(['merchant','region'])，最后发现三个结果对不上——因为没处理好索引层级，也没考虑时间窗口的边界效应；又或者，业务方发来一封加急邮件：“请今天下班前给出各城市、各年龄段、各消费频次区间的客单价中位数+标准差+90分位数，还要带滚动30天趋势”，你盯着pandas文档里那行agg({'col': ['mean', 'std']})发呆，心想：“这玩意儿能嵌套函数吗？能带条件吗？能和rolling一起用吗？”

这不是玄学，这是每天发生在银行、保险、电商、SaaS公司数据团队的真实现场。而Part 20讲的“多维聚合”，根本不是教你怎么用.agg()这个API，它是在教你怎么把业务问题翻译成数据操作语言，再把数据操作结果翻译回业务决策语言。我干了十年数据分析和数据工程，从给支行做手工报表，到搭建千万级用户行为分析平台，踩过的坑比读过的文档还多。今天这篇，就是我把所有“当时要是有人早点告诉我就好了”的经验，全掏出来给你。

核心关键词——多维聚合、滚动计算、自定义聚合、unstack重塑、生产级分组策略——它们不是孤立的语法点，而是一套完整的“问题拆解-数据建模-结果交付”工作流。比如“客户盈利能力分产品线+区域”这个需求，表面看是groupby(['product','region'])，但实际要解决的是：如何让财务部看到毛利率，让运营部看到单客复购率，让风控部看到异常交易占比，且三者必须基于同一份清洗后的数据源，不能各自为政跑三次groupby再merge。这就逼着你必须掌握多列聚合映射、自定义函数封装、结果扁平化等一整套组合拳。

这篇文章面向的不是刚学完df.groupby().sum()的新手，也不是只写SQL的分析师，而是那些已经能跑通基础分析，但一到复杂业务场景就卡壳、返工、被业务方反复追问“这个数字怎么来的”的实战派。它不讲理论推导，只讲我在某股份制银行做信用卡反欺诈模型时，怎么用expanding().std()动态调整风险阈值；在某头部电商做大促复盘时，怎么用unstack(fill_value=0)生成零缺失的品类渗透矩阵；在给监管报送材料时，怎么用命名函数+docstring让审计人员一眼看懂“加权平均”背后的业务逻辑。下面，我们就从最常被低估的第一步开始：为什么“一次写对”比“多次调试”重要十倍。

2. 核心设计思路：从“能跑通”到“可维护、可审计、可扩展”的思维跃迁

2.1 多维聚合的本质：业务维度的数学表达

很多人把groupby(['region','product'])当成一个技术动作，其实它是对业务世界的一次建模。Region（区域）和Product（产品）不是两个并列的字符串列，而是具有明确层级关系的业务实体：区域有行政层级（大区→省→市），产品有业务归属（主产品线→子类目→SKU）。当你执行df.groupby(['region','product'])时，pandas默认按字面值精确匹配，但真实业务中，“华东大区”下可能包含“上海”“江苏”“浙江”，而“华东”这个汇总层在原始数据里并不存在。所以真正的多维聚合，第一步永远不是写代码，而是确认业务维度的完备性与一致性。

我经历过最惨的一次：某次季度经营分析会，销售总监指着大屏上“华东区Widget销量环比+15%”的数据问：“为什么和我们CRM系统里的数字差了7%？”查了三天才发现，原始数据里“华东”被拆成了“上海”“江苏”“浙江”三条记录，而CRM系统里“华东”是人工维护的汇总值，且包含了部分未入账的预售订单。最后不是改代码，而是推动数据治理团队建立统一的区域编码主数据表，并在ETL层强制关联。这件事让我彻底明白：没有干净的维度，就没有可靠的聚合。所以我在所有项目启动时，第一件事就是拉着业务方画维度关系图，明确哪些是原子维度（不可再分）、哪些是衍生维度（需计算）、哪些是口径维度（如“活跃用户”需明确定义为“近30天登录≥3次”）。

2.2 为什么拒绝“分开算再合并”？性能与语义的双重陷阱

新手最常犯的错误，就是把一个复合需求拆成多个独立groupby：先算各区域的均值，再算各产品的均值，最后用pd.merge()拼起来。表面上看逻辑清晰，实则埋下两大隐患：

第一是性能灾难。假设你有100万条交易记录，要按['customer_id','category']分组统计均值、中位数、计数。如果分开算：

• df.groupby('customer_id')['amount'].mean()→ 扫描100万行
• df.groupby('category')['amount'].median()→ 再扫描100万行
• df.groupby(['customer_id','category'])['fee'].count()→ 第三次扫描100万行
  总共300万行IO。而用agg()一次搞定：

df.groupby(['customer_id','category']).agg({ 'amount': ['mean', 'median'], 'fee': 'count' })

pandas内部会单次遍历数据，同时计算所有指标，实测在100万行数据上，耗时从2.3秒降到0.8秒，且内存占用降低40%。这不是微优化，当你的数据量涨到亿级，这种差距就是“能跑通”和“超时失败”的区别。

第二是语义断裂。分开算意味着每个groupby都独立处理缺失值、重复值、类型转换。比如amount列有5%的空值，mean()默认跳过，median()也跳过，但如果你手动merge，就可能因索引对齐问题导致某行mean有值而median是NaN。更危险的是时间维度——滚动计算必须保证窗口内数据完全一致，分开算rolling会导致每个指标的窗口起始点错位。我在某支付公司做实时风控时，就因rolling.mean()和rolling.std()分开计算，导致波动率指标在窗口切换时出现毫秒级偏差，误触发了数百次告警。后来全部重构为单次rolling().agg()，问题根除。

2.3 自定义函数：不是“写代码”，而是“固化业务知识”

看到lambda x: x.max() - x.min()，很多人的反应是“哦，求极差”。但在我眼里，这是把一句业务需求翻译成可执行指令的过程。风控经理说：“我们要监控商户交易金额的离散程度，离散度高的商户要提高审核强度。”这句话里的“离散程度”，在数学上可以是标准差、IQR、变异系数，但业务方选“极差”，是因为他们发现：对小商户而言，一笔5000元的异常交易，比连续10笔500元的交易更值得警惕。所以max()-min()不是随便选的，它背后是业务对风险形态的深刻理解。

因此，我坚持用命名函数替代lambda，哪怕只有一行：

def transaction_range(series): """业务定义：交易金额极差 = 最高单笔 - 最低单笔，用于识别高波动商户""" return series.max() - series.min()

好处有三：一是函数名transaction_range比lambda更易读；二是docstring里写明了业务意图，半年后你或同事接手时，不用翻会议纪要就知道为什么用极差而不是标准差；三是便于单元测试——你可以单独验证transaction_range([100,200,500]) == 400，确保逻辑无误。我在某基金公司做业绩归因时，曾用一个200行的calculate_active_return()函数封装了Brinson模型的所有步骤，每次迭代只需改函数内部，调用层代码一行不动，上线后零故障。

2.4 滚动与扩展窗口：时间不是标尺，而是业务节奏的刻度

rolling(window=7).mean()看起来简单，但window=7代表什么？是7个自然日？7个交易日？还是7笔交易？这直接决定分析结论是否有效。在银行理财销售分析中，“7日滚动”必须排除周末和节假日，否则周一的均值会因缺少周五数据而失真；但在电商直播GMV分析中，“7场直播滚动”比“7日滚动”更有意义，因为每场直播的流量结构、商品组合、主播状态都不同。

同样，expanding().sum()也不是简单的累加。我在做SaaS客户LTV预测时，发现直接expanding().sum()会把早期小客户的低额续费和后期大客户的高额合同混在一起，导致曲线失真。最终方案是：先按客户规模分层（A/B/C类），再在每层内做expanding().sum()，最后加权合成。这说明窗口函数的选择，本质是对业务生命周期的理解。没有放之四海而皆准的window=7，只有“贴合业务节奏的window=N”。

3. 实操细节解析：从代码到业务价值的每一处关键注释

3.1 多列聚合的“字典映射”：不只是语法，更是数据契约

看这段代码：

result = df.groupby('merchant_category').agg({ 'transaction_amount': ['mean','median'], 'processing_fee': ['min','max'] })

输出是带MultiIndex列的DataFrame：

transaction_amount processing_fee mean median min max

这个结构看似麻烦，实则是数据契约的体现：外层transaction_amount声明了指标归属的业务域（金额相关），内层mean/median声明了计算逻辑（集中趋势度量）。当你要把结果喂给BI工具或下游API时，这个结构能避免歧义——比如'mean'不会被误认为是processing_fee的均值。

但生产环境必须处理两个现实问题：

第一，列名扁平化。BI工具（如Tableau、Power BI）通常不支持MultiIndex列。我的标准做法是：

result.columns = ['_'.join(col).strip() for col in result.columns] # 输出：['transaction_amount_mean', 'transaction_amount_median', ...]

注意strip()——因为pandas有时会在连接符前后加空格，导致字段名含不可见字符，引发下游解析失败。这个细节我踩过三次坑，第一次是Power BI报“字段名无效”，第二次是Airflow调度任务因字段名含空格失败，第三次才意识到是join()引入的。

第二，缺失值处理。当某商户类别下只有1条记录时，median()会返回该值，但std()会返回NaN（样本标准差要求n≥2）。业务方往往不理解“为什么均值有数，标准差是空？”，你需要主动解释并提供兜底方案：

def safe_std(series): """业务兜底：当记录数<2时，返回0而非NaN，表示'无波动'""" if len(series) < 2: return 0.0 return series.std() result = df.groupby('merchant_category').agg({ 'transaction_amount': ['mean', safe_std] })

这里safe_std不是技术妥协，而是业务语义的显式声明：标准差为0，代表该商户交易金额完全稳定，这本身就是一个有价值的风控信号。

3.2 自定义函数的“三重校验”：让业务逻辑经得起审计

写一个自定义函数容易，让它在生产环境稳定运行难。我给自己定下铁律：任何自定义聚合函数必须通过三重校验。

校验一：输入鲁棒性。函数必须能处理空Series、全NaN Series、单值Series。比如加权平均函数：

def weighted_average(series): if series.empty or series.isna().all(): return np.nan if len(series) == 1: return float(series.iloc[0]) # 正常加权逻辑...

漏掉series.empty检查，当某商户无交易时，np.linspace()会报ZeroDivisionError，整个groupby中断。

校验二：业务边界。比如“高价值交易占比”函数：

def high_value_ratio(series, threshold=300): """业务规则：单笔≥300元为高价值，占比=高价值笔数/总笔数""" if len(series) == 0: return 0.0 high_count = (series >= threshold).sum() return round(high_count / len(series) * 100, 1)

这里threshold=300是硬编码，但实际业务中，这个阈值可能随季度调整。我的做法是：在函数签名中保留参数，但在调用层用配置文件管理：

# config.py HIGH_VALUE_THRESHOLD = { 'credit_card': 300, 'debit_card': 500, 'wallet': 100 } # 调用时 result = df.groupby('payment_type').agg({ 'amount': lambda x: high_value_ratio(x, HIGH_VALUE_THRESHOLD['credit_card']) })

这样，阈值变更只需改配置，无需动分析代码，符合“配置与代码分离”原则。

校验三：输出可解释性。函数返回的必须是业务能理解的值，而非技术中间态。比如计算“交易集中度”（赫芬达尔指数）：

def hhi_concentration(series): """业务定义：HHI指数 = 各笔交易金额占比的平方和，0-10000，越高越集中""" if len(series) == 0: return 0.0 total = series.sum() if total == 0: return 0.0 shares = (series / total) * 100 # 转换为百分比 return int((shares ** 2).sum()) # 返回整数，符合业务报告习惯

注意：int()强制转整数，因为业务方看报表时，没人关心HHI是2456.78还是2456；shares * 100转百分比，因为业务文档里写的“占比”都是百分数形式。这些细节，决定了你的分析是“能跑通”，还是“被业务方信任”。

3.3 滚动窗口的“三重陷阱”：为什么前两行总是NaN？

看这个输出：

date rolling_avg 2024-01-01 NaN 2024-01-02 NaN 2024-01-03 1243.33

新手常问：“怎么把前两个NaN填上？”但真正的问题是：你是否理解NaN在此处的业务含义？在风控场景中，NaN代表“数据不足，无法计算有效指标”，强行填充（如用fillna(method='ffill')）等于伪造信号。我在某银行做反洗钱模型时，就因用前向填充滚动均值，导致系统在新商户首笔交易后就给出“低风险”评级，而实际上该商户后续连续5笔均为可疑交易。正确做法是：

• 明确标注数据可用性：在结果中增加is_valid_window列，标记len(window_data) >= 3
• 业务兜底策略：当is_valid_window==False时，返回业务默认值（如“新商户，暂无历史参考”）
• 监控告警：对is_valid_window==False的记录数设置阈值，超限即告警，提示数据采集异常

另一个陷阱是窗口对齐方式。rolling(window=3)默认是右对齐（即窗口包含当前行及前2行），但有些场景需要左对齐（当前行及后2行）或居中对齐。pandas用closed参数控制：

# 右对齐（默认）：[t-2, t-1, t] df.rolling(window=3, closed='right') # 左对齐：[t, t+1, t+2] —— 需要未来数据，仅适用于批处理 df.rolling(window=3, closed='left')

在实时流处理中，左对齐是禁忌，因为它依赖未发生的事件。我见过最离谱的案例：某券商用左对齐滚动计算“未来3分钟涨跌幅”，结果模型在开盘前就给出预测，被监管认定为“利用未公开信息”。

3.4 Unstack重塑：从“程序员视角”到“业务方视角”的终极翻译

unstack()的价值，远不止于“把行变列”。它的本质是将数据结构与业务认知对齐。看这个例子：

result = df.groupby(['region','product'])['revenue'].mean().unstack()

输出：

product Gadget Widget region North 12000.0 15500.0 South 13750.0 18000.0

对程序员，这是个DataFrame；对销售总监，这是张决策地图：他一眼看出“Widget在南方卖得更好”，立刻决定把下季度的南方市场预算增加20%。如果不用unstack()，结果是：

region product North Gadget 12000.0 Widget 15500.0 South Gadget 13750.0 Widget 18000.0

这需要他横向扫描、纵向对比，效率极低。

但unstack()有三大实操雷区：

雷区一：缺失组合的处理。如果某区域没有某产品销售，unstack()默认留NaN。业务方看到“North-Gadget=NaN”会问：“是没卖？还是数据丢了？”我的标准做法是：

result = df.groupby(['region','product'])['revenue'].mean().unstack(fill_value=0)

fill_value=0明确传达“该组合无销售”，而非“数据缺失”。在某快消品公司，我们甚至用fill_value=-1表示“该渠道未铺货”，用颜色区分（0=灰色，-1=红色），让BI看板一目了然。

雷区二：多级索引的unstack层级。当groupby有3个维度时：

df.groupby(['region','product','channel'])['revenue'].mean() # 索引是三级：region→product→channel

unstack()默认unstack最内层（channel），但业务可能想看“region×channel”矩阵。此时用：

result = grouped.unstack(level='channel') # 指定层级 # 或 result = grouped.unstack(-1) # -1表示最内层，-2表示倒数第二层

雷区三：列名冲突。如果product列有重复值（如‘Widget’和‘widget’大小写不一致），unstack()会报错ValueError: Index contains duplicate entries。必须前置清洗：

df['product'] = df['product'].str.strip().str.title() # 统一为首字母大写

这个str.title()我加了三年，因为总有上游系统把“GADGET”传成“gadget”，导致月度报表准时失败。

4. 完整实操流程：从原始数据到高管简报的七步炼金术

4.1 步骤一：数据探查与维度清洗（耗时占全程40%，却决定成败）

别急着写groupby。先用这三行代码做“体检”：

# 1. 查看基础统计 print(df.describe(include='all')) # 2. 检查维度唯一性 print("Region唯一值:", df['region'].nunique(), "/", len(df)) print("Product唯一值:", df['product'].nunique(), "/", len(df)) # 3. 检查空值与异常值 print("\n空值分布:") print(df.isnull().sum()) print("\nAmount异常值（>99.9%分位数）:") print(df['amount'].quantile(0.999))

重点看两个数字：

• 如果df['region'].nunique()远小于len(df)，说明区域字段存在大量重复或脏数据（如“华东”“华东区”“EC”混用）
• 如果amount的99.9%分位数是5000，但最大值是500000，大概率有录入错误（如多输了一个0）

我的清洗清单（必做）：

• 维度标准化：用映射字典统一名称

  region_map = {'EC': 'East China', 'North': 'North China', 'BJ': 'Beijing'} df['region'] = df['region'].map(region_map).fillna(df['region']) # 未映射的保留原值

• 数值型字段业务校验：amount不能为负，fee不能超过amount*0.1

  df = df[(df['amount'] > 0) & (df['fee'] <= df['amount'] * 0.1)]

• 时间字段对齐：确保date是datetime类型，且无未来日期

  df['date'] = pd.to_datetime(df['date']) df = df[df['date'] <= pd.Timestamp.today()]

这一步我坚持“宁可慢，不可错”。在某保险公司的车险理赔分析中，因未清洗claim_type字段（含“车损”“车损险”“车辆损失”三种写法），导致groupby('claim_type')算出三个不同的“车损”指标，误导了精算模型，返工两周。

4.2 步骤二：构建核心聚合（七种武器，按需组合）

基于清洗后的数据，按业务需求选择聚合模式。我总结为“七种武器”：

关键技巧：当需要同时用滚动和分组时，必须先sort再groupby：

# ✅ 正确：先按时间排序，再分组滚动 df_sorted = df.sort_values(['customer_id','date']) result = df_sorted.groupby('customer_id')['amount'].rolling(7).mean() # ❌ 错误：groupby后rolling，窗口不按时间序 df.groupby('customer_id')['amount'].rolling(7).mean() # 结果随机！

这个错误我见太多次，因为pandas的rolling()在groupby后默认不保证时间顺序，必须显式sort_values()。

4.3 步骤三：结果扁平化与业务适配（让技术输出变成业务语言）

聚合结果往往是MultiIndex或嵌套结构，必须转化为业务方能直接使用的格式。我的标准流水线：

1. 列名标准化：

def flatten_columns(df): """将MultiIndex列转为下划线连接的扁平列名""" if isinstance(df.columns, pd.MultiIndex): df.columns = ['_'.join(col).strip() for col in df.columns] return df result = flatten_columns(result) # 输出：['region_product_sum', 'region_product_mean', ...]

2. 添加业务元数据：

# 在结果中加入计算时间戳和版本号，便于审计 result['calculation_date'] = pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S') result['analysis_version'] = 'v2.1.0' # 对应Git Tag

3. 生成业务摘要（高管简报核心）：

def generate_exec_summary(df_grouped): """生成高管可读的摘要：TOP3、同比、环比较""" summary = {} # TOP3区域 top3_region = df_grouped['revenue'].sum().nlargest(3) summary['top3_regions'] = top3_region.index.tolist() # 同比增长（假设数据含year_month列） if 'year_month' in df_grouped.index.names: current = df_grouped.xs('2024-06', level='year_month')['revenue'].sum() last_year = df_grouped.xs('2023-06', level='year_month')['revenue'].sum() summary['yoy_growth'] = round((current - last_year) / last_year * 100, 1) return summary exec_summary = generate_exec_summary(result) print(f"【高管简报】TOP3区域：{exec_summary['top3_regions']}；同比增速：{exec_summary['yoy_growth']}%")

4.4 步骤四：可视化与交付（不是画图，而是讲故事）

聚合结果不等于洞察。我交付的从来不是DataFrame，而是带注释的叙事图表。用matplotlib/seaborn时，必加三要素：

要素一：业务标注。在柱状图顶部加文字说明：

ax.bar(range(len(data)), data['revenue']) # 在最高柱上加标注 max_idx = data['revenue'].idxmax() ax.text(max_idx, data['revenue'].max() * 1.02, f"最高：{data.loc[max_idx, 'region']} ({data.loc[max_idx, 'revenue']:.0f}万)", ha='center', fontweight='bold')

要素二：基准线。所有图表必须有业务基准：

# 添加行业平均线 ax.axhline(y=industry_avg, color='red', linestyle='--', label=f'行业均值({industry_avg:.0f}万)')

要素三：数据来源水印：

fig.text(0.02, 0.02, f'数据源：{source_system} | 更新时间：{last_update}', fontsize=8, color='gray', style='italic')

在某基金公司，我们甚至把“数据延迟说明”写进图表：'注：本数据T+1更新，2024-06-15数据截至2024-06-14 23:59'，避免业务方拿实时数据质疑。

4.5 步骤五：自动化与监控（让分析从“手工活”变成“自来水”）

生产环境必须监控聚合质量。我在Airflow中部署的检查项：

• 数据完整性：每日检查df.shape[0]是否较前日下降>10%
• 维度漂移：检查df['region'].nunique()是否新增未知值（如出现“Antarctica”）
• 指标合理性：revenue.mean()是否在历史3σ范围内
• 执行时效：任务是否在SLA（如30分钟）内完成

一旦告警，自动触发：

1. 发送企业微信消息给负责人
2. 保存当日原始数据快照到隔离区
3. 运行诊断脚本，输出问题定位（如“region字段新增值：'XX Special Zone'，未在主数据表中”）

这套机制让我们在某电商大促期间，提前2小时发现“物流区域编码”字段异常，避免了千万级GMV误报。

5. 常见问题与排查技巧实录：那些年我踩过的坑

5.1 问题一：groupby().agg()返回空DataFrame，但数据明明不为空

现象：df.groupby('region').agg({'sales':'sum'})返回空，df['region'].isnull().sum()为0。

排查路径：

1. 检查分组键类型：df['region'].dtype是否为object？如果是，检查是否有不可见字符：

   print(repr(df['region'].iloc[0])) # 显示'\x00Beijing'，说明有空字符 df['region'] = df['region'].str.strip('\x00') # 清洗

2. 检查分组键唯一性：df['region'].nunique()是否为0？可能是全NaN被dropna=True过滤了（groupby默认dropna=True）：

   df.groupby('region', dropna=False).agg({'sales':'sum'}) # 强制保留NaN组

3. 检查列名是否存在：'sales'列名是否拼写错误？用df.columns.tolist()确认。

我的避坑技巧：在所有groupby前加断言：

assert 'region' in df.columns, "缺失region列" assert not df['region'].isnull().all(), "region列全为空" assert df.shape[0] > 0, "输入数据为空"

5.2 问题二：滚动计算结果全是NaN，或数量不对

现象：df.rolling(3).mean()返回全NaN，或行数少于预期。

根本原因：rolling()要求数据按时间/序号严格排序，且窗口内必须有足够数据。

排查三步法：

1. 验证排序：

   # 检查是否已排序 is_sorted = df['date'].is_monotonic_increasing if not is_sorted: df = df.sort_values('date').reset_index(drop=True)

2. 检查窗口大小：rolling(3)需要至少3行，用min_periods放宽：

   # 允许最小2行参与计算 df.rolling(3, min_periods=2).mean()

3. 检查数据类型：rolling()只对数值列有效，date列是datetime会报错：

   # 只对数值列滚动 numeric_cols = df.select_dtypes(include=[np.number]).columns df[numeric_cols] = df[numeric_cols].rolling(3).mean()

血泪教训：在某支付公司，因未检查min_periods，滚动计算在月初数据不足时全为NaN，导致风控模型失效3天。现在我的标准模板是：

def safe_rolling(df, window, min_periods=2, **kwargs): """安全滚动：自动处理排序、类型、缺失""" df_sorted = df.sort_values('date') if 'date' in df.columns else df numeric_df = df_sorted.select_dtypes(include=[np.number]) return numeric_df.rolling(window, min_periods=min_periods).mean()

5.3 问题三：unstack()报错Index contains duplicate entries

现象：df.groupby(['A','B'])['C'].sum().unstack()报错。

原因：A和B的组合不唯一，如('North','Widget')出现两次。

解决方案：

1. 先聚合去重：

   # 用agg确保每组唯一 grouped = df.groupby(['A','B'])['C'].sum() # sum()天然去重 result = grouped.unstack()

2. 检查重复组合：

   duplicates = df.duplicated(subset=['A','B'], keep=False) print("重复组合示例：", df[duplicates].head())

3. 业务决策：是删除重复（df.drop_duplicates(['A','B'])），还是合并（df.groupby(['A','B']).first()）？这取决于业务规则。

我的经验：90%的重复是数据质量问题，必须溯源。我在某零售系统发现，重复源于POS机网络抖动导致同一笔交易上传两次，最终推动IT部门在网关层加幂等控制。

5.4 问题四：自定义函数在apply()中报错，但单独调用正常

现象：df.groupby('A')['B'].apply(my_func)报错，但my_func(df['B'])正常。

真相：apply()传入的是Series子集，其index是原始index，可能不连续。

典型错误：

def bad_func(series): return series.iloc[0] + series.iloc[1] # 当组内只有1条时，iloc[1]越界

修复方案：

def good_func(series): if len(series) < 2: return series.iloc[0] if len(series) > 0 else np.nan return series.iloc[0] + series.iloc[1] # 更健壮：用values避开index问题 def robust_func