1. 项目概述为什么多维聚合不是“加个groupby”就能搞定的事我在银行数据平台组干了八年从最早用SQL写几十行嵌套子查询做客户分层到现在带团队设计日均处理20亿条交易的实时聚合管道。说实话刚看到“多维聚合”这个词时我也以为就是GROUP BY region, product, category再套几个SUM()和AVG()——直到某天凌晨三点风控系统因为一个没处理好的滚动窗口计算把三家正常商户标成了高风险触发了自动冻结流程。那次事故后我们复盘发现90%的数据问题不来自算法逻辑错误而源于对聚合行为边界的误判。比如你用rolling(window7).mean()算客户周均消费但没意识到pandas默认会按原始索引顺序滚动而真实交易数据是按时间戳排序的又比如用unstack()生成区域-产品矩阵时漏掉了fill_value0参数导致下游BI工具把缺失值当空字符串处理最终报表里出现大量NaN被当成0参与求和……这些坑文档里不会写教程里很少提但每个在生产环境跑过聚合任务的人都踩过。这篇文章讲的就是我在银行、保险、支付机构真实项目中反复验证过的多维聚合实战框架。它不讲agg()函数有多少种写法而是聚焦五个必须闭环的问题第一如何让一次聚合同时输出均值、中位数、极差且结果结构能直接喂给BI系统第二当业务规则要求“近30天大额交易占比超40%的客户进入人工审核队列”怎么把这种带条件分支的逻辑安全地封装进聚合第三滚动窗口计算时为什么前N行一定是NaN这个空缺该补还是该删补的话用什么策略最稳第四累计求和看似简单但当你需要“每个客户从开户日起的累计消费”而数据里混着不同开户时间的客户怎么避免跨客户污染第五当老板要你出一张“各城市TOP5商户类型销售额对比表”背后涉及的多级分组透视缺失值填充哪一步出错都会让整张表失去业务意义。关键词里的“Towards AI”不是随便贴的标签——它代表一种拒绝玩具数据的务实态度。文中的所有代码都来自我经手的三个真实场景某股份制银行信用卡中心的欺诈识别模型特征工程、某保险集团车险理赔的地域风险热力图生成、某跨境支付平台的商户分层运营看板。它们共同的特点是数据量在千万到十亿级、字段含业务强约束比如手续费率必须在0.5%-3.5%之间、结果要直连监管报送系统。所以你看不到df pd.read_csv(toy_data.csv)这种写法取而代之的是明确标注数据来源、采样逻辑、边界校验的完整链路。如果你正在为报表延迟、指标口径不一致、或者模型特征漂移发愁这篇内容可能比你读十篇论文都管用。2. 核心思路拆解为什么生产环境必须放弃“单点思维”2.1 聚合的本质是信息压缩而压缩必然丢失细节很多人把groupby().agg()当成数据整理工具其实它是有损压缩算法。举个例子你对1000笔餐饮交易按商户类别分组计算mean(amount)得到55.10元。这个数字本身没问题但它抹掉了三类关键信息一是交易金额的分布形态是集中在50-60元的窄峰还是20元和90元两极分化二是时间序列特征这1000笔是均匀分布在30天内还是集中在周末爆发三是异常值影响如果其中一笔是999元的婚宴消费均值会被拉高多少。在分析型场景里这些被压缩掉的信息恰恰是业务决策的依据。所以我在银行做的第一件事就是强制所有聚合操作必须配套三重验证机制分布验证对同一分组必须同时输出mean、median、std当|mean-median|/std 0.8时自动标记该分组需人工核查时序验证对含时间字段的数据聚合前先用df.sort_values(trans_time).reset_index(dropTrue)确保顺序再检查rolling(window7)的首尾索引是否连续边界验证对业务敏感字段如手续费聚合后立即执行result[fee_rate].between(0.005, 0.035).all()断言失败则中断流程并告警。这种思路直接改变了我们的代码结构。不再写df.groupby(cat)[amt].mean()而是封装成safe_agg(df, group_cols[cat], agg_dict{amt: [mean,median,std]}, validators[fee_rate_validator])。函数内部自动完成排序、验证、异常处理调用方只关心“我要什么结果”不用操心“怎么不出错”。2.2 多维聚合的真正难点在于维度间的语义冲突看原文示例里groupby([region,product])[revenue].mean().unstack()表面是语法问题实则是业务语义冲突。假设“华北区”有“手机”和“配件”两个产品线“华南区”只有“手机”那么unstack()后的结果里“华南-配件”单元格是NaN。但业务上这代表“华南区不销售配件”还是“数据采集遗漏”如果是前者填0合理如果是后者填0会导致后续计算如区域总销售额严重失真。我在某保险项目就遇到过类似问题车险保单按“省份车型”分组统计赔付率新疆某地市因新能源车保有量低所有车型分组都是空值。当时BI团队直接填0结果整个西北区的新能源车赔付率被拉低37%差点导致定价模型误判。解决方案是建立维度语义字典。对每个分组字段明确定义cardinality基数该维度实际存在的合法组合数如全国34个省级行政区但业务只覆盖28个sparsity_rule稀疏规则空值代表“无业务”还是“数据缺失”fill_strategy填充策略0、np.nan、N/A或业务默认值如“未覆盖地区”填-1。代码层面unstack()前必须调用validate_dimension_compatibility(df, dims[region,product], semantic_dictSEMANTIC_DICT)该函数会检查当前数据中实际出现的组合数是否符合cardinality预期并根据sparsity_rule决定是否允许fill_value参数生效。这种设计让技术实现和业务规则彻底解耦——当市场部新增一个销售区域时只需更新SEMANTIC_DICT所有聚合逻辑自动适配。2.3 滚动与扩展窗口不是时间函数而是状态机原文把rolling()和expanding()归为时间序列操作这在教学场景没问题但在生产环境会埋雷。关键区别在于滚动窗口是局部状态扩展窗口是全局状态。以滚动7日均值为例它的计算只依赖当前行及前6行内存占用恒定而扩展窗口需要从数据起点累积计算当处理10亿行数据时内存会随行数线性增长。更危险的是状态一致性——如果数据按customer_id分组后做滚动计算但分组内时间戳不连续比如客户A在1月1日和1月10日有交易中间缺8天rolling(window7)仍会强行计算最近7条记录的均值结果完全失真。我的做法是把窗口计算重构为显式状态机。核心原则所有窗口操作必须声明state_key状态标识和time_col时间基准。例如# 错误示范隐式状态无法处理时间断点 df.groupby(customer_id)[amount].rolling(window7).mean() # 正确实践显式状态机自动处理时间断点 def rolling_window_stateful(df, state_keycustomer_id, time_coltrans_time, window_days7, agg_funcmean): # 1. 按state_key分组每组内按time_col严格排序 df_sorted df.sort_values([state_key, time_col]) # 2. 计算时间差标记断点相邻行时间差window_days视为新周期 df_sorted[time_diff] df_sorted.groupby(state_key)[time_col].diff().dt.days df_sorted[is_break] (df_sorted[time_diff] window_days) | df_sorted[time_diff].isna() # 3. 为每个连续周期分配唯一id df_sorted[cycle_id] df_sorted.groupby(state_key)[is_break].cumsum() # 4. 在每个cycle_id内执行滚动计算 result df_sorted.groupby([customer_id, cycle_id])[amount].rolling( windowwindow_days, min_periods1).agg(agg_func).reset_index(level[0,1], dropTrue) return result这个函数会自动检测时间断点对每个连续交易周期独立计算滚动均值。当遇到1月1日和1月10日的交易时它会识别出中间8天断点将两笔交易划入不同cycle_id从而避免跨周期污染。虽然代码变长了但换来的是可审计、可复现、可解释的结果——这正是生产环境的核心诉求。3. 实操细节解析那些文档里绝不会写的硬核技巧3.1 多重聚合的列结构陷阱与平铺方案原文示例中result df.groupby(merchant_category).agg({transaction_amount: [mean,median], processing_fee: [min,max]})输出的层级列MultiIndex看似优雅但在实际工程中会引发连锁问题。最典型的是当你要把结果存入数据库时PostgreSQL不支持嵌套列名MySQL的列名长度限制64字符会让(transaction_amount, mean)这种元组直接报错更麻烦的是下游BI工具如Tableau导入时会把层级列识别为transaction_amount_mean和transaction_amount_median但某些版本会错误地将下划线当作分隔符导致字段名被截断。我的解决方案是三级平铺策略根据下游消费方自动适配Level 1基础平铺用result.columns.map(_.join)生成transaction_amount_mean格式适用于大多数SQL数据库Level 2语义平铺对业务敏感字段用业务缩写替代全称如trans_amt_mean、proc_fee_min并建立映射表供审计Level 3动态平铺当列名超长时如(customer_lifetime_value, rolling_90day_std)采用哈希截断f{col[0][:10]}_{hashlib.md5(str(col).encode()).hexdigest()[:6]}同时生成column_mapping.json记录原始名与别名对应关系。关键代码如下def flatten_columns(result_df, strategysemantic, max_length30): strategy: basic | semantic | dynamic max_length: 列名最大长度仅dynamic策略使用 if strategy basic: new_cols [_.join(col).strip() for col in result_df.columns] elif strategy semantic: # 定义业务缩写映射 abbrev_map { transaction_amount: trans_amt, processing_fee: proc_fee, mean: avg, median: med, min: min_val, max: max_val } new_cols [] for col in result_df.columns: parts list(col) for i, part in enumerate(parts): if part in abbrev_map: parts[i] abbrev_map[part] new_cols.append(_.join(parts)) else: # dynamic new_cols [] mapping {} for col in result_df.columns: col_str _.join(col) if len(col_str) max_length: new_cols.append(col_str) else: # 生成哈希别名 hash_key hashlib.md5(col_str.encode()).hexdigest()[:6] alias f{col_str[:max_length-7]}_{hash_key} new_cols.append(alias) mapping[alias] col_str # 将映射表存入DataFrame属性便于后续审计 result_df._column_mapping mapping result_df.columns new_cols return result_df # 使用示例 result df.groupby(merchant_category).agg({ transaction_amount: [mean,median], processing_fee: [min,max] }) flattened flatten_columns(result, strategysemantic) print(flattened.columns.tolist()) # 输出: [trans_amt_avg, trans_amt_med, proc_fee_min_val, proc_fee_max_val]提示永远不要在生产代码中用result.reset_index()直接展平MultiIndex它会把分组键变成普通列破坏数据血缘关系。正确做法是先flatten_columns()再reset_index()确保分组键始终作为索引存在。3.2 自定义聚合函数的性能陷阱与编译优化原文用lambda x: x.max() - x.min()演示自定义聚合这在小数据集上很优雅但放到千万级数据时会成为性能瓶颈。原因在于pandas对lambda函数无法进行向量化优化每次调用都要触发Python解释器开销。我在某支付项目中实测过对1000万行数据计算极差lambda版本耗时42秒而向量化版本仅需1.8秒。优化路径分三层第一层向量化用numpy原生函数替代Python循环。x.max()-x.min()直接写成np.ptp(x)peak-to-peak这是numpy内置的向量化极差计算第二层JIT编译对复杂逻辑如加权平均用numba.jit编译。注意numba不支持pandas Series需传入x.values第三层Cython加速对极致性能要求场景如实时风控用Cython重写核心逻辑。以加权平均为例原文的weighted_average函数可优化为from numba import jit import numpy as np jit(nopythonTrue) def weighted_average_numba(values, weights): Numba编译的加权平均比纯Python快20倍 if len(values) 2: return np.mean(values) weighted_sum 0.0 weight_sum 0.0 for i in range(len(values)): weighted_sum values[i] * weights[i] weight_sum weights[i] return weighted_sum / weight_sum def weighted_average_optimized(series): 优化版加权平均自动选择最优实现 if len(series) 1000: # 小数据用原生numpy weights np.linspace(0.5, 1.5, len(series)) return np.average(series, weightsweights) else: # 大数据用numba weights np.linspace(0.5, 1.5, len(series)) return weighted_average_numba(series.values, weights) # 性能对比测试 %timeit df.groupby(merchant_category)[transaction_amount].apply(weighted_average_optimized) # 原始版本: 12.4 s per loop # 优化版本: 580 ms per loop注意numba函数必须用jit(nopythonTrue)强制编译否则会回退到Python解释模式。且传入参数必须是numpy数组不能是pandas Series——这是很多教程忽略的关键点。3.3 滚动窗口的NaN处理不是补0那么简单原文提到“前两行显示NaN值是因为3日窗口需要三个数据点”但没说清楚为什么必须是NaN以及如何科学处理它。在风控场景中用fillna(methodffill)向前填充会制造虚假趋势比如把第1天的异常高交易额延续到第2、3天而dropna()又会丢失关键起始点。我的经验是NaN是系统发出的健康信号应该分级响应而非粗暴填充。具体策略Level 1诊断层在计算滚动窗口后立即统计NaN比例。若某分组NaN_ratio 0.3触发数据质量告警Level 2业务层对不同业务场景配置差异化填充策略欺诈检测用bfill()向后填充假设首日异常是设备故障次日数据可信营收预测用interpolate(methodlinear)线性插值假设营收呈平滑变化合规报送严格保留NaN并在元数据中标记data_gap_flagTrueLevel 3工程层用min_periods参数控制最小有效点数。例如rolling(window7, min_periods4)表示只要有4个有效点就计算均值避免全空。代码实现def robust_rolling(df, time_col, value_col, window_days7, fill_strategybusiness, min_periods1): 健壮的滚动计算内置NaN分级处理 fill_strategy: none | forward | backward | interpolate | business # 先按时间排序并设置索引 df_sorted df.sort_values(time_col).set_index(time_col) # 执行滚动计算 rolling_result df_sorted[value_col].rolling( windowf{window_days}D, # 用字符串形式支持日期偏移 min_periodsmin_periods ).mean() # 分级填充 if fill_strategy none: pass elif fill_strategy forward: rolling_result rolling_result.fillna(methodffill) elif fill_strategy backward: rolling_result rolling_result.fillna(methodbfill) elif fill_strategy interpolate: rolling_result rolling_result.interpolate(methodlinear) elif fill_strategy business: # 业务策略欺诈检测用bfill营收用interpolate if fraud in value_col.lower(): rolling_result rolling_result.fillna(methodbfill) else: rolling_result rolling_result.interpolate(methodlinear) return rolling_result.reset_index(namef{value_col}_rolling_{window_days}d) # 使用示例欺诈检测场景 fraud_rolling robust_rolling( df_transactions, time_coldate, value_colamount, window_days7, fill_strategybusiness )4. 完整实操流程从原始交易数据到高管决策看板4.1 数据准备与质量基线校验所有聚合操作前必须完成三道质量防火墙。这不是可选项而是我所在团队的上线红线。以信用卡交易数据为例原始数据来自核心银行系统包含trans_id,customer_id,trans_time,merchant_category,amount,fee等字段。防火墙设计如下防火墙1Schema一致性校验检查字段类型、空值率、唯一性约束是否符合数据字典。例如fee字段必须是数值型且fee/amount比率应在[0.005, 0.035]区间def validate_schema(df): issues [] # 类型校验 if not pd.api.types.is_numeric_dtype(df[amount]): issues.append(amount字段非数值型) if not pd.api.types.is_datetime64_any_dtype(df[trans_time]): issues.append(trans_time字段非时间类型) # 业务规则校验 fee_rate df[fee] / df[amount] invalid_rates fee_rate[~fee_rate.between(0.005, 0.035)] if len(invalid_rates) 0: issues.append(ffee/amount比率异常共{len(invalid_rates)}条记录超出[0.5%,3.5%]) # 空值校验 null_pct df.isnull().mean() * 100 high_null_cols null_pct[null_pct 5].index.tolist() if high_null_cols: issues.append(f空值率5%的字段{high_null_cols}) if issues: raise ValueError(fSchema校验失败{; .join(issues)}) return True # 执行校验 validate_schema(df_transactions)防火墙2时间序列完整性校验检查时间戳是否连续、是否存在重复或乱序。这对滚动窗口至关重要def validate_time_series(df, time_coltrans_time, freqD): freq: D日频, H小时频, T分钟频 df_sorted df.sort_values(time_col).copy() # 检查重复时间戳 dup_times df_sorted.duplicated(subset[time_col], keepFalse) if dup_times.any(): issues df_sorted[dup_times][time_col].value_counts() raise ValueError(f时间戳重复{issues.to_dict()}) # 检查时间断点按指定频率 expected_freq pd.tseries.frequencies.to_offset(freq) time_diffs df_sorted[time_col].diff() gaps time_diffs[time_diffs expected_freq] if len(gaps) 0: print(f警告检测到{len(gaps)}处时间断点最大间隔{gaps.max()}) # 断点不阻断流程但记录日志供分析 # 检查乱序时间戳倒流 is_reverse (df_sorted[time_col].diff().dt.total_seconds() 0).any() if is_reverse: raise ValueError(检测到时间戳倒流请检查数据源) return df_sorted # 执行校验 df_clean validate_time_series(df_transactions, time_coldate, freqD)防火墙3业务逻辑合理性校验基于领域知识检查数据是否符合常识。例如信用卡交易中单笔金额超过5万元需人工审核若数据中存在大量5万交易而无审核标记则数据可能被篡改def validate_business_logic(df): issues [] # 大额交易检查 high_value_threshold 50000 high_value_count (df[amount] high_value_threshold).sum() if high_value_count 0: # 检查是否有对应的审核标记字段 if review_flag not in df.columns: issues.append(f存在{high_value_count}笔超{high_value_threshold}元交易但缺少review_flag字段) else: unreviewed_high df[(df[amount] high_value_threshold) (df[review_flag] ! approved)] if len(unreviewed_high) 0: issues.append(f存在{len(unreviewed_high)}笔超阈值未审核交易) # 商户类别一致性检查 valid_categories {Groceries, Dining, Travel, Retail, Electronics} invalid_cats set(df[category].unique()) - valid_categories if invalid_cats: issues.append(f商户类别异常{invalid_cats}) if issues: raise ValueError(f业务逻辑校验失败{; .join(issues)}) return True validate_business_logic(df_clean)实操心得这三道防火墙必须封装成独立模块在ETL流水线每个关键节点后自动执行。我们用Airflow调度时每个聚合任务前都插入schema_validation time_validation business_validation三个Operator任何一道失败都会中断流程并发送企业微信告警。这看似增加开发成本但避免了90%的线上事故。4.2 七步聚合流水线从原始数据到决策看板基于前述校验构建端到端聚合流水线。每一步都对应一个明确的业务目标且结果可独立验证Step 1基础分组聚合回答“谁买了什么”按客户商户类别计算基础统计为后续分析提供原子指标# 计算每个客户在每个商户类别的交易统计 base_agg df_clean.groupby([customer_id, category]).agg({ amount: [count, sum, mean, std, min, max], fee: [sum, mean] }).round(2) # 平铺列名并重命名 base_agg.columns [_.join(col).strip() for col in base_agg.columns] base_agg base_agg.rename(columns{ amount_count: trans_count, amount_sum: total_spend, amount_mean: avg_trans, amount_std: trans_std, fee_sum: total_fee, fee_mean: avg_fee }) # 添加衍生指标 base_agg[fee_rate] (base_agg[total_fee] / base_agg[total_spend] * 100).round(2) base_agg[high_value_ratio] ( (df_clean.groupby([customer_id, category])[amount] .apply(lambda x: (x 300).sum() / len(x) * 100) .round(1) ).values # 注意这里需确保顺序一致用.values避免索引对齐问题Step 2时间维度增强回答“何时发生”为每笔交易添加时间特征支撑滚动和扩展计算# 添加时间特征列 df_enhanced df_clean.copy() df_enhanced[year] df_enhanced[date].dt.year df_enhanced[month] df_enhanced[date].dt.month df_enhanced[week] df_enhanced[date].dt.isocalendar().week df_enhanced[day_of_week] df_enhanced[date].dt.dayofweek # 0周一 df_enhanced[is_weekend] df_enhanced[day_of_week].isin([5,6]) # 计算客户生命周期阶段按首次交易时间 first_trans df_enhanced.groupby(customer_id)[date].min().rename(first_trans_date) df_enhanced df_enhanced.merge(first_trans, oncustomer_id) df_enhanced[days_since_first] (df_enhanced[date] - df_enhanced[first_trans_date]).dt.days df_enhanced[lifecycle_stage] pd.cut( df_enhanced[days_since_first], bins[0, 30, 90, 365, float(inf)], labels[New, Active, Mature, Loyal] )Step 3滚动窗口计算回答“趋势如何”计算客户级7日滚动均值用于实时监控# 按客户分组对交易金额计算7日滚动均值 rolling_7d df_enhanced.sort_values([customer_id, date]).groupby(customer_id)[amount].rolling( window7D, # 使用日期偏移自动处理不规则时间间隔 min_periods3 # 至少3个点才计算避免噪声 ).mean().reset_index(level0, dropTrue) # 合并回原数据 df_enhanced[rolling_7d_avg] rolling_7d.values # 业务化处理NaN对新客户首次交易7天内用首笔交易额填充 first_dates df_enhanced.groupby(customer_id)[date].min() df_enhanced[days_from_first] (df_enhanced[date] - first_dates).dt.days df_enhanced.loc[df_enhanced[days_from_first] 7, rolling_7d_avg] df_enhanced[amount]Step 4扩展窗口计算回答“累计如何”计算客户累计消费用于LTV客户终身价值分析# 按客户分组计算累计消费 cumulative_spend df_enhanced.sort_values([customer_id, date]).groupby(customer_id)[amount].expanding().sum() df_enhanced[cumulative_spend] cumulative_spend.reset_index(level0, dropTrue).values # 计算累计交易笔数 cumulative_count df_enhanced.sort_values([customer_id, date]).groupby(customer_id)[amount].expanding().count() df_enhanced[cumulative_count] cumulative_count.reset_index(level0, dropTrue).valuesStep 5多维交叉分析回答“结构如何”生成客户-商户类别矩阵支持交叉销售分析# 创建交叉表客户为行商户类别为列值为平均交易额 crosstab_avg df_enhanced.groupby([customer_id, category])[amount].mean().unstack(fill_value0) # 添加行列总计 crosstab_avg[Total] crosstab_avg.sum(axis1) crosstab_avg.loc[Total] crosstab_avg.sum(axis0) # 计算占比矩阵每行百分比 crosstab_pct crosstab_avg.div(crosstab_avg[Total], axis0).multiply(100).round(1) crosstab_pct crosstab_pct.drop(Total, axis1) # 移除总计列只保留占比Step 6风险分层建模回答“风险如何”基于业务规则对客户进行风险评分def risk_scoring(row): 客户风险评分函数返回0-100分 score 0 # 交易频率风险7日内交易10笔扣分 recent_trans df_enhanced[ (df_enhanced[customer_id] row.name) (df_enhanced[date] row[date] - pd.Timedelta(days7)) ] if len(recent_trans) 10: score 20 # 金额波动风险标准差/均值1.5扣分 if row[trans_std] and row[avg_trans] 0: cv row[trans_std] / row[avg_trans] if cv 1.5: score 30 # 大额交易集中度300元交易占比50%扣分 if row[high_value_ratio] 50: score 25 # 周末交易占比70%扣分可能为套现 weekend_trans df_enhanced[ (df_enhanced[customer_id] row.name) df_enhanced[is_weekend] ] if len(weekend_trans) 0: weekend_ratio len(weekend_trans) / len(recent_trans) * 100 if weekend_ratio 70: score 25 return min(score, 100) # 最高100分 # 应用风险评分 risk_scores base_agg.apply(risk_scoring, axis1) base_agg[risk_score] risk_scores base_agg[risk_level] pd.cut( base_agg[risk_score], bins[-1, 30, 70, 100], labels[Low, Medium, High] )Step 7高管摘要生成回答“结论如何”汇总关键指标生成一页纸决策报告def generate_exec_summary(df_agg): 生成高管摘要 summary {} # 整体业绩 summary[total_customers] df_agg.index.get_level_values(customer_id).nunique() summary[total_transactions] df_agg[trans_count].sum() summary[total_revenue] df_agg[total_spend].sum() summary[avg_customer_value] (summary[total_revenue] / summary[total_customers]).round(2) # 风险分布 risk_dist df_agg[risk_level].value_counts(normalizeTrue).multiply(100).round(1) summary[high_risk_pct] risk_dist.get(High, 0) summary[medium_risk_pct] risk_dist.get(Medium, 0) summary[low_risk_pct] risk_dist.get(Low, 0) # 高价值客户top 10%消费额 top_customers df_agg.nlargest(int(len(df_agg)*0.1), total_spend) summary[top10_percent_contribution] (top_customers[total_spend].sum() / summary[total_revenue] * 100).round(1) # 增长洞察 recent_data df_enhanced[df_enhanced[date] df_enhanced[date].max() - pd.Timedelta(days30)] recent_revenue recent_data[amount].sum() prior_data df_enhanced[ (df_enhanced[date] df_enhanced[date].max() - pd.Timedelta(days60)) (df_enhanced[date] df_enhanced[date].max() - pd.Timedelta(days30)) ] prior_revenue prior_data[amount].sum() summary[momo_growth] ((recent_revenue - prior_revenue) / prior_revenue * 100).round(1) if prior_revenue 0 else 0 return pd.Series(summary) exec_summary generate_exec_summary(base_agg) print( 高管决策摘要 ) for k, v in exec_summary.items(): print(f{k}: {v})4.3 结果交付与血缘追踪所有聚合结果必须附带数据血缘元数据这是金融行业合规审计的硬性要求。我们用JSON Schema定义元数据结构import json from datetime import datetime def generate_metadata(df_source, agg_config, execution_timeNone): 生成聚合结果元数据 if execution_time is None: execution_time datetime.now() metadata {