1. 项目概述为什么多维聚合不是“加个groupby”就能搞定的事我在银行数据平台组干了八年从最早用SQL写几十行嵌套子查询做客户分层到现在每天在Jupyter里调试pandas的agg链式调用踩过的坑比写的代码还多。今天这篇讲的“多维聚合”绝不是教你怎么把df.groupby(col).sum()敲得更顺——那是实习生第一天就能学会的。真正卡住业务分析师、拖慢风控模型上线、让报表系统半夜报警的永远是那些看似简单、实则暗藏玄机的聚合需求比如“请输出每个客户在餐饮、零售、旅游三大类商户的月均交易额、中位数、最大单笔、最小单笔、标准差再叠加过去30天滚动均值最后按地区维度横向展开成Excel可读格式”。你试试看一行groupby能搞定吗不能。它需要你像搭积木一样把五种不同逻辑的聚合模块严丝合缝地拼在一起还要保证每一块都扛得住生产环境每秒上万条的交易流冲击。核心关键词就三个多维聚合、滚动计算、业务可解释性。这不是技术炫技而是银行每天真实发生的场景——信贷经理要看某客户是否突然在高风险商户如珠宝、赌场集中消费反洗钱系统要识别“小额分散转入、大额集中转出”的可疑模式运营团队要判断某区域新上线的优惠活动到底拉动了哪类客群的哪类消费。这些需求背后是数据必须同时回答空间地区/产品、时间滚动/累计、统计维度中心趋势离散程度极值三重问题。我见过太多团队把这类需求硬塞进一个SQL视图里结果报表跑一次要17分钟运维同事天天盯着数据库CPU告警。而用对pandas的聚合组合技同样逻辑本地笔记本上2秒出结果导出到Spark集群也只需改两行配置。关键不在工具在于你是否理解每种聚合背后的业务语义约束和计算代价边界。比如为什么“滚动均值”必须用rolling().mean()而不是手动切片循环因为前者底层调用的是NumPy的向量化C函数后者在Python层循环10万次性能差两个数量级。再比如为什么unstack()之后一定要用fill_value0因为没成交的“客户×品类”组合在原始数据里根本不存在不填0的话下游BI工具会直接报错“列缺失”。这些细节文档里不会写但线上故障单上全是它们的名字。这篇文章面向三类人一是刚转行做金融数据分析的新人别再被“groupby基础语法”教程带偏真实业务里80%的聚合需求都超纲二是有SQL经验但刚接触pandas的数据工程师你需要知道pandas的聚合不是SQL的平移而是用Python思维重构数据流三是已经会写agg({col:[mean,std]})但总被业务方追问“这个中位数怎么算的为什么和我们Excel不一样”的分析师——你缺的不是代码是能把技术操作翻译成业务语言的能力。接下来所有内容都来自我亲手交付的7个银行级数据管道包括某国有大行信用卡中心的实时欺诈监控看板、某股份制银行跨境支付的风险敞口日报。没有玩具数据集只有真实字段名、真实阈值、真实报错日志。现在我们拆开这台“多维聚合引擎”看看每个齿轮怎么咬合。2. 核心设计思路五种聚合模式的业务定位与技术选型逻辑很多人一上来就猛敲代码结果写到一半发现滚动窗口算出来全是NaN多级索引展不开自定义函数在空组时报错。根本原因在于没想清楚——每种聚合模式解决的是完全不同的业务问题强行混用只会制造灾难。我画过一张贴在工位上的决策树今天直接分享给你。这张图不是理论推演而是我处理过237个生产需求后总结的血泪经验。2.1 多列多函数聚合解决“并行指标对比”问题典型业务场景财务部要同时看“交易金额的平均值反映常规消费水平”和“中位数规避大额异常单干扰”而运营部要盯“手续费的最小值和最大值监控渠道费率波动”。注意这里的关键是指标间无计算依赖——中位数不需要用到手续费数据手续费极值也不依赖交易金额分布。所以必须用agg({col1:[func1,func2], col2:[func3,func4]})这种字典映射方式。如果错误地写成两次groupby再merge不仅内存翻倍更致命的是当某客户在餐饮类无交易时第一次groupby会丢掉该客户第二次merge后该客户在最终结果里彻底消失而业务方需要的是“零值占位”。提示agg()返回的MultiIndex列结构是双刃剑。外层是原始列名如transaction_amount内层是函数名如mean。这种结构在后续做reset_index()或to_dict()时极易出错。我的经验是只要结果要导出给BI或Excel立刻执行result.columns [_.join(col).strip() for col in result.columns.values]扁平化列名。比如(transaction_amount, mean)变成transaction_amount_mean。别嫌麻烦这是避免下游同事半夜打电话问“为什么列名里有括号”的唯一方法。2.2 自定义聚合函数解决“业务规则编码”问题典型业务场景风控部门要求计算“交易金额范围max-min”但这个值不能直接用内置np.ptp()因为业务定义里“范围”必须排除单笔交易的客户即只对交易次数≥2的客户计算。这时候lambda x: x.max()-x.min()会报错——当某客户只有一笔交易时x.min()和x.max()是同一个值但业务上认为“单笔无范围概念”。正确解法是写命名函数def business_range(series): if len(series) 2: return np.nan # 明确返回NaN而非0避免被误读为“范围为0” return series.max() - series.min()为什么必须用命名函数而非lambda两点第一lambda无法加docstring半年后你自己都忘了这个range是按什么业务逻辑算的第二pandas对命名函数的错误堆栈更友好——当business_range内部报错时日志里会清晰显示File risk_metrics.py, line 45, in business_range而lambda报错只显示lambda排查成本翻倍。2.3 滚动窗口聚合解决“时间动态性”问题典型业务场景反洗钱系统检测“近期消费突增”。不能只看“本月总额比上月高50%”因为月初可能还没数据。必须用“过去7天滚动均值 vs 历史均值”。这里有个致命陷阱rolling(window7).mean()默认要求窗口内7个非空值但实际交易数据有节假日、周末休市连续7天可能只有3笔交易。如果按默认设置结果全是NaN。正确做法是加参数min_periods3表示只要有3个值就计算均值。这个3不是拍脑袋定的而是根据业务SLA风控规则要求“至少覆盖3天交易行为才触发预警”。注意滚动窗口必须在set_index(date)后进行且date列必须是datetime64类型。我曾遇到一个坑原始数据里date是字符串2024-01-01rolling()会把它当普通索引结果算出的“滚动均值”其实是按行号滚动而非按日期滚动修复只需一行df[date] pd.to_datetime(df[date])。2.4 扩展窗口聚合解决“累积效应”问题典型业务场景客户生命周期价值CLV计算。不是看“当前月消费”而是“从开户至今的累计消费”。expanding().sum()完美匹配但要注意其与cumsum()的本质区别expanding().sum()是窗口函数支持min_periods参数如expanding(min_periods1)而cumsum()是纯累加遇到第一个NaN就全盘崩溃。在真实数据中客户首笔交易可能因系统延迟晚录1小时导致首行amount为NaN——用cumsum()整列变NaN用expanding().sum()则从第二个有效值开始累加。2.5 多级分组unstack解决“业务视角转换”问题典型业务场景销售总监要一张表横轴是产品线Widget/Gadget纵轴是大区North/South单元格是“平均客单价”。这本质是三维数据客户×产品×地区降维到二维矩阵。groupby([region,product])[revenue].mean().unstack()是标准解法但unstack()默认用np.nan填充缺失组合。而业务方要的是“0”——没卖过的产品显示0元而非空白。所以必须加fill_value0。更深层的问题是如果某大区某产品销量为0unstack()后该单元格是0但业务上需要区分“0销量”和“数据未采集”。这时我会额外加一列is_data_available用groupby().size()统计每组记录数再unstack(fill_value0)这样既能显示0销量又能标记数据完整性。这五种模式不是孤立的而是构成一个聚合流水线先用多列聚合产出基础指标再用自定义函数注入业务规则接着用滚动/扩展窗口添加时间维度最后用unstack()适配业务汇报格式。任何跳步都会导致结果失真。比如先unstack()再滚动计算不行unstack()后数据结构已变rolling()无法识别时间序列。必须严格遵循“分组→聚合→时间窗→重塑”的顺序。这个顺序是我用3个通宵回滚生产事故后刻进DNA的。3. 实操细节解析从代码到业务落地的12个关键控制点光知道五种模式还不够真实生产环境里90%的失败源于细节失控。下面这12个控制点每一个都对应我亲手处理过的线上故障。别跳过它们比语法更重要。3.1 控制点1空值处理——不是填0那么简单业务数据里空值NaN无处不在新客户首笔交易手续费未计费、跨境交易汇率字段为空、测试环境模拟数据故意留空。agg()对NaN的默认行为是静默跳过这很危险。比如计算mean()时10笔交易有2笔NaNagg()只对8笔有效值求均值但业务上可能要求“NaN视为0”或“NaN视为缺失需告警”。解决方案是预处理# 方案A全局填充谨慎使用 df[fee] df[fee].fillna(0) # 仅适用于明确业务含义为0的字段 # 方案B聚合时指定skipnaFalse强制报错 df.groupby(category)[fee].agg(lambda x: x.mean(skipnaFalse)) # 任一NaN则整个组返回NaN触发监控告警 # 方案C业务规则填充推荐 df[fee] df.apply( lambda row: 0 if pd.isna(row[fee]) and row[amount] 1000 else row[fee], axis1 )最后一行是真实案例手续费为空且交易额超1000元时按0填充因系统规则大额交易免手续费否则保留原值。这比fillna(0)严谨得多。3.2 控制点2数据类型陷阱——int64和float64的隐式转换pandas的agg()在混合类型列上会自动升格为float64。比如transaction_count是int64agg({count:[sum,mean]})后sum仍是整数但mean变成浮点数。下游系统如Oracle数据库导入时float64列可能被截断小数位导致“平均交易笔数3.0000000000000004”。修复方法聚合后显式转换result df.groupby(category).agg({count: [sum, mean]}) result[(count, mean)] result[(count, mean)].round(2).astype(float) # 保留2位小数3.3 控制点3性能杀手——避免在agg中调用apply新手常写df.groupby(customer_id).apply(lambda x: x[amount].rolling(7).mean())。这是灾难apply()是逐组Python循环10万客户就要循环10万次而rolling()本身是向量化操作。正确姿势是先sort_values(date).set_index(date)再groupby(customer_id)[amount].rolling(7).mean()。实测10万客户数据前者耗时47秒后者耗时1.2秒。3.4 控制点4时序对齐——滚动窗口的索引必须是单调递增rolling()要求索引严格递增。但真实交易数据常有乱序系统延迟导致后发生的交易先入库或批量补录数据时间戳相同。df.sort_values(date).set_index(date)后若存在重复日期rolling()会报错。解决方案# 步骤1去重并保留最新记录 df df.sort_values(ingestion_time).drop_duplicates(subset[date,customer_id], keeplast) # 步骤2对重复日期添加微秒偏移 df[date] df.groupby(date).cumcount().apply(lambda x: pd.Timedelta(x, unitus)) df[date]3.5 控制点5内存优化——大表聚合前必做列类型压缩处理千万级交易表时object类型列如category占内存极大。category列实际只有4个值Dining/Retail等却以object存储内存占用是category类型的5倍。必须在groupby前压缩df[category] df[category].astype(category) # 内存减少80% df[customer_id] df[customer_id].astype(category) # 同理3.6 控制点6业务一致性——确保rolling/expanding的分组键唯一groupby(customer_id)[amount].rolling(7).mean()要求每个customer_id组内date索引唯一。但若同一客户同一天有多笔交易rolling()会把它们当同一时间点处理导致窗口计算错误。必须先按时间聚合# 先按客户日期聚合日度指标 daily_df df.groupby([customer_id, date])[amount].sum().reset_index() # 再对日度数据做滚动 daily_df daily_df.sort_values([customer_id,date]).set_index(date) daily_df[rolling_7day] daily_df.groupby(customer_id)[amount].rolling(7D).mean()注意这里用7D7天而非window77D是时间窗口自动处理非交易日window7是行窗口强制取7行可能跨月。3.7 控制点7unstack的维度安全——永远指定fill_valueunstack()不加fill_value缺失组合会是NaN。而NaN在数值计算中具传染性NaN 100 NaN。业务方看到“North区Gadget产品销售额NaN”会以为系统故障。必须result df.groupby([region,product])[revenue].mean().unstack(fill_value0) # 进一步加固标记数据状态 result[data_status] np.where(result.eq(0).all(axis1), no_data, partial_data)3.8 控制点8自定义函数的异常防御——空组、全NaN组、单值组自定义函数最怕边缘情况。df.groupby(category).apply(risk_metrics)中若某category组为空如新上线品类暂无交易apply()直接报错。必须在函数内兜底def risk_metrics(series): if len(series) 0: # 空组 return pd.Series({high_value_count: 0, high_value_pct: 0, regular_avg: 0}) if series.isna().all(): # 全NaN组 return pd.Series({high_value_count: 0, high_value_pct: 0, regular_avg: np.nan}) # ...正常逻辑3.9 控制点9结果验证——用SQL交叉验证关键指标再自信的pandas代码也要用SQL验证。我坚持的流程对任意聚合结果抽样3个客户用SQL重算-- pandas算出C001的rolling_7day_avg 264.087 SELECT AVG(amount) OVER (PARTITION BY customer_id ORDER BY date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) as rolling_7day FROM transactions WHERE customer_id C001 AND date 2024-01-07;若结果不一致一定是pandas的sort_values或set_index逻辑有偏差。3.10 控制点10可审计性——在结果中嵌入计算元数据业务方常问“这个中位数是按什么规则算的是不是用了插值法”pandas的median()默认用linear插值但业务可能要求lower取下界。解决方案在结果DataFrame中加元数据列result df.groupby(category).agg({ amount: [(median_lower, lambda x: x.median(lower)), (std, std)] }) result.attrs[calculation_rules] { median_lower: pandas median with methodlower, std: sample standard deviation (ddof1) }3.11 控制点11增量更新——避免全量重算生产系统不能每次跑全量。对滚动窗口我用last_computed_date记录上次计算截止日只处理新增数据last_date get_last_computed_date() # 从元数据表查 new_data df[df[date] last_date] # 只对new_data做rolling再与历史结果concat3.12 控制点12监控告警——为聚合结果设置业务阈值agg()结果不是终点而是监控起点。例如transaction_range超过5000元应告警可能为测试数据或异常商户range_result df.groupby(category)[amount].agg(business_range) if (range_result 5000).any(): send_alert(fHigh range detected: {range_result[range_result 5000]})这12个控制点每一个都来自血泪教训。记住在金融数据领域正确的结果不叫成功可验证、可追溯、可监控的结果才叫交付。下一节我们进入完整实操用一个真实的银行信用卡分析案例把这12个点全部串起来。4. 完整实操构建银行级信用卡客户分析流水线现在我们把前面所有理论焊接到一个真实场景里为某城商行信用卡中心构建T1客户行为分析流水线。需求来自风控总监的邮件“需每日早9点前输出《高价值客户异动日报》包含1各客户近7天滚动消费均值2消费金额范围max-min3高价值交易300元占比4按地区×品类的交叉分析表5每位客户累计消费总额。所有指标需支持下钻到单客户明细。”——这就是典型的多维聚合生产需求。下面是我的完整实现含所有避坑细节。4.1 数据准备与清洗从原始交易表到分析就绪原始数据是MySQL表credit_card_transactions每日增量同步到数据湖。第一步不是写groupby而是用SQL做前置清洗pandas不适合做复杂ETL-- 创建临时清洗表 CREATE TABLE temp_cleaned AS SELECT customer_id, DATE(transaction_time) as date, -- 强制转日期解决时间戳精度问题 CASE WHEN merchant_category IN (Jewelry,Casino,Luxury Goods) THEN High_Risk WHEN merchant_category IN (Groceries,Pharmacy) THEN Essential ELSE merchant_category END as category, ROUND(amount, 2) as amount, -- 统一金额精度 CASE WHEN fee IS NULL THEN 0 ELSE ROUND(fee, 2) END as fee, transaction_time FROM credit_card_transactions WHERE DATE(transaction_time) DATE_SUB(CURDATE(), INTERVAL 90 DAY) -- 只取90天控内存 AND amount 0; -- 排除退款、冲正等负值然后用pandas读取清洗后数据# 关键指定列类型省80%内存 dtype { customer_id: category, category: category, amount: float32, # float64太重float32精度足够 fee: float32 } df pd.read_sql(SELECT * FROM temp_cleaned, con, dtypedtype) # 验证数据质量 print(f数据时间范围: {df[date].min()} 到 {df[date].max()}) print(f客户数: {df[customer_id].nunique()}, 品类数: {df[category].nunique()}) print(f空值检查:\n{df.isna().sum()})实操心得永远不要相信上游数据。我遇到过最诡异的Bug某批次数据里customer_id是字符串C001 末尾有空格导致同一客户被分成两组。修复只需一行df[customer_id] df[customer_id].str.strip()。这个strip()我加在每一版代码的第二行。4.2 构建核心聚合流水线五步不可逆顺序步骤1多列多函数聚合——产出基础指标池# 业务要求每个客户每个品类计算均值、中位数、计数、手续费极值 base_agg df.groupby([customer_id, category]).agg({ amount: [mean, median, count, lambda x: x.max() - x.min()], # 业务范围 fee: [min, max, sum] }) # 扁平化列名加业务前缀 base_agg.columns [fbase_{col[0]}_{col[1]} for col in base_agg.columns] base_agg base_agg.reset_index() # 关键控制点处理空值组 base_agg[base_amount_lambda] base_agg[base_amount_lambda].fillna(0) # 范围为空0步骤2自定义函数注入——计算高价值交易占比def high_value_ratio(series): 计算300元交易占比空组返回0 if len(series) 0: return 0.0 high_val_count (series 300).sum() return round((high_val_count / len(series)) * 100, 1) # 保留1位小数 # 应用到每个客户不分品类因业务要求是客户级指标 hv_ratio df.groupby(customer_id)[amount].apply(high_value_ratio) hv_ratio hv_ratio.rename(hv_ratio_pct).reset_index()步骤3滚动窗口计算——7天滚动均值# 先按客户日期聚合日度消费 daily_customer df.groupby([customer_id, date])[amount].sum().reset_index() # 排序并设索引关键 daily_customer daily_customer.sort_values([customer_id, date]) daily_customer daily_customer.set_index(date) # 计算滚动均值用7D时间窗口非window7 daily_customer[rolling_7day] ( daily_customer.groupby(customer_id)[amount] .rolling(7D, min_periods3) # 至少3天有数据才计算 .mean() .reset_index(level0, dropTrue) ) # 关键将滚动结果merge回原表避免索引错位 rolling_result daily_customer[[customer_id, rolling_7day]].reset_index()步骤4扩展窗口计算——累计消费总额# 同样基于日度数据 daily_customer[cumulative_spend] ( daily_customer.groupby(customer_id)[amount] .expanding(min_periods1) # 从第一天开始累加 .sum() .reset_index(level0, dropTrue) ) cumulative_result daily_customer[[customer_id, cumulative_spend]].reset_index()步骤5多级分组unstack——生成交叉分析表# 业务要求客户×品类的平均消费额矩阵 crosstab df.groupby([customer_id, category])[amount].mean().unstack( fill_value0 ).round(2) # 金额保留2位小数 # 为下游BI准备重命名列加前缀 crosstab.columns [fcrosstab_{col} for col in crosstab.columns] crosstab crosstab.reset_index()4.3 结果整合与业务封装从技术表到业务报告现在有5张中间表base_agg、hv_ratio、rolling_result、cumulative_result、crosstab。下一步是按业务逻辑合并而非简单pd.merge()# 主表以base_agg为基础因它含客户品类粒度 final_report base_agg.copy() # 合并客户级指标hv_ratio, cumulative_spend final_report final_report.merge(hv_ratio, oncustomer_id, howleft) final_report final_report.merge( cumulative_result.drop(date, axis1), # 去掉date列避免冲突 oncustomer_id, howleft ) # 合并滚动结果注意rolling_result有date列需取最新日期 latest_rolling rolling_result.loc[rolling_result.groupby(customer_id)[date].idxmax()] final_report final_report.merge( latest_rolling[[customer_id, rolling_7day]], oncustomer_id, howleft ) # 合并交叉表需explode展开因crosstab是宽表 # 先将crosstab转为长表再merge crosstab_long crosstab.melt( id_varscustomer_id, var_namecategory_col, value_namecrosstab_avg_amount ) crosstab_long[category] crosstab_long[category_col].str.replace(crosstab_, ) final_report final_report.merge( crosstab_long, on[customer_id, category], howleft ) # 最终清理重命名、排序、去重 final_report final_report.rename(columns{ base_amount_mean: avg_amount, base_amount_median: median_amount, base_amount_count: transaction_count, base_amount_lambda: amount_range, base_fee_min: min_fee, base_fee_max: max_fee, base_fee_sum: total_fee, rolling_7day: rolling_7day_avg, cumulative_spend: total_cumulative_spend, crosstab_avg_amount: category_avg_amount }) # 按客户、品类排序便于阅读 final_report final_report.sort_values([customer_id, category]).reset_index(dropTrue)4.4 生产部署与监控让流水线自己说话代码写完只是开始。生产环境必须有“自我诊断”能力# 1. 数据质量监控 quality_check { total_records: len(final_report), null_rate: final_report.isna().sum().sum() / final_report.size, zero_range_count: (final_report[amount_range] 0).sum(), hv_ratio_outlier: ((final_report[hv_ratio_pct] 0) | (final_report[hv_ratio_pct] 100)).sum() } if quality_check[null_rate] 0.01: send_alert(f空值率超标: {quality_check[null_rate]:.2%}) # 2. 业务阈值告警 high_risk_customers final_report[ (final_report[amount_range] 5000) (final_report[hv_ratio_pct] 70) ][[customer_id, category, amount_range, hv_ratio_pct]] if not high_risk_customers.empty: send_alert(f高风险客户: {len(high_risk_customers)}人, high_risk_customers) # 3. 性能监控 import time start time.time() # ...执行聚合... end time.time() if end - start 300: # 超5分钟告警 send_alert(f聚合超时: {end-start:.1f}秒) # 4. 输出到业务系统 final_report.to_parquet(s3://bank-data/reports/daily_customer_analytics.parquet, indexFalse, compressionsnappy) # 同时生成Excel供人工核查 final_report.head(1000).to_excel(daily_report_sample.xlsx, indexFalse)这个流水线已在生产运行11个月日均处理2300万条交易平均耗时42秒。它的核心不是多酷炫的代码而是把12个控制点像螺丝一样拧紧在每个环节。比如rolling(7D)里的min_periods3就是为应对国庆长假7天无交易crosstab的fill_value0让BI工具无需额外处理空值hv_ratio函数里的round(...,1)确保业务方看到的百分比是“45.0%”而非“45.00000000000001%”。真正的专业藏在这些毫米级的细节里。5. 常见问题与实战排障那些让你凌晨三点爬起来的Bug再完美的设计也会在生产环境撞上意想不到的墙。我把这些年最常遇到的7类问题连同根因和速查方案整理成这张表。这不是理论清单而是我电脑里troubleshooting.md的实时备份。问题现象根本原因速查命令解决方案我的血泪教训滚动均值全为NaNdate列不是datetime64或索引未排序df[date].dtype,df.index.is_monotonic_increasingdf[date] pd.to_datetime(df[date]);df df.sort_values(date).set_index(date)曾因date是字符串rolling()按字典序滚动算出的“7天均值”其实是按2024-01-01到2024-01-07的字符串排序而非日期unstack后列名带括号BI工具报错MultiIndex列未扁平化result.columnsresult.columns [_.join(col).strip() for col in result.columns.values]某次导出到Tableau因列名(amount,mean)含括号Tableau直接拒绝连接客户会议前2小时紧急修复。自定义函数报错“Series object is not callable”函数名与pandas内置方法重名如sumdir(pd.Series)重命名函数如my_sum避免用sum、mean、count等新人常写def sum(x): return x.sum()结果agg({col: sum})调用时pandas把sum当成了内置方法而非你的函数。内存爆满MemoryErrorobject类型列未压缩或未用float32df.memory_usage(deepTrue)df[col] df[col].astype(category);df[col] df[col].astype(float32)处理1亿行数据时customer_id从object转category内存从12GB降到2.3GB。rolling结果行数变少min_periods默认为window值导致首window-1行为空len(rolling_result) len(df)显式设置min_periods1或用fillna(methodffill)风控要求“首日也需有滚动值”必须fillna(methodffill)但要注意这会让首日值等于当日值业务上是否可接受groupby后结果顺序混乱groupby默认不保持原始顺序df.groupby(...).apply(...).reset_index()加参数sortFalse:df.groupby(col, sortFalse).agg(...)某次报表客户排名按groupby后顺序输出结果业务方说“为什么VIP客户排在最后”因groupby按字母序排了customer_id。空组导致agg报错agg()遇到空组
1. 项目概述为什么多维聚合不是“加个groupby”就能搞定的事我在银行数据平台组干了八年从最早用SQL写几十行嵌套子查询做客户分层到现在每天在Jupyter里调试pandas的agg链式调用踩过的坑比写的代码还多。今天这篇讲的“多维聚合”绝不是教你怎么把df.groupby(col).sum()敲得更顺——那是实习生第一天就能学会的。真正卡住业务分析师、拖慢风控模型上线、让报表系统半夜报警的永远是那些看似简单、实则暗藏玄机的聚合需求比如“请输出每个客户在餐饮、零售、旅游三大类商户的月均交易额、中位数、最大单笔、最小单笔、标准差再叠加过去30天滚动均值最后按地区维度横向展开成Excel可读格式”。你试试看一行groupby能搞定吗不能。它需要你像搭积木一样把五种不同逻辑的聚合模块严丝合缝地拼在一起还要保证每一块都扛得住生产环境每秒上万条的交易流冲击。核心关键词就三个多维聚合、滚动计算、业务可解释性。这不是技术炫技而是银行每天真实发生的场景——信贷经理要看某客户是否突然在高风险商户如珠宝、赌场集中消费反洗钱系统要识别“小额分散转入、大额集中转出”的可疑模式运营团队要判断某区域新上线的优惠活动到底拉动了哪类客群的哪类消费。这些需求背后是数据必须同时回答空间地区/产品、时间滚动/累计、统计维度中心趋势离散程度极值三重问题。我见过太多团队把这类需求硬塞进一个SQL视图里结果报表跑一次要17分钟运维同事天天盯着数据库CPU告警。而用对pandas的聚合组合技同样逻辑本地笔记本上2秒出结果导出到Spark集群也只需改两行配置。关键不在工具在于你是否理解每种聚合背后的业务语义约束和计算代价边界。比如为什么“滚动均值”必须用rolling().mean()而不是手动切片循环因为前者底层调用的是NumPy的向量化C函数后者在Python层循环10万次性能差两个数量级。再比如为什么unstack()之后一定要用fill_value0因为没成交的“客户×品类”组合在原始数据里根本不存在不填0的话下游BI工具会直接报错“列缺失”。这些细节文档里不会写但线上故障单上全是它们的名字。这篇文章面向三类人一是刚转行做金融数据分析的新人别再被“groupby基础语法”教程带偏真实业务里80%的聚合需求都超纲二是有SQL经验但刚接触pandas的数据工程师你需要知道pandas的聚合不是SQL的平移而是用Python思维重构数据流三是已经会写agg({col:[mean,std]})但总被业务方追问“这个中位数怎么算的为什么和我们Excel不一样”的分析师——你缺的不是代码是能把技术操作翻译成业务语言的能力。接下来所有内容都来自我亲手交付的7个银行级数据管道包括某国有大行信用卡中心的实时欺诈监控看板、某股份制银行跨境支付的风险敞口日报。没有玩具数据集只有真实字段名、真实阈值、真实报错日志。现在我们拆开这台“多维聚合引擎”看看每个齿轮怎么咬合。2. 核心设计思路五种聚合模式的业务定位与技术选型逻辑很多人一上来就猛敲代码结果写到一半发现滚动窗口算出来全是NaN多级索引展不开自定义函数在空组时报错。根本原因在于没想清楚——每种聚合模式解决的是完全不同的业务问题强行混用只会制造灾难。我画过一张贴在工位上的决策树今天直接分享给你。这张图不是理论推演而是我处理过237个生产需求后总结的血泪经验。2.1 多列多函数聚合解决“并行指标对比”问题典型业务场景财务部要同时看“交易金额的平均值反映常规消费水平”和“中位数规避大额异常单干扰”而运营部要盯“手续费的最小值和最大值监控渠道费率波动”。注意这里的关键是指标间无计算依赖——中位数不需要用到手续费数据手续费极值也不依赖交易金额分布。所以必须用agg({col1:[func1,func2], col2:[func3,func4]})这种字典映射方式。如果错误地写成两次groupby再merge不仅内存翻倍更致命的是当某客户在餐饮类无交易时第一次groupby会丢掉该客户第二次merge后该客户在最终结果里彻底消失而业务方需要的是“零值占位”。提示agg()返回的MultiIndex列结构是双刃剑。外层是原始列名如transaction_amount内层是函数名如mean。这种结构在后续做reset_index()或to_dict()时极易出错。我的经验是只要结果要导出给BI或Excel立刻执行result.columns [_.join(col).strip() for col in result.columns.values]扁平化列名。比如(transaction_amount, mean)变成transaction_amount_mean。别嫌麻烦这是避免下游同事半夜打电话问“为什么列名里有括号”的唯一方法。2.2 自定义聚合函数解决“业务规则编码”问题典型业务场景风控部门要求计算“交易金额范围max-min”但这个值不能直接用内置np.ptp()因为业务定义里“范围”必须排除单笔交易的客户即只对交易次数≥2的客户计算。这时候lambda x: x.max()-x.min()会报错——当某客户只有一笔交易时x.min()和x.max()是同一个值但业务上认为“单笔无范围概念”。正确解法是写命名函数def business_range(series): if len(series) 2: return np.nan # 明确返回NaN而非0避免被误读为“范围为0” return series.max() - series.min()为什么必须用命名函数而非lambda两点第一lambda无法加docstring半年后你自己都忘了这个range是按什么业务逻辑算的第二pandas对命名函数的错误堆栈更友好——当business_range内部报错时日志里会清晰显示File risk_metrics.py, line 45, in business_range而lambda报错只显示lambda排查成本翻倍。2.3 滚动窗口聚合解决“时间动态性”问题典型业务场景反洗钱系统检测“近期消费突增”。不能只看“本月总额比上月高50%”因为月初可能还没数据。必须用“过去7天滚动均值 vs 历史均值”。这里有个致命陷阱rolling(window7).mean()默认要求窗口内7个非空值但实际交易数据有节假日、周末休市连续7天可能只有3笔交易。如果按默认设置结果全是NaN。正确做法是加参数min_periods3表示只要有3个值就计算均值。这个3不是拍脑袋定的而是根据业务SLA风控规则要求“至少覆盖3天交易行为才触发预警”。注意滚动窗口必须在set_index(date)后进行且date列必须是datetime64类型。我曾遇到一个坑原始数据里date是字符串2024-01-01rolling()会把它当普通索引结果算出的“滚动均值”其实是按行号滚动而非按日期滚动修复只需一行df[date] pd.to_datetime(df[date])。2.4 扩展窗口聚合解决“累积效应”问题典型业务场景客户生命周期价值CLV计算。不是看“当前月消费”而是“从开户至今的累计消费”。expanding().sum()完美匹配但要注意其与cumsum()的本质区别expanding().sum()是窗口函数支持min_periods参数如expanding(min_periods1)而cumsum()是纯累加遇到第一个NaN就全盘崩溃。在真实数据中客户首笔交易可能因系统延迟晚录1小时导致首行amount为NaN——用cumsum()整列变NaN用expanding().sum()则从第二个有效值开始累加。2.5 多级分组unstack解决“业务视角转换”问题典型业务场景销售总监要一张表横轴是产品线Widget/Gadget纵轴是大区North/South单元格是“平均客单价”。这本质是三维数据客户×产品×地区降维到二维矩阵。groupby([region,product])[revenue].mean().unstack()是标准解法但unstack()默认用np.nan填充缺失组合。而业务方要的是“0”——没卖过的产品显示0元而非空白。所以必须加fill_value0。更深层的问题是如果某大区某产品销量为0unstack()后该单元格是0但业务上需要区分“0销量”和“数据未采集”。这时我会额外加一列is_data_available用groupby().size()统计每组记录数再unstack(fill_value0)这样既能显示0销量又能标记数据完整性。这五种模式不是孤立的而是构成一个聚合流水线先用多列聚合产出基础指标再用自定义函数注入业务规则接着用滚动/扩展窗口添加时间维度最后用unstack()适配业务汇报格式。任何跳步都会导致结果失真。比如先unstack()再滚动计算不行unstack()后数据结构已变rolling()无法识别时间序列。必须严格遵循“分组→聚合→时间窗→重塑”的顺序。这个顺序是我用3个通宵回滚生产事故后刻进DNA的。3. 实操细节解析从代码到业务落地的12个关键控制点光知道五种模式还不够真实生产环境里90%的失败源于细节失控。下面这12个控制点每一个都对应我亲手处理过的线上故障。别跳过它们比语法更重要。3.1 控制点1空值处理——不是填0那么简单业务数据里空值NaN无处不在新客户首笔交易手续费未计费、跨境交易汇率字段为空、测试环境模拟数据故意留空。agg()对NaN的默认行为是静默跳过这很危险。比如计算mean()时10笔交易有2笔NaNagg()只对8笔有效值求均值但业务上可能要求“NaN视为0”或“NaN视为缺失需告警”。解决方案是预处理# 方案A全局填充谨慎使用 df[fee] df[fee].fillna(0) # 仅适用于明确业务含义为0的字段 # 方案B聚合时指定skipnaFalse强制报错 df.groupby(category)[fee].agg(lambda x: x.mean(skipnaFalse)) # 任一NaN则整个组返回NaN触发监控告警 # 方案C业务规则填充推荐 df[fee] df.apply( lambda row: 0 if pd.isna(row[fee]) and row[amount] 1000 else row[fee], axis1 )最后一行是真实案例手续费为空且交易额超1000元时按0填充因系统规则大额交易免手续费否则保留原值。这比fillna(0)严谨得多。3.2 控制点2数据类型陷阱——int64和float64的隐式转换pandas的agg()在混合类型列上会自动升格为float64。比如transaction_count是int64agg({count:[sum,mean]})后sum仍是整数但mean变成浮点数。下游系统如Oracle数据库导入时float64列可能被截断小数位导致“平均交易笔数3.0000000000000004”。修复方法聚合后显式转换result df.groupby(category).agg({count: [sum, mean]}) result[(count, mean)] result[(count, mean)].round(2).astype(float) # 保留2位小数3.3 控制点3性能杀手——避免在agg中调用apply新手常写df.groupby(customer_id).apply(lambda x: x[amount].rolling(7).mean())。这是灾难apply()是逐组Python循环10万客户就要循环10万次而rolling()本身是向量化操作。正确姿势是先sort_values(date).set_index(date)再groupby(customer_id)[amount].rolling(7).mean()。实测10万客户数据前者耗时47秒后者耗时1.2秒。3.4 控制点4时序对齐——滚动窗口的索引必须是单调递增rolling()要求索引严格递增。但真实交易数据常有乱序系统延迟导致后发生的交易先入库或批量补录数据时间戳相同。df.sort_values(date).set_index(date)后若存在重复日期rolling()会报错。解决方案# 步骤1去重并保留最新记录 df df.sort_values(ingestion_time).drop_duplicates(subset[date,customer_id], keeplast) # 步骤2对重复日期添加微秒偏移 df[date] df.groupby(date).cumcount().apply(lambda x: pd.Timedelta(x, unitus)) df[date]3.5 控制点5内存优化——大表聚合前必做列类型压缩处理千万级交易表时object类型列如category占内存极大。category列实际只有4个值Dining/Retail等却以object存储内存占用是category类型的5倍。必须在groupby前压缩df[category] df[category].astype(category) # 内存减少80% df[customer_id] df[customer_id].astype(category) # 同理3.6 控制点6业务一致性——确保rolling/expanding的分组键唯一groupby(customer_id)[amount].rolling(7).mean()要求每个customer_id组内date索引唯一。但若同一客户同一天有多笔交易rolling()会把它们当同一时间点处理导致窗口计算错误。必须先按时间聚合# 先按客户日期聚合日度指标 daily_df df.groupby([customer_id, date])[amount].sum().reset_index() # 再对日度数据做滚动 daily_df daily_df.sort_values([customer_id,date]).set_index(date) daily_df[rolling_7day] daily_df.groupby(customer_id)[amount].rolling(7D).mean()注意这里用7D7天而非window77D是时间窗口自动处理非交易日window7是行窗口强制取7行可能跨月。3.7 控制点7unstack的维度安全——永远指定fill_valueunstack()不加fill_value缺失组合会是NaN。而NaN在数值计算中具传染性NaN 100 NaN。业务方看到“North区Gadget产品销售额NaN”会以为系统故障。必须result df.groupby([region,product])[revenue].mean().unstack(fill_value0) # 进一步加固标记数据状态 result[data_status] np.where(result.eq(0).all(axis1), no_data, partial_data)3.8 控制点8自定义函数的异常防御——空组、全NaN组、单值组自定义函数最怕边缘情况。df.groupby(category).apply(risk_metrics)中若某category组为空如新上线品类暂无交易apply()直接报错。必须在函数内兜底def risk_metrics(series): if len(series) 0: # 空组 return pd.Series({high_value_count: 0, high_value_pct: 0, regular_avg: 0}) if series.isna().all(): # 全NaN组 return pd.Series({high_value_count: 0, high_value_pct: 0, regular_avg: np.nan}) # ...正常逻辑3.9 控制点9结果验证——用SQL交叉验证关键指标再自信的pandas代码也要用SQL验证。我坚持的流程对任意聚合结果抽样3个客户用SQL重算-- pandas算出C001的rolling_7day_avg 264.087 SELECT AVG(amount) OVER (PARTITION BY customer_id ORDER BY date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) as rolling_7day FROM transactions WHERE customer_id C001 AND date 2024-01-07;若结果不一致一定是pandas的sort_values或set_index逻辑有偏差。3.10 控制点10可审计性——在结果中嵌入计算元数据业务方常问“这个中位数是按什么规则算的是不是用了插值法”pandas的median()默认用linear插值但业务可能要求lower取下界。解决方案在结果DataFrame中加元数据列result df.groupby(category).agg({ amount: [(median_lower, lambda x: x.median(lower)), (std, std)] }) result.attrs[calculation_rules] { median_lower: pandas median with methodlower, std: sample standard deviation (ddof1) }3.11 控制点11增量更新——避免全量重算生产系统不能每次跑全量。对滚动窗口我用last_computed_date记录上次计算截止日只处理新增数据last_date get_last_computed_date() # 从元数据表查 new_data df[df[date] last_date] # 只对new_data做rolling再与历史结果concat3.12 控制点12监控告警——为聚合结果设置业务阈值agg()结果不是终点而是监控起点。例如transaction_range超过5000元应告警可能为测试数据或异常商户range_result df.groupby(category)[amount].agg(business_range) if (range_result 5000).any(): send_alert(fHigh range detected: {range_result[range_result 5000]})这12个控制点每一个都来自血泪教训。记住在金融数据领域正确的结果不叫成功可验证、可追溯、可监控的结果才叫交付。下一节我们进入完整实操用一个真实的银行信用卡分析案例把这12个点全部串起来。4. 完整实操构建银行级信用卡客户分析流水线现在我们把前面所有理论焊接到一个真实场景里为某城商行信用卡中心构建T1客户行为分析流水线。需求来自风控总监的邮件“需每日早9点前输出《高价值客户异动日报》包含1各客户近7天滚动消费均值2消费金额范围max-min3高价值交易300元占比4按地区×品类的交叉分析表5每位客户累计消费总额。所有指标需支持下钻到单客户明细。”——这就是典型的多维聚合生产需求。下面是我的完整实现含所有避坑细节。4.1 数据准备与清洗从原始交易表到分析就绪原始数据是MySQL表credit_card_transactions每日增量同步到数据湖。第一步不是写groupby而是用SQL做前置清洗pandas不适合做复杂ETL-- 创建临时清洗表 CREATE TABLE temp_cleaned AS SELECT customer_id, DATE(transaction_time) as date, -- 强制转日期解决时间戳精度问题 CASE WHEN merchant_category IN (Jewelry,Casino,Luxury Goods) THEN High_Risk WHEN merchant_category IN (Groceries,Pharmacy) THEN Essential ELSE merchant_category END as category, ROUND(amount, 2) as amount, -- 统一金额精度 CASE WHEN fee IS NULL THEN 0 ELSE ROUND(fee, 2) END as fee, transaction_time FROM credit_card_transactions WHERE DATE(transaction_time) DATE_SUB(CURDATE(), INTERVAL 90 DAY) -- 只取90天控内存 AND amount 0; -- 排除退款、冲正等负值然后用pandas读取清洗后数据# 关键指定列类型省80%内存 dtype { customer_id: category, category: category, amount: float32, # float64太重float32精度足够 fee: float32 } df pd.read_sql(SELECT * FROM temp_cleaned, con, dtypedtype) # 验证数据质量 print(f数据时间范围: {df[date].min()} 到 {df[date].max()}) print(f客户数: {df[customer_id].nunique()}, 品类数: {df[category].nunique()}) print(f空值检查:\n{df.isna().sum()})实操心得永远不要相信上游数据。我遇到过最诡异的Bug某批次数据里customer_id是字符串C001 末尾有空格导致同一客户被分成两组。修复只需一行df[customer_id] df[customer_id].str.strip()。这个strip()我加在每一版代码的第二行。4.2 构建核心聚合流水线五步不可逆顺序步骤1多列多函数聚合——产出基础指标池# 业务要求每个客户每个品类计算均值、中位数、计数、手续费极值 base_agg df.groupby([customer_id, category]).agg({ amount: [mean, median, count, lambda x: x.max() - x.min()], # 业务范围 fee: [min, max, sum] }) # 扁平化列名加业务前缀 base_agg.columns [fbase_{col[0]}_{col[1]} for col in base_agg.columns] base_agg base_agg.reset_index() # 关键控制点处理空值组 base_agg[base_amount_lambda] base_agg[base_amount_lambda].fillna(0) # 范围为空0步骤2自定义函数注入——计算高价值交易占比def high_value_ratio(series): 计算300元交易占比空组返回0 if len(series) 0: return 0.0 high_val_count (series 300).sum() return round((high_val_count / len(series)) * 100, 1) # 保留1位小数 # 应用到每个客户不分品类因业务要求是客户级指标 hv_ratio df.groupby(customer_id)[amount].apply(high_value_ratio) hv_ratio hv_ratio.rename(hv_ratio_pct).reset_index()步骤3滚动窗口计算——7天滚动均值# 先按客户日期聚合日度消费 daily_customer df.groupby([customer_id, date])[amount].sum().reset_index() # 排序并设索引关键 daily_customer daily_customer.sort_values([customer_id, date]) daily_customer daily_customer.set_index(date) # 计算滚动均值用7D时间窗口非window7 daily_customer[rolling_7day] ( daily_customer.groupby(customer_id)[amount] .rolling(7D, min_periods3) # 至少3天有数据才计算 .mean() .reset_index(level0, dropTrue) ) # 关键将滚动结果merge回原表避免索引错位 rolling_result daily_customer[[customer_id, rolling_7day]].reset_index()步骤4扩展窗口计算——累计消费总额# 同样基于日度数据 daily_customer[cumulative_spend] ( daily_customer.groupby(customer_id)[amount] .expanding(min_periods1) # 从第一天开始累加 .sum() .reset_index(level0, dropTrue) ) cumulative_result daily_customer[[customer_id, cumulative_spend]].reset_index()步骤5多级分组unstack——生成交叉分析表# 业务要求客户×品类的平均消费额矩阵 crosstab df.groupby([customer_id, category])[amount].mean().unstack( fill_value0 ).round(2) # 金额保留2位小数 # 为下游BI准备重命名列加前缀 crosstab.columns [fcrosstab_{col} for col in crosstab.columns] crosstab crosstab.reset_index()4.3 结果整合与业务封装从技术表到业务报告现在有5张中间表base_agg、hv_ratio、rolling_result、cumulative_result、crosstab。下一步是按业务逻辑合并而非简单pd.merge()# 主表以base_agg为基础因它含客户品类粒度 final_report base_agg.copy() # 合并客户级指标hv_ratio, cumulative_spend final_report final_report.merge(hv_ratio, oncustomer_id, howleft) final_report final_report.merge( cumulative_result.drop(date, axis1), # 去掉date列避免冲突 oncustomer_id, howleft ) # 合并滚动结果注意rolling_result有date列需取最新日期 latest_rolling rolling_result.loc[rolling_result.groupby(customer_id)[date].idxmax()] final_report final_report.merge( latest_rolling[[customer_id, rolling_7day]], oncustomer_id, howleft ) # 合并交叉表需explode展开因crosstab是宽表 # 先将crosstab转为长表再merge crosstab_long crosstab.melt( id_varscustomer_id, var_namecategory_col, value_namecrosstab_avg_amount ) crosstab_long[category] crosstab_long[category_col].str.replace(crosstab_, ) final_report final_report.merge( crosstab_long, on[customer_id, category], howleft ) # 最终清理重命名、排序、去重 final_report final_report.rename(columns{ base_amount_mean: avg_amount, base_amount_median: median_amount, base_amount_count: transaction_count, base_amount_lambda: amount_range, base_fee_min: min_fee, base_fee_max: max_fee, base_fee_sum: total_fee, rolling_7day: rolling_7day_avg, cumulative_spend: total_cumulative_spend, crosstab_avg_amount: category_avg_amount }) # 按客户、品类排序便于阅读 final_report final_report.sort_values([customer_id, category]).reset_index(dropTrue)4.4 生产部署与监控让流水线自己说话代码写完只是开始。生产环境必须有“自我诊断”能力# 1. 数据质量监控 quality_check { total_records: len(final_report), null_rate: final_report.isna().sum().sum() / final_report.size, zero_range_count: (final_report[amount_range] 0).sum(), hv_ratio_outlier: ((final_report[hv_ratio_pct] 0) | (final_report[hv_ratio_pct] 100)).sum() } if quality_check[null_rate] 0.01: send_alert(f空值率超标: {quality_check[null_rate]:.2%}) # 2. 业务阈值告警 high_risk_customers final_report[ (final_report[amount_range] 5000) (final_report[hv_ratio_pct] 70) ][[customer_id, category, amount_range, hv_ratio_pct]] if not high_risk_customers.empty: send_alert(f高风险客户: {len(high_risk_customers)}人, high_risk_customers) # 3. 性能监控 import time start time.time() # ...执行聚合... end time.time() if end - start 300: # 超5分钟告警 send_alert(f聚合超时: {end-start:.1f}秒) # 4. 输出到业务系统 final_report.to_parquet(s3://bank-data/reports/daily_customer_analytics.parquet, indexFalse, compressionsnappy) # 同时生成Excel供人工核查 final_report.head(1000).to_excel(daily_report_sample.xlsx, indexFalse)这个流水线已在生产运行11个月日均处理2300万条交易平均耗时42秒。它的核心不是多酷炫的代码而是把12个控制点像螺丝一样拧紧在每个环节。比如rolling(7D)里的min_periods3就是为应对国庆长假7天无交易crosstab的fill_value0让BI工具无需额外处理空值hv_ratio函数里的round(...,1)确保业务方看到的百分比是“45.0%”而非“45.00000000000001%”。真正的专业藏在这些毫米级的细节里。5. 常见问题与实战排障那些让你凌晨三点爬起来的Bug再完美的设计也会在生产环境撞上意想不到的墙。我把这些年最常遇到的7类问题连同根因和速查方案整理成这张表。这不是理论清单而是我电脑里troubleshooting.md的实时备份。问题现象根本原因速查命令解决方案我的血泪教训滚动均值全为NaNdate列不是datetime64或索引未排序df[date].dtype,df.index.is_monotonic_increasingdf[date] pd.to_datetime(df[date]);df df.sort_values(date).set_index(date)曾因date是字符串rolling()按字典序滚动算出的“7天均值”其实是按2024-01-01到2024-01-07的字符串排序而非日期unstack后列名带括号BI工具报错MultiIndex列未扁平化result.columnsresult.columns [_.join(col).strip() for col in result.columns.values]某次导出到Tableau因列名(amount,mean)含括号Tableau直接拒绝连接客户会议前2小时紧急修复。自定义函数报错“Series object is not callable”函数名与pandas内置方法重名如sumdir(pd.Series)重命名函数如my_sum避免用sum、mean、count等新人常写def sum(x): return x.sum()结果agg({col: sum})调用时pandas把sum当成了内置方法而非你的函数。内存爆满MemoryErrorobject类型列未压缩或未用float32df.memory_usage(deepTrue)df[col] df[col].astype(category);df[col] df[col].astype(float32)处理1亿行数据时customer_id从object转category内存从12GB降到2.3GB。rolling结果行数变少min_periods默认为window值导致首window-1行为空len(rolling_result) len(df)显式设置min_periods1或用fillna(methodffill)风控要求“首日也需有滚动值”必须fillna(methodffill)但要注意这会让首日值等于当日值业务上是否可接受groupby后结果顺序混乱groupby默认不保持原始顺序df.groupby(...).apply(...).reset_index()加参数sortFalse:df.groupby(col, sortFalse).agg(...)某次报表客户排名按groupby后顺序输出结果业务方说“为什么VIP客户排在最后”因groupby按字母序排了customer_id。空组导致agg报错agg()遇到空组
双梁正问题 【 torch 实战】（Python代码实现）)
