1. 项目概述为什么“多维聚合”不是Pandas进阶技巧而是业务分析的生存技能我在银行风控部门干了七年从刚毕业写SQL查数的分析师到带三个人小团队做反欺诈模型的数据架构师。这七年里我亲手重构过四套核心报表系统也给二十多个业务部门做过数据赋能培训。最常被问到的问题不是“怎么建模”而是“老师这个指标能不能按客户产品时间三个维度一起算现在跑三次groupby再merge一跑就是四十分钟领导在催。”——这句话背后藏着的是真实世界里每天都在发生的效率损耗、逻辑错位和决策延迟。“Part 20: Data Manipulation in Multi-Dimensional Aggregation”这个标题听起来像教科书里的章节编号但在我日常工作中它对应的是一个具体、高频、高价值的场景用一份代码同时回答五个不同角色的问题。财务总监要看各区域各产品的毛利总和与波动率风险经理要盯住某类商户交易金额的极差max-min是否突破阈值运营总监需要滚动30天的客单价趋势线来判断营销活动效果客户成功团队则依赖“客户×品类”的交叉表快速定位高潜力客户群而CEO办公室的晨会简报要求把上述所有结果压缩进一张A4纸的Executive Summary。这些需求绝不是df.groupby(region).sum()能解决的。它们要求你把Pandas当作一台可编程的分析引擎而不是一个单功能计算器。关键词里提到的“Towards AI”其实暗示了这篇文章的底层气质它不讲理论推导不堆砌API文档而是聚焦于“生产环境里真正跑得通、扛得住、改得动”的模式。比如为什么示例中agg()传入的是字典而不是列表因为列表只能统一应用同一函数而字典允许你对transaction_amount求均值和中位数对processing_fee却只取极值——这种混合策略在银行对账系统里是刚需金额要抗异常值所以用中位数费用要防篡改所以看上下限。再比如为什么滚动窗口计算后要用reset_index(level0, dropTrue)因为不这么做索引会变成双层结构后续join其他表时直接报错。这些细节教科书不会写但线上任务失败一次你就得加班两小时排查。我见过太多人卡在“知道语法但不会设计”的阶段。他们能复现示例代码但一拿到真实数据就懵字段名不一致、缺失值没处理、时间序列不连续、分组后数据量爆炸……这不是能力问题是缺乏对“业务问题→数据结构→聚合逻辑→输出形态”全链路的拆解训练。这篇博文就是把我踩过的坑、压测过的方案、和业务方吵架后妥协出的最佳实践全部摊开来讲。它不承诺让你成为Pandas专家但能确保你下次接到“按客户、渠道、月份统计留存率复购率ARPU并叠加滚动季度均值”的需求时心里有底手上不抖。2. 核心思路拆解为什么必须放弃“单步单聚合”的思维定式2.1 传统方式的三大硬伤慢、错、难维护先说个真实案例。去年我们给信用卡中心做月度经营分析原始需求是“统计每个客户在餐饮、零售、旅游三大类商户的近90天交易笔数、总金额、平均单笔、最大单笔、最小单笔”。业务方提供的原始SQL脚本是这样的-- 步骤1算餐饮类 SELECT customer_id, COUNT(*) as dining_cnt, SUM(amount) as dining_amt FROM transactions WHERE categoryDining AND date DATE_SUB(CURDATE(), INTERVAL 90 DAY) GROUP BY customer_id; -- 步骤2算零售类复制粘贴改category SELECT customer_id, COUNT(*) as retail_cnt, SUM(amount) as retail_amt FROM transactions WHERE categoryRetail AND date DATE_SUB(CURDATE(), INTERVAL 90 DAY) GROUP BY customer_id; -- 步骤3算旅游类再复制粘贴 ... -- 步骤N最后LEFT JOIN所有结果这套方案上线后月度跑批耗时从23分钟飙升到87分钟。原因很朴素数据库要对同一张大表扫描四次三次WHERE过滤一次JOIN。更致命的是当某天ETL流程异常导致“餐饮”类数据延迟入库而“零售”类数据已更新最终报表里会出现大量NULL值但业务方根本不知道是数据问题还是逻辑问题——因为错误被层层JOIN掩盖了。我把这个问题带到技术评审会上运维同事当场指出“你们这个查询没有走索引date和category字段都没建联合索引每次都是全表扫描。” 业务方反问“那加索引要多久” 运维答“测试环境要两天生产环境排期两周。” ——你看技术债最终都转化成了业务响应延迟。而Pandas的多维聚合本质是用内存换时间、用结构换清晰。它把整个数据集加载进内存现代服务器几十G内存很常见然后通过一次groupby完成所有维度切片再用向量化操作并行计算各类指标。这不是炫技是工程权衡当你的数据量在千万级以内银行单月交易流水通常如此内存计算的确定性远高于反复IO的SQL查询。2.2 四大核心模式的业务映射逻辑原文提到了Multiple Aggregations、Custom Functions、Rolling/Expanding Windows、Multi-Level Grouping四大模式。但光列名字没用得知道它们各自解决什么层级的业务痛点模式解决的业务问题典型场景为什么不能用简单groupby替代Multiple Aggregations同一维度下不同字段需不同统计口径财务看“收入总额毛利率中位数”风控看“交易笔数单笔金额标准差”简单groupby只能对所有字段用同一函数无法混合mean/std/countCustom Aggregation统计口径需嵌入业务规则非数学函数可表达“高价值交易占比”金额300元的笔数/总笔数“资金沉淀率”月末余额/当月流入总额内置函数无法处理条件分支、阈值判断、多步骤计算Rolling Window需要对比“当前状态”与“近期趋势”识别突变反欺诈系统监测“单日交易额是否超过去7天均值2倍”运营看“促销后3天复购率环比”静态聚合丢失时间维度无法捕捉动态变化Multi-Level Grouping Unstack结果需以业务自然语言呈现而非技术索引销售总监要的不是Index(regionNorth, productWidget)而是“北区Widget产品销售额”这张Excel表多层索引对人不友好下游系统如BI工具、邮件模板难以直接消费特别强调一点Unstack不是为了好看而是为了可交付。我曾把一份未unstack的MultiIndex结果发给分行行长对方回复“这个表格我打不开Excel提示‘无法识别格式’”。后来我改成unstack(fill_value0)对方秒回“收到数据已导入。” ——技术人的优雅在业务侧可能就是一道墙。2.3 生产环境的隐形约束性能、可读性、可审计性很多教程忽略了一个关键事实生产代码不是写给自己看的。它要经受三重考验性能考验某次我们给财富管理部做高净值客户分析原始脚本用循环遍历每个客户计算滚动均值10万客户跑了6小时。改成groupby().rolling().mean()后降到47秒。提速450倍不是算法奇迹是向量化计算对CPU缓存的极致利用。可读性考验半年后新来的分析师要修改报表逻辑。如果他看到df.groupby([cust,prod]).agg({amt:[mean,std],fee:sum})30秒内就能理解意图如果看到一堆merge()和临时DataFrame可能得花半天画流程图。可审计性考验监管检查时要求提供“客户风险评分计算过程”。自定义函数risk_score(series)里有清晰的docstring说明“依据银保监[2023]XX号文第5条对单笔超50万交易加权0.3系数”比SQL里一段CASE WHEN嵌套更容易被审计师认可。所以选择这些模式不是因为“它很酷”而是因为它让代码具备了业务可解释性、性能可预测性、变更可追溯性。这才是资深从业者和新手的本质区别前者写的每一行代码都在为未来三个月的维护成本埋单。3. 实操细节解析从代码到业务结果的完整链路3.1 Multiple Aggregations如何避免“列名地狱”原文示例中result df.groupby(merchant_category).agg({transaction_amount: [mean,median], processing_fee: [min,max]})输出了一个MultiIndex列结构。这在Jupyter里看着清爽但放到生产环境就是灾难——想象一下你要把这个结果写入数据库字段名是(transaction_amount, mean)数据库根本不认这种元组命名。实操心得永远在agg后立即扁平化列名。我的标准做法是# 原始agg result df.groupby(merchant_category).agg({ transaction_amount: [mean, median, std], processing_fee: [min, max, sum] }) # 扁平化用下划线连接内外层列名 result.columns [_.join(col).strip() for col in result.columns] result result.reset_index() # 确保merchant_category变回普通列这样输出就是merchant_categorytransaction_amount_meantransaction_amount_median...提示strip()很重要某些版本pandas会在列名前后加空格导致后续result[transaction_amount_mean]报KeyError。我吃过亏加了strip后稳定运行两年。更进一步业务方往往需要“语义化列名”。比如财务总监不要transaction_amount_std而要amt_volatility金额波动率。这时我会建一个映射字典col_mapping { transaction_amount_mean: avg_amt, transaction_amount_std: amt_volatility, processing_fee_min: min_fee, processing_fee_max: max_fee } result result.rename(columnscol_mapping)这个映射字典本身就可以作为业务术语表Glossary交给数据治理团队一举两得。3.2 Custom Aggregation Functions别让lambda毁掉可维护性原文用了lambda x: x.max() - x.min()计算极差简洁但危险。为什么因为lambda无法被pickle序列化。这意味着如果你用Dask或Spark做分布式计算或者要把这个agg函数存到模型仓库里它会直接报错。实操心得所有自定义函数必须独立定义且满足三个条件函数名见名知意calc_transaction_range而非f1有完整docstring说明业务依据引用监管文件或内部制度包含基础防御性检查空值、长度不足等def calc_transaction_range(series): 计算交易金额极差最大值-最小值 依据《银行反洗钱可疑交易监测指引》第3.2条 对商户类别内交易金额离散度超过200%的触发人工核查 if len(series) 0: return np.nan if series.isna().all(): return np.nan return series.max() - series.min() # 使用时 result df.groupby(merchant_category).agg({amount: calc_transaction_range})注意series.isna().all()比len(series)0更安全因为series可能全是NaN。我曾在线上环境因这个疏漏导致整张报表的极差列全为0被风控部紧急叫停。对于更复杂的逻辑比如原文的weighted_average我建议增加参数校验def weighted_average(series, weight_factor1.0): 加权平均近期交易权重更高 weight_factor: 权重斜率默认1.0线性增长 if not isinstance(weight_factor, (int, float)) or weight_factor 0: raise ValueError(weight_factor must be positive number) n len(series) if n 2: return series.mean() # 生成权重越靠后权重越大但保证总和为1 weights np.linspace(1, 1 weight_factor * (n-1), n) weights weights / weights.sum() # 归一化 return np.average(series, weightsweights)这样当业务方说“把权重调成1.5试试”你只需改一个参数不用动核心逻辑。3.3 Rolling Expanding Windows时间窗口的业务语义陷阱原文示例用rolling(window3)计算3日均值但没提一个致命细节窗口对齐方式。默认是backward向后对齐即2024-01-03的值是[01-01,01-02,01-03]的均值。但业务需求常是“截至当日的滚动均值”这要求forward对齐——然而pandas的rolling不支持forward得用shift(-2)补救。实操心得永远显式指定min_periods参数。原文示例中前两行是NaN这是正确的但生产环境常需填充# 方案1用min_periods1让首日就有值但意义存疑 df_ts[rolling_avg] df_ts.groupby(category)[daily_revenue].rolling( window3, min_periods1 ).mean().reset_index(level0, dropTrue) # 方案2业务上更合理——用当日值填充相当于无历史数据时视同历史均值当前值 df_ts[rolling_avg] df_ts.groupby(category)[daily_revenue].rolling( window3 ).mean().fillna(methodbfill).reset_index(level0, dropTrue)注意fillna(methodbfill)是向后填充即用下一个有效值填当前NaN。但rolling结果的NaN都在开头所以实际是用第一个有效滚动值去填前面的空——这符合“首日无历史就用首日自身值”的业务直觉。对于Expanding窗口原文用expanding().sum()但要注意Expanding默认从第一个值开始累积。如果数据有缺失expanding().sum()会返回NaN直到遇到第一个非空值。更稳健的做法是# 先用ffill填充缺失值业务上常假设无记录0交易 df_ts[daily_revenue_filled] df_ts[daily_revenue].fillna(0) df_ts[cumulative_sum] df_ts.groupby(category)[daily_revenue_filled].expanding().sum()3.4 Multi-Level Grouping with Unstack重塑业务认知的魔法原文df_sales.groupby([region,product])[revenue].mean().unstack()输出矩阵但没解决一个关键问题当某个region-product组合不存在时unstack默认留NaN而业务方要的是0。实操心得unstack(fill_value0)是底线但更好的是用pivot_table它原生支持缺失值填充和聚合函数# 更健壮的写法 result df_sales.pivot_table( valuesrevenue, indexregion, columnsproduct, aggfuncmean, # 可指定任意函数不止mean fill_value0, # 不存在的组合填0 marginsTrue, # 自动加总计行/列业务最爱 dropnaFalse # 即使region或product有NaN也保留 )marginsTrue生成的All行就是“各产品全量平均值”这对销售总监做资源分配决策至关重要。而dropnaFalse能捕获脏数据——比如某条记录region为空它会被归到NaN行提醒你清洗数据。提示pivot_table比groupby().unstack()多一次隐式排序性能略低但胜在语义清晰、容错性强。在报表类场景我永远选pivot_table。4. 端到端实战构建银行信用卡客户分析流水线4.1 数据准备模拟真实世界的脏数据原文生成的模拟数据过于干净。真实信用卡数据有三大特征时间不连续、字段缺失、类型混杂。我重写了数据生成逻辑加入这些痛点import pandas as pd import numpy as np from datetime import datetime, timedelta np.random.seed(42) # 模拟不连续日期跳过周末和节假日 base_dates pd.date_range(2024-01-01, 2024-03-31, freqD) # 随机去掉30%的日期模拟系统故障或数据延迟 dates np.random.choice(base_dates, sizeint(len(base_dates)*0.7), replaceFalse) dates np.sort(dates) # 客户数据加入休眠客户最后30天无交易 customers [C001, C002, C003, C004, C005] # C004最后15天无交易C005最后30天无交易 active_mask np.random.choice([True, False], sizelen(dates), p[0.9, 0.1]) # 强制C004/C005在末期无交易 end_idx int(len(dates)*0.8) for i in range(end_idx, len(dates)): if np.random.rand() 0.7: # 70%概率不生成C004交易 active_mask[i] False # 生成交易记录 records [] for date in dates: # 每日交易量波动 daily_count max(1, int(np.random.poisson(5))) # 泊松分布模拟交易频次 for _ in range(daily_count): # 随机选客户但倾向活跃客户 cust np.random.choice(customers, p[0.3, 0.3, 0.2, 0.1, 0.1]) # 休眠客户不生成后期交易 if cust in [C004, C005] and date dates[end_idx]: continue category np.random.choice([Groceries, Dining, Travel, Retail, Utilities], p[0.25, 0.2, 0.15, 0.25, 0.15]) # 金额不同品类有不同分布 if category Travel: amount round(np.random.lognormal(6.2, 0.5), 2) # 高值偏态 elif category Utilities: amount round(np.random.normal(80, 20), 2) # 低值正态 else: amount round(np.random.lognormal(5.5, 0.4), 2) # 加入缺失值1%概率fee为空 fee round(amount * 0.025, 2) if np.random.rand() 0.01 else np.nan records.append({ date: date, customer_id: cust, category: category, amount: amount, fee: fee }) df pd.DataFrame(records) print(f原始数据量{len(df)}行) print(f日期范围{df[date].min()} 至 {df[date].max()}) print(f客户分布\n{df[customer_id].value_counts().sort_index()})这段代码生成的数据天然包含时间断点非每日连续客户休眠C004/C005后期无交易字段缺失fee有1% NaN金额分布差异旅行类金额远高于水电类这才是你明天早上打开邮箱看到的真实数据。4.2 分析1多维聚合——客户×品类×时间的立体透视业务需求“统计每个客户在各品类的近30天交易均值、中位数、标准差并标记是否为‘高波动客户’标准差均值的50%”。# 步骤1确定时间窗口最近30天 cutoff_date df[date].max() window_start cutoff_date - pd.Timedelta(days30) # 步骤2筛选窗口内数据 df_window df[df[date] window_start].copy() # 步骤3多维聚合客户品类 agg_result df_window.groupby([customer_id, category]).agg({ amount: [mean, median, std], fee: [count, sum] }).round(2) # 步骤4扁平化列名 agg_result.columns [_.join(col).strip() for col in agg_result.columns] agg_result agg_result.reset_index() # 步骤5计算业务指标 agg_result[volatility_ratio] (agg_result[amount_std] / agg_result[amount_mean]).fillna(0) agg_result[is_high_volatility] agg_result[volatility_ratio] 0.5 # 步骤6透视成业务友好格式客户为行品类为列 pivot_result agg_result.pivot_table( indexcustomer_id, columnscategory, values[amount_mean, amount_std, is_high_volatility], aggfuncfirst, # 每个组合唯一用first即可 fill_value0 ) # 展平列名便于阅读 pivot_result.columns [_.join(col).strip() for col in pivot_result.columns] pivot_result pivot_result.reset_index() print(客户×品类×30天分析结果) print(pivot_result.head())输出示例customer_idamount_mean_Groceriesamount_std_Groceriesis_high_volatility_Groceries...C001125.3045.200实操心得pivot_table(aggfuncfirst)比unstack()更安全因为它明确声明“每个单元格只取一个值”避免了MultiIndex索引混乱的风险。而fillna(0)确保休眠客户如C005在Groceries列显示0而非NaN业务方一眼就知道“该客户近期无此品类交易”。4.3 分析2自定义聚合——构建客户风险画像业务需求“对每个客户计算其‘高价值交易占比’金额300元、‘夜间交易占比’交易时间在22:00-05:00、‘跨省交易占比’商户省份≠客户注册省份”。# 假设我们有客户省份和商户省份映射表真实场景从维度表join province_map { C001: Beijing, C002: Shanghai, C003: Guangdong, C004: Zhejiang, C005: Sichuan } # 商户省份模拟简化 merchant_province [Beijing, Shanghai, Guangdong, Zhejiang, Sichuan] * 10 df[merchant_province] np.random.choice(merchant_province, len(df)) # 自定义风险画像函数 def build_risk_profile(group): 为单个客户构建风险画像 total len(group) if total 0: return pd.Series({ high_value_pct: 0, night_trade_pct: 0, cross_province_pct: 0, risk_score: 0 }) # 高价值交易 high_value (group[amount] 300).sum() # 夜间交易提取小时 group[hour] pd.to_datetime(group[date]).dt.hour night_trade ((group[hour] 22) | (group[hour] 5)).sum() # 跨省交易 cust_province province_map.get(group[customer_id].iloc[0], Unknown) cross_province (group[merchant_province] ! cust_province).sum() # 综合风险分加权 risk_score ( (high_value / total) * 0.4 (night_trade / total) * 0.3 (cross_province / total) * 0.3 ) return pd.Series({ high_value_pct: round((high_value / total) * 100, 1), night_trade_pct: round((night_trade / total) * 100, 1), cross_province_pct: round((cross_province / total) * 100, 1), risk_score: round(risk_score, 3) }) # 应用聚合 risk_profile df.groupby(customer_id).apply(build_risk_profile).reset_index() print(\n客户风险画像) print(risk_profile)输出customer_idhigh_value_pctnight_trade_pctcross_province_pctrisk_scoreC00112.58.325.00.172注意groupby().apply()比agg()更灵活因为它接收整个子DataFrame可做跨字段计算如hour提取。但性能稍低数据量大时慎用。我的经验是百万行以内放心用千万行以上考虑用assign()预计算字段再agg。4.4 分析3滚动窗口——识别客户行为突变业务需求“对每个客户计算其近7天交易金额的滚动均值并标记‘突增客户’当日金额 近7天均值的2倍”。# 步骤1按客户和日期排序关键 df_sorted df.sort_values([customer_id, date]).set_index(date) # 步骤2计算滚动均值注意min_periods3避免早期NaN过多 df_sorted[rolling_7d_mean] df_sorted.groupby(customer_id)[amount].rolling( window7, min_periods3 ).mean().reset_index(level0, dropTrue) # 步骤3标记突增需对齐索引 df_sorted[is_surge] ( df_sorted[amount] df_sorted[rolling_7d_mean] * 2 ).astype(int) # 步骤4取最新一天的结果业务最关心“当前” latest_date df_sorted.index.max() surge_today df_sorted[df_sorted.index latest_date][[customer_id, amount, rolling_7d_mean, is_surge]] surge_today surge_today[surge_today[is_surge] 1].copy() surge_today[surge_ratio] (surge_today[amount] / surge_today[rolling_7d_mean]).round(2) print(f\n{latest_date} 突增客户金额超7日均值2倍) print(surge_today[[customer_id, amount, rolling_7d_mean, surge_ratio]])输出customer_idamountrolling_7d_meansurge_ratioC0021250.00420.502.97提示reset_index(level0, dropTrue)中的level0指重置第一层索引即customer_iddropTrue表示不把customer_id放回列中。这是保持索引对齐的关键漏掉会引发ValueError: cannot reindex from a duplicate axis。4.5 分析4多级透视——生成管理层简报业务需求“生成一张Excel报表包含1各客户在各品类的平均交易额交叉表2各客户总交易额排名3各品类总交易额排名”。# 交叉表客户×品类 crosstab df.pivot_table( valuesamount, indexcustomer_id, columnscategory, aggfuncmean, fill_value0 ).round(2) # 总交易额排名 total_by_customer df.groupby(customer_id)[amount].sum().sort_values(ascendingFalse) rank_by_customer total_by_customer.rank(methodmin, ascendingFalse).astype(int) # 品类排名 total_by_category df.groupby(category)[amount].sum().sort_values(ascendingFalse) rank_by_category total_by_category.rank(methodmin, ascendingFalse).astype(int) # 合并结果 summary pd.DataFrame({ total_spend: total_by_customer, spend_rank: rank_by_customer, category_total: df.groupby(customer_id)[category].apply(lambda x: x.mode().iloc[0] if not x.mode().empty else Unknown), top_category_spend: [ df[df[customer_id]cid].groupby(category)[amount].sum().sort_values(ascendingFalse).iloc[0] for cid in total_by_customer.index ] }).round(2) print(\n管理层简报摘要) print(summary) print(\n客户×品类交叉表) print(crosstab)输出摘要customer_idtotal_spendspend_rankcategory_totaltop_category_spendC00212540.301Dining4250.20实操心得rank(methodmin)确保并列时取最小名次如两个客户并列第一则都标1下一名次为3。这比默认的average更符合业务排名习惯。而x.mode().iloc[0]取众数避免了用value_counts().index[0]在空Series时报错。5. 常见问题与避坑指南那些让我凌晨三点改代码的瞬间5.1 问题速查表高频报错与根因分析报错信息根本原因解决方案我的血泪史ValueError: Index contains duplicate entriesgroupby后unstack()时存在重复的region-product组合用drop_duplicates(subset[region,product])去重或改用pivot_table(aggfuncfirst)曾因商户数据同步延迟导致同一天同一商户产生两条相同记录unstack直接崩溃重启服务花了40分钟TypeError: incompatible index of inserted column with frame indexrolling()结果索引与原DataFrame索引不匹配务必用.reset_index(level0, dropTrue)重置分组索引第一次写滚动计算时没加这句结果rolling_avg列全是NaN排查了3小时才发现索引错位MemoryErrorwhengroupby().agg()on large data多维分组后中间结果爆炸如10万客户×100品类1000万行改用pd.cut()对连续字段分箱如pd.cut(amount, bins10)或用sample(frac0.1)抽样验证逻辑处理全量信用卡数据时groupby([cust,card_type,month])生成2亿行中间表服务器内存爆满SettingWithCopyWarningwhen assigning torollingresult对rolling结果直接赋值pandas无法确定是视图还是副本用loc明确索引或先copy()再赋值df.loc[:, rolling_avg] df.groupby(...)...这个警告看似无害但会导致后续计算结果随机错乱线上跑了两周才发现部分客户指标不准5.2 那些没人告诉你的性能陷阱陷阱1agg()字典的键顺序影响性能很多人以为{amt:[mean,std], fee:[sum]}和{fee:[sum], amt:[mean,std]}一样快。实测发现把计算量大的函数如std放在前面能提升15%速度。因为pandas内部会复用中间结果先算std时mean可直接复用其均值计算。陷阱2rolling().mean()vsrolling().apply(np.mean)前者是C实现后者是Python调用速度差5-10倍。但apply支持自定义函数所以我的原则是内置函数优先用原生方法自定义逻辑才用apply。陷阱3unstack()前不sort_index()当MultiIndex未排序时unstack()会触发隐式排序消耗大量CPU。我的固定流程是result df.groupby([a,b]).agg({...}) result result.sort_index() # 显式排序 result result.unstack(fill_value0)5.3 生产环境必备的防御性编程所有线上脚本我强制加入这三道防线# 防线1输入数据质量检查 def validate_input(df): assert not df.empty, 输入数据为空 assert customer_id in df.columns, 缺少customer_id字段 assert df[date].dtype datetime64[ns], date字段非datetime类型 assert df[amount].min() 0, 存在负金额交易请检查数据源 # 防线2聚合结果完整性检查 def validate_agg_result(result, expected_cols): missing_cols set