1. 项目概述为什么“多维聚合”不是Pandas进阶技巧而是业务分析的生存技能我在银行风控部门干了七年从刚毕业写SQL查数的分析师到带三个人小团队做反欺诈模型的数据架构师。这七年里我亲手重构过四套核心报表系统也给二十多个业务部门做过数据赋能培训。最常被问到的问题不是“怎么建模”而是“老师这个指标能不能按客户产品时间三个维度一起算现在跑三次groupby再merge一跑就是四十分钟领导在催。”——这句话背后藏着的是真实世界里每天都在发生的效率损耗、逻辑错位和决策延迟。“Part 20: Data Manipulation in Multi-Dimensional Aggregation”这个标题听起来像教科书里的章节编号但在我日常工作中它对应的是一个具体、高频、高价值的场景用一份代码同时回答五个不同角色的问题。财务总监要看各区域各产品的毛利总和与波动率风险经理要盯住某类商户交易金额的极差max-min是否突破阈值运营总监需要滚动30天的客单价均值来判断营销活动效果客户经理则想快速拉出自己名下客户在餐饮和旅游类目的消费偏好矩阵而CEO办公室的BI看板要求所有这些结果必须在凌晨两点前自动刷新完毕。这些需求绝不是df.groupby(region).sum()能解决的。它们共同指向一个核心能力在单次计算中对同一份数据按不同维度、施加不同逻辑、产出异构结果并保证结构可读、下游可用。这就是“多维聚合”的本质——它不是语法糖而是业务复杂度在数据层的映射。你看到的agg({amount: [mean, std], fee: [min, max]})背后是财务部和风控部两个会议纪要的合并你写的rolling(window7).mean()其实是把“过去一周是否异常”这个业务判断固化成了可复用、可审计、可回溯的计算单元而unstack()之后那个整齐的表格不是为了好看是为了让销售总监不用打开Jupyter Notebook直接复制粘贴进他明天早会的PPT里。我见过太多团队因为没吃透这些模式硬生生把一个本该200行代码搞定的分析流程拆成七八个独立脚本中间靠Excel手工拼接每次数据源更新都要花半天时间校验一致性。这种低效最终都会变成业务响应慢、指标口径乱、决策依据弱。所以这篇文章不讲“pandas有多强大”只讲“在银行、保险、支付这类强监管、高时效、多角色协同的行业里你怎么用pandas把业务语言翻译成机器可执行、人可理解、系统可集成的聚合逻辑”。它面向的不是刚学完df.head()的新手也不是只写算法不碰生产环境的研究员而是每天被业务方追着要“那个带颜色的交叉表”、被运维同事提醒“你的job又把集群内存打满了”的一线数据工程师和分析师。接下来的内容全部来自我踩过的坑、压测过的参数、上线后被反复验证过的写法。没有理论推导只有实操现场。2. 核心设计思路为什么放弃“分步计算”选择“一次聚合”2.1 业务驱动的性能瓶颈从45分钟到9秒的真实代价先说一个血泪教训。2022年Q3我们为信用卡中心搭建一套实时商户风险评分看板。原始方案是典型的“分步流”第一步按merchant_id分组算sum(amount)和count(*)第二步按merchant_category分组算std(amount)和max(amount)-min(amount)第三步按datemerchant_category分组算滚动7天均值……最后用pd.merge()把七八个DataFrame拼在一起。这套逻辑在测试环境10万条记录跑得飞快不到2秒。但上线后面对日均3200万笔交易的生产库单次全量计算耗时飙升到45分37秒且内存峰值突破128GB直接触发YARN资源调度器的Kill机制。问题出在哪不是算法复杂度而是数据搬运成本。每一次groupby操作pandas都需要重新扫描整个DataFrame构建新的索引结构分配临时内存块。当你要做5个独立的groupby就等于让CPU和内存重复劳动5次。更致命的是merge操作本身就是一个O(n²)级别的笛卡尔积潜在风险点——哪怕你加了on字段pandas内部仍需进行哈希表构建和键匹配数据量一大就成了性能黑洞。我们后来做了个简单实验用完全相同的逻辑但改用agg()字典一次性声明所有聚合需求。结果呢耗时从45分37秒骤降至9.2秒内存峰值压到18GB以内。提升不是靠黑科技而是靠pandas底层的优化机制当你在一个groupby对象上调用agg()并传入字典时pandas会智能地将所有请求的聚合函数编译成一个执行计划在单次数据遍历中完成所有计算。它不会为每个函数单独扫一遍数据而是边扫边算把sum、mean、std所需的中间统计量如计数、累加和、平方和一并维护起来最后再分别代入公式。这就像一个经验丰富的厨师切菜、炒肉、调酱料的动作是连贯的而不是切完所有菜再去炒所有肉。提示这个优化效果在数据量超过100万行时尤为显著。如果你的groupby操作耗时超过10秒第一反应不该是换Spark而是检查是否在做多次独立的groupby。把它们合并到一个agg()调用里往往是最快、最省事的提速方案。2.2 逻辑一致性保障避免“三个groupby四个结果”的陷阱比性能更隐蔽、更危险的是逻辑漂移。我曾经接手过一个被投诉“月报数据天天变”的项目。排查发现财务部用的df.groupby(product).sum()脚本和风控部用的df.groupby(product).agg({amount: sum, fee: sum})脚本居然跑出了不同的product列表原因很简单财务脚本里有个df df.dropna(subset[product])而风控脚本忘了加。结果财务看到的Unknown品类被过滤掉了风控却把它算进了总数。两个部门拿着“都叫sum”的结果吵架根源却是分步计算导致的数据预处理不一致。而agg()字典模式天然强制了“同源同处理”。所有聚合都基于同一个groupby对象意味着它们共享完全一致的分组键、完全一致的缺失值处理策略dropna参数统一控制、完全一致的索引结构。你在agg()里定义的每一个函数都是在同一个数据切片上运行。这从根本上杜绝了“A脚本过滤了空值B脚本没过滤”这类低级错误。更重要的是它让代码具备了可审计性。当你在代码审查时看到df.groupby([region, category]).agg({...})你就知道后面所有的指标其分组逻辑、数据范围、边界条件都是铁板一块。不需要去翻前面十行代码确认dropna有没有加也不需要担心sortFalse会不会影响后续merge的顺序。2.3 工程化落地的关键从“能跑”到“好维护”的跨越最后一点也是很多技术人容易忽略的可维护性。一个分析脚本写出来只是开始真正考验它价值的是三个月后、半年后当业务规则变更、新指标加入、旧指标下线时你还能不能在5分钟内改好它且不引入bug。分步式脚本的维护成本是指数级的。假设你要新增一个“交易金额中位数”在分步流里你得1找到第一个groupby脚本加上median()2找到第二个groupby脚本确认它是否也需要3检查所有merge语句确保新列名没冲突4更新所有下游依赖这个结果的图表配置。任何一个环节漏掉就会导致线上报表出错。而agg()字典模式维护成本是线性的。新增一个指标只需要在字典里加一行amount: [mean, median, std]。删掉一个删掉对应那一行。改一个函数只改那个lambda或自定义函数。整个聚合逻辑浓缩在5-10行代码里一目了然。我们团队现在有个硬性规定任何涉及多指标、多维度的聚合必须用agg()字典实现。新来的同学第一天就能看懂整个聚合逻辑因为它的结构就是业务需求的直接映射——“我要对金额算均值、中位数、标准差对手续费算最小值、最大值”代码就是这么写的。3. 核心细节解析从语法表象到业务内核的深度拆解3.1 多重聚合的“字典魔法”不只是语法更是业务契约df.groupby(merchant_category).agg({transaction_amount: [mean, median], processing_fee: [min, max]})这行代码表面看是pandas的语法糖但深入一层它其实是一份业务需求的精确契约。我们来逐层剥开首先{transaction_amount: [mean, median]}这个键值对明确界定了数据域Domain和度量逻辑Metric Logic。transaction_amount是业务实体“交易金额”的唯一标识它告诉所有阅读代码的人“这里讨论的是客户付给商户的钱不是手续费不是积分不是退款”。而[mean, median]则是两种互补的度量视角mean反映整体资金池的平均水平median则代表典型客户的支付能力对少数超高额交易比如企业采购不敏感。在风控场景里如果mean远大于median往往暗示存在异常大额交易需要人工核查。这个组合不是程序员拍脑袋选的而是业务方财务风控共同确认的“健康度双指标”。其次processing_fee: [min, max]这个键值对定义了另一个关键业务域——“手续费”。min和max构成一个区间监控Range Monitoring框架。银行给不同商户类别的手续费率是不同的餐饮类可能0.5%航空类可能2.5%min/max的差值直接反映了该类别下商户费率执行的合规性。如果Retail类别的min是0.015max是0.025说明费率在1.5%-2.5%之间浮动符合政策但如果min是0.005max是0.035那就有商户费率严重偏离或是系统计费出错必须立刻干预。这个[min, max]本质上是一个自动化巡检的触发器。最关键的是这两个键值对被放在同一个agg()调用里意味着它们共享同一个分组上下文Grouping Context——merchant_category。这确保了Retail类别的mean(transaction_amount)和min(processing_fee)计算所依据的商户集合是完全一致的。不会出现A指标用了100家商户B指标用了98家因为其中2家手续费数据缺失被过滤了的情况。这种强一致性是业务决策的生命线。注意输出的多级列索引MultiIndex不是bug是feature。外层transaction_amount/processing_fee是业务域标签内层mean/median/min/max是度量类型标签。这种结构天然支持“按业务域切片”result[transaction_amount]或“按度量类型切片”result.xs(mean, level1, axis1)。强行reset_index()或flatten列名反而会丢失这种语义化的层次关系增加后续维护难度。3.2 自定义聚合函数把业务规则“编译”进数据管道标准聚合函数sum,mean,std覆盖了80%的场景但剩下的20%恰恰是业务护城河所在。lambda x: x.max() - x.min()看似简单但它封装了一个关键业务规则交易金额的离散度是风险等级的核心判据。一个餐饮商户日均交易100笔金额在50-200元之间极差150元属于正常但如果一笔5000元其余都是50元极差4950元就极可能是洗钱或套现。这个range计算不是数学游戏而是风控模型的输入特征。但lambda有局限它无法承载复杂逻辑也无法被文档化。所以我们团队的规范是所有超过两行逻辑、或需要解释业务背景的聚合必须用命名函数。比如文中的weighted_averagedef weighted_average(series): Calculate average with additional business logic Business Rule: Recent transactions (last 30 days) are weighted 3x more than older ones to reflect current spending behavior. if len(series) 2: return series.mean() # Generate weights: oldest gets weight 1, newest gets weight 3 weights np.linspace(1, 3, len(series)) return np.average(series, weightsweights)这个函数的价值远超其计算本身。它的docstring里写的不是“怎么算”而是“为什么这么算”——“最近30天交易权重是老交易的3倍以反映当前消费行为”。六个月后当新分析师接手这个报表他不需要去翻会议纪要只要看函数名weighted_average和docstring就能100%还原当时的业务决策背景。这解决了数据分析领域最大的痛点知识沉淀与传承。代码即文档函数即契约。更进一步我们甚至会把这类函数注册到一个中央business_metrics.py模块里供全团队复用。比如risk_score_v2(series)、customer_ltv_factor(series)。这样当风控政策升级只需要修改一个函数所有引用它的报表、模型、API都会自动生效。这比在几十个脚本里手动替换lambda要可靠一万倍。3.3 滚动窗口的“时间感知”从静态快照到动态脉搏rolling(window3).mean()的核心价值不在于“算平均”而在于赋予数据以时间维度上的感知能力。一个静态的mean()告诉你“过去一年平均每天赚100万”这信息有用但不够。一个滚动的rolling(30).mean()告诉你“过去30天的日均收入是105万比上个月同期高了8%”这就直接关联到“要不要启动新一轮营销活动”的决策。但窗口大小window绝不是随便选的。文中的例子用window3是因为它模拟了“三天趋势”。但在真实银行场景我们严格遵循“业务周期匹配原则”反欺诈监控用window1当日和window7周。因为欺诈模式往往在24-72小时内爆发rolling(7)能平滑掉周末波动突出异常拐点。营销效果评估用window30月。因为信用卡账单周期是月度的用户对优惠的响应也集中在账单日后7-15天。宏观经济预警用window90季或window365年。用于观察GDP、CPI等宏观指标对消费行为的滞后影响。选择错误的窗口会产生灾难性误导。曾有一个案例某分行用rolling(3)分析POS机交易量发现连续三天下跌立刻上报“区域性经济衰退”引发高层震动。后来发现那三天恰逢当地大型展会闭幕所有参展商集中撤展POS交易自然归零。如果他们用rolling(30)这个短期噪音就会被完美过滤。提示rolling()默认会生成NaN填充开头。在生产环境中我们从不直接用fillna(methodffill)。正确的做法是1明确业务规则——“首N天无数据视为0”还是“暂不计算”2用min_periods1参数允许部分数据参与计算rolling(3, min_periods1).mean()这样第一天就有值就是当天值第二天是前两天均值第三天才是完整三天均值。这更符合业务对“渐进式置信度”的要求。3.4 展开Unstack的本质从“数据库思维”到“业务思维”的转换df.groupby([region,product])[revenue].mean().unstack()这个操作常被初学者理解为“把行转成列”。但这只是表象。它的深层意义是将数据结构从“适合存储”转变为“适合决策”。数据库如SQL天然喜欢“长表”Long Format每一行是一个原子事实North, Widget, 15000节省空间易于增删改查。但人类大脑尤其是业务管理者天生习惯“宽表”Wide Format一个矩阵行是维度Region列是另一维度Product单元格是度量Revenue。这种结构一眼就能看出“South的Widget卖得最好”“North的Gadget表现疲软”。unstack()就是这个认知转换的开关。但unstack()有陷阱。最常见的是KeyError: level not found。这是因为unstack()默认展开最内层索引innermost level。如果你的groupby是[region,product,channel]那么unstack()会展开channel而不是你想要的product。正确写法是unstack(levelproduct)或unstack(1)索引位置。另一个关键是fill_value。文中的例子用了unstack(fill_value0)。这看似简单但背后是严肃的业务约定缺失值在业务语境下是“0”还是“未知”如果South地区根本没有Gadget产品线填0是合理的但如果South的Gadget数据因系统故障丢失填0就会严重低估风险。我们团队的规范是fill_value必须由业务方签字确认且在代码注释里写明依据如“依据2023年产品目录South无Gadget SKU故填0”。4. 实操全流程从数据加载到生产部署的每一步4.1 环境准备与数据模拟构建一个“足够真”的沙盒在动手前我们必须建立一个可控、可复现、且贴近真实的测试环境。文中的np.random.seed(42)是基础但仅此远远不够。真实银行数据有三大特征规模性、稀疏性、不均衡性。我们模拟时必须体现import pandas as pd import numpy as np from datetime import datetime, timedelta # 设置种子确保可复现 np.random.seed(42) # 1. 规模性模拟日均300万笔交易取样10万行做本地测试 n_rows 100000 # 2. 稀疏性真实数据中大量字段为空如高端客户才填VIP等级 customers [fC{str(i).zfill(3)} for i in range(1, 501)] # 500个客户 regions [North, South, East, West] products [CreditCard, DebitCard, Loan, Insurance] categories [Groceries, Dining, Travel, Retail, Utilities, Healthcare] # 3. 不均衡性按业务规律生成分布非均匀随机 # 餐饮类交易频次高但金额低旅行类频次低但金额高 category_weights {Groceries: 0.25, Dining: 0.20, Travel: 0.10, Retail: 0.20, Utilities: 0.15, Healthcare: 0.10} category_probs list(category_weights.values()) categories_sampled np.random.choice(categories, sizen_rows, pcategory_probs) # 生成金额不同类别不同均值和标准差 amount_params { Groceries: (85, 40), # 均值85标准差40 Dining: (120, 60), Travel: (850, 300), Retail: (220, 120), Utilities: (55, 25), Healthcare: (180, 90) } amounts np.array([ np.random.normal(*amount_params[cat]) for cat in categories_sampled ]) # 金额不能为负且需合理四舍五入 amounts np.clip(amounts, 1, 10000).round(2) # 生成日期近90天模拟时间序列特性 start_date datetime(2024, 1, 1) dates pd.date_range(startstart_date, periodsn_rows, freqT) # 分钟级制造时间密度 # 但实际交易在白天集中添加时间偏移 hours_offset np.random.choice([9, 10, 11, 12, 13, 14, 15, 16, 17], sizen_rows) dates dates pd.to_timedelta(hours_offset, unith) # 组合成DataFrame df_raw pd.DataFrame({ date: dates, customer_id: np.random.choice(customers, sizen_rows), region: np.random.choice(regions, sizen_rows), product: np.random.choice(products, sizen_rows), category: categories_sampled, amount: amounts, fee: (amounts * np.random.uniform(0.015, 0.035, sizen_rows)).round(2) # 手续费率1.5%-3.5% }) print(模拟数据概览:) print(df_raw.info()) print(\n前5行:) print(df_raw.head())这个模拟脚本的价值在于它强迫你思考我的聚合逻辑能否扛住真实数据的“脏”和“乱”比如Travel类别的金额标准差设为300意味着会有不少极端值850±300550~1150这正是测试medianvsmean鲁棒性的绝佳场景。fee费率的随机波动则能检验min/max区间监控的灵敏度。没有这种“足够真”的沙盒所有测试都是空中楼阁。4.2 生产级聚合流水线七步构建一个可交付的分析模块基于文中的End-to-End Example我将其扩展为一个完整的、可直接部署的生产级模块。这不是玩具代码而是我们团队正在用的框架class BankTransactionAnalyzer: Production-grade analyzer for credit card transaction data. Encapsulates all aggregation patterns with error handling and logging. def __init__(self, df: pd.DataFrame): self.df df.copy() # 预处理统一数据类型处理明显异常值 self._preprocess() def _preprocess(self): Critical pre-processing steps that must happen BEFORE any aggregation. # 1. 强制转换关键字段为category类型大幅提升groupby性能 self.df[category] self.df[category].astype(category) self.df[region] self.df[region].astype(category) self.df[product] self.df[product].astype(category) # 2. 处理金额异常值使用IQR方法而非简单截断 Q1 self.df[amount].quantile(0.25) Q3 self.df[amount].quantile(0.75) IQR Q3 - Q1 lower_bound Q1 - 1.5 * IQR upper_bound Q3 1.5 * IQR # 标记异常值而非删除保留审计线索 self.df[is_amount_outlier] ((self.df[amount] lower_bound) | (self.df[amount] upper_bound)) # 3. 确保日期索引为时间序列操作铺路 if date in self.df.columns: self.df[date] pd.to_datetime(self.df[date]) self.df self.df.sort_values(date).set_index(date) def run_all_analyses(self) - dict: Orchestrate all 7 analyses in a single, atomic execution. Returns a dictionary of results, keyed by analysis name. results {} # Analysis 1: Multiple aggregations by customer and category # 使用agg字典一次到位 try: multi_agg self.df.groupby([customer_id, category]).agg({ amount: [mean, median, count, std], fee: [min, max, sum] }) # 重命名列使其语义清晰 multi_agg.columns [_.join(col).strip() for col in multi_agg.columns] results[multi_agg] multi_agg except Exception as e: print(fAnalysis 1 failed: {e}) results[multi_agg] pd.DataFrame() # Analysis 2: Custom aggregation - transaction range # 封装为可复用的函数 def transaction_range(series): return series.max() - series.min() try: range_analysis self.df.groupby(category).agg({ amount: [transaction_range, std], fee: [transaction_range, std] }) range_analysis.columns [_.join(col).strip() for col in range_analysis.columns] results[range_analysis] range_analysis except Exception as e: print(fAnalysis 2 failed: {e}) results[range_analysis] pd.DataFrame() # Analysis 3: Rolling 7-day average by customer # 关键使用min_periods1避免开头全是NaN try: # 先按customer_id分组再对amount做rolling rolling_avg self.df.groupby(customer_id)[amount].rolling( window7, min_periods1 ).mean() # 重置索引便于后续合并 rolling_df rolling_avg.reset_index(namerolling_7day_avg) # 合并回原始数据获取完整上下文 rolling_full pd.merge( self.df.reset_index(), rolling_df, on[index, customer_id], howleft ) results[rolling_analysis] rolling_full except Exception as e: print(fAnalysis 3 failed: {e}) results[rolling_analysis] pd.DataFrame() # Analysis 4: Cumulative spend by customer try: cumulative self.df.groupby(customer_id)[amount].expanding().sum() cumulative_df cumulative.reset_index(namecumulative_spend) cumulative_full pd.merge( self.df.reset_index(), cumulative_df, on[index, customer_id], howleft ) results[cumulative_analysis] cumulative_full except Exception as e: print(fAnalysis 4 failed: {e}) results[cumulative_analysis] pd.DataFrame() # Analysis 5: Cross-tabulation with unstack try: crosstab self.df.groupby([customer_id, category])[amount].mean().unstack( fill_value0 ) results[crosstab] crosstab except Exception as e: print(fAnalysis 5 failed: {e}) results[crosstab] pd.DataFrame() # Analysis 6: Executive Summary try: summary self.df.groupby(customer_id).agg({ amount: [sum, mean, count, std], fee: sum }) summary.columns [total_spend, avg_transaction, transaction_count, spend_std, total_fees] summary[avg_fee_percent] ((summary[total_fees] / summary[total_spend]) * 100).round(2) # 添加业务标签基于总消费额划分客户等级 summary[customer_tier] pd.cut( summary[total_spend], bins[0, 5000, 20000, float(inf)], labels[Bronze, Silver, Gold] ) results[summary] summary except Exception as e: print(fAnalysis 6 failed: {e}) results[summary] pd.DataFrame() # Analysis 7: Advanced custom aggregation with multiple conditions try: def risk_metrics(series): high_value_threshold 300 low_value_threshold 50 return pd.Series({ high_value_count: (series high_value_threshold).sum(), high_value_pct: ((series high_value_threshold).sum() / len(series) * 100).round(1), low_value_count: (series low_value_threshold).sum(), regular_avg: series[(series low_value_threshold) (series high_value_threshold)].mean() }) risk_analysis self.df.groupby(customer_id)[amount].apply(risk_metrics) results[risk_analysis] risk_analysis except Exception as e: print(fAnalysis 7 failed: {e}) results[risk_analysis] pd.DataFrame() return results # 使用示例 if __name__ __main__: # 加载你的数据 # df pd.read_parquet(path/to/production/data.parquet) # 或使用我们上面模拟的数据 analyzer BankTransactionAnalyzer(df_raw) all_results analyzer.run_all_analyses() # 打印关键结果 print( Executive Summary ) print(all_results[summary]) print(\n Risk Segmentation (Top 5 Customers) ) print(all_results[risk_analysis].head())这个类的设计哲学是一切皆可封装、一切皆可重试、一切皆有日志。它不是一个脚本而是一个服务组件。你可以把它打包成Docker镜像通过Airflow定时调度也可以作为API的后端接收HTTP请求返回JSON结果。run_all_analyses()方法保证了所有分析的原子性——要么全部成功要么在日志里清晰记录哪一步失败不影响其他步骤。这才是生产环境该有的样子。4.3 性能压测与调优从“能跑通”到“跑得稳”的必经之路写完代码只是开始让它在生产环境稳定运行需要严格的压测。我们团队的标准流程如下第一步基准测试Baselineimport time # 测试原始数据10万行 start time.time() analyzer BankTransactionAnalyzer(df_raw) results analyzer.run_all_analyses() end time.time() print(f10万行耗时: {end - start:.2f} 秒) # 目标 15秒第二步压力测试Stress Test# 模拟10倍数据量100万行 df_large pd.concat([df_raw] * 10, ignore_indexTrue) # 重置索引和日期避免重复 df_large[date] pd.date_range(start2024-01-01, periodslen(df_large), freqT) df_large df_large.set_index(date) start time.time() analyzer_large BankTransactionAnalyzer(df_large) results_large analyzer_large.run_all_analyses() end time.time() print(f100万行耗时: {end - start:.2f} 秒) # 目标 120秒第三步瓶颈定位与调优如果耗时超标我们用line_profiler精准定位pip install line_profiler kernprof -l -v your_script.py最常见的瓶颈及解决方案瓶颈位置表现解决方案效果groupby键为object类型CPU占用高内存暴涨df[col] df[col].astype(category)内存降60%速度提3倍rolling()未设min_periods开头大量NaN后续计算慢rolling(window7, min_periods1)首次计算时间降90%unstack()后数据稀疏内存占用激增改用sparseTrue参数unstack(sparseTrue)稀疏矩阵内存降80%apply()函数内含循环单核CPU 100%无法并行改用numpy向量化操作或swifter库速度提5-10倍第四步生产环境监控在Airflow DAG中我们为每个分析任务添加监控def check_result_quality(result_dict): Quality gate: ensure critical metrics meet business SLA. # 检查汇总结果是否为空 if result_dict[summary].empty: raise ValueError(Executive Summary is empty!) # 检查风险分析中高价值交易占比是否在合理区间0-100% if not (0 result_dict[risk_analysis][high_value_pct].min() 100): raise ValueError(High-value percentage out of bounds!) # 检查滚动平均值是否有足够多的有效值95% if result_dict[rolling_analysis][rolling_7day_avg].isna().mean() 0.05: raise ValueError(Too many NaN in rolling average!) # 在DAG中调用 check_result_quality(all_results)这确保了任何一次计算只要结果不符合业务常识