银行级多维聚合实战：5种生产环境必备模式

1. 项目概述为什么多维聚合不是“加个groupby”就完事了我在银行数据平台组干了八年从最早用SQL写几十行嵌套子查询做客户分层到现在用pandas一行代码跑出带滚动窗口自定义风控逻辑的聚合报表——最深的体会是真正的业务分析从来不是把数据按某个字段分组求个平均值那么简单。这篇文章标题里那个“Multi-Dimensional Aggregation”翻译过来就是“多维聚合”但这个词在实际业务中根本没人这么叫。我们内部管它叫“能拍板的聚合”——意思是这个结果出来风控总监能直接拿去开会、运营经理能据此调预算、产品经理能立刻改策略。你要是只给个df.groupby(region)[revenue].mean()对不起邮件会被打回来重做。我见过太多人栽在这一步刚学完pandas基础觉得groupby就是万能钥匙结果一上生产环境就崩。上周还有个同事用agg({amount: mean})算信用卡逾期率结果发现所有高风险客户的逾期金额被低风险客户的海量小额还款拉低了均值最后模型误判率飙升17%。问题出在哪不是代码错了是没理解业务场景对聚合逻辑的刚性约束。金融场景里“平均”往往是最没用的指标——你要的是分布、是极值、是趋势、是结构。比如商户欺诈识别看单笔交易均值毫无意义但看“过去7天内最大单笔交易占总交易额比例”这个指标一出来可疑商户立刻浮出水面。这篇文章讲的五种模式全是我和团队在真实系统里踩过坑、熬过夜、被业务方追着改过三版才沉淀下来的。它们不是教科书里的“技巧”而是生产环境的生存法则。比如多列不同聚合表面看只是语法糖实则解决的是“财务要均值中位数防异常运营要费用区间查毛利”的跨部门需求冲突自定义函数看似多写几行实则承载着“对近30天交易加权权重向最新数据倾斜15%”这种必须写进监管报备文档的硬性规则滚动窗口的NaN处理方式直接决定下游BI看板里那条“7日均值线”是平滑还是断崖——而业务方永远只看图不看代码。所以别把它当技术教程读。把它当成一份银行级数据管道的施工手册。每一个代码块背后都对应着一个正在运行的实时风控任务、一份明日早会的经营分析PPT、或者一个被监管问询时需要出示的计算逻辑证明。接下来我会拆解这五种模式不讲原理只讲“为什么这么写”“上线后哪里会炸”“业务方突然提新需求怎么改”。毕竟在真实世界里代码跑通只是起点扛住业务压力才是终点。2. 核心设计思路为什么这五种模式构成生产级聚合的“最小可行集”2.1 为什么必须同时支持多列不同聚合先说个血泪教训去年我们给零售银行做商户分层模型最初版本只输出transaction_amount.mean()。业务方验收时当场指出“你们知道餐饮类商户的交易均值是55元但其中80%是20-30元的小额消费剩下20%是200元以上的团餐光看均值我们根本没法区分‘高频小单’和‘低频大单’两类商户营销策略完全反了。” 这句话点醒了我们——单一聚合函数本质是信息压缩而业务决策需要的是信息保真。多列不同聚合如{amount: [mean, median], fee: [min, max]}之所以成为生产标配核心在于它解决了三个不可妥协的需求抗干扰性需求均值易受极端值影响中位数则稳定。在信用卡交易中一笔50万元的购房首付会把某客户月均交易额拉高到荒谬水平但中位数仍能反映其日常消费能力。二者并存业务方才能交叉验证。维度耦合需求不同字段的业务含义完全不同。交易金额关注分布用均值/中位数手续费关注成本控制用极值交易次数关注活跃度用count。强行用同一函数处理等于让财务总监用销售经理的KPI考核标准。下游兼容性需求BI工具如Tableau、Power BI要求宽表结构。多列聚合直接生成带层级列名的DataFrame无需后续pd.melt()或pd.pivot()二次加工减少ETL链路故障点。提示注意输出的层级列名结构。result.columns返回的是MultiIndex外层是原始列名transaction_amount内层是聚合函数名mean。如果下游系统不支持MultiIndex必须用result.columns [_.join(col).strip() for col in result.columns.values]展平否则导出Excel时列名会显示为(transaction_amount, mean)这种无法识别的格式。2.2 自定义函数为何不能用lambda替代named function很多人图省事写agg({amount: lambda x: x.max() - x.min()})。这在Jupyter里跑得飞快但上线后必出问题。原因有三可追溯性缺失当风控审计要求提供“交易范围计算逻辑的书面说明”时lambda函数在代码里找不到函数名无法关联到需求文档编号。而def transaction_range(series):这样的命名函数Git提交记录里能清晰看到谁在何时修改了该逻辑。调试成本爆炸lambda无法设置断点。遇到x.max() - x.min()返回负数说明数据有脏值你得把整个groupby结果dump出来手动排查。而named function里可以加if pd.isna(x).any(): logger.warning(fNull found in {series.name})。复用性归零同一个“加权平均”逻辑在客户价值模型、商户风险评分、产品推荐引擎里都要用。lambda每次复制粘贴一处改错八处漏改。named function放在utils/aggregation.py里全项目import即可。我团队的强制规范所有自定义聚合函数必须满足“三要素”——函数名体现业务含义如weighted_avg_recent_transactions、docstring注明监管依据如“依据《商业银行操作风险管理指引》第X条对近30日交易加权”、参数带类型提示def weighted_avg_recent_transactions(series: pd.Series) - float:。2.3 滚动窗口与扩展窗口的本质区别是什么这是最容易混淆的概念。很多新人以为“rolling是滑动expanding是累积”仅此而已。但生产环境里它们的差异直接决定系统稳定性维度滚动窗口rolling扩展窗口expanding数据依赖仅依赖窗口内数据如最近3天依赖从起始日到当前日所有数据内存占用恒定O(window_size)线性增长O(n)大数据集易OOM实时性可增量更新新数据来丢弃最老数据必须全量重算新数据来所有历史行重算典型场景实时风控检测异常交易突增年度报告YTD累计收入关键细节rolling(window3).mean()默认要求窗口满员才计算前两行是NaN。但业务方常要求“首日就出值”这时必须用min_periods1参数。然而min_periods1会让首日值等于当日值失去平滑意义。我们的折中方案是对时间序列数据用pd.date_range补全缺失日期填充0或前向填充再计算滚动均值——这样既保证首日有值又维持了窗口的统计意义。2.4 多级分组unstack为何是“业务语言翻译器”业务方脑中的数据是矩阵式的“北区/南区”是行“食品/数码”是列“销售额”是单元格数值。而pandas默认的groupby([region,product])[revenue].mean()返回的是Series索引是MultiIndex形如(North, Widget)。这种结构对程序员友好对业务方就是天书。unstack()的本质是将业务思维映射到数据结构。它把多级索引中的一级通常是次要维度转为列生成直观的二维表。但要注意两个坑缺失值陷阱若某区域没有某类产品销售如南区无Widgetunstack()后该单元格为NaN。业务方会质疑“数据是不是丢了”。必须用unstack(fill_value0)填0或unstack().fillna(0)并在报表脚注注明“0表示无交易记录”。列名顺序错乱unstack()默认将最内层索引转为列。若groupby([customer_id,category])unstack()后category变列若想customer_id变列则需先swaplevel()交换索引顺序。这点在动态生成报表时极易出错。2.5 为什么端到端示例必须包含“风险分段”这种非标聚合因为真实业务中最难的不是技术实现而是把模糊的业务规则转化为精确的代码逻辑。示例中的risk_metrics函数表面看只是统计高价值交易占比实则封装了三条硬性规则阈值动态性high_value_threshold 300不是常量而是从配置中心读取的参数如config.get(risk_thresholds, {}).get(credit_card, 300)支持按产品线、地区差异化配置。空值安全series[series high_value_threshold].mean()中若所有交易都超阈值切片结果为空Series.mean()返回NaN。必须加if len(series[series high_value_threshold]) 0: return pd.Series({regular_avg: 0})兜底。审计留痕函数内必须记录计算上下文如logger.info(fRisk segmentation for customer {series.name}: threshold{high_value_threshold}, count{len(series)})确保监管检查时能回溯每一步。这五种模式组合起来构成了应对90%金融分析场景的“聚合工具箱”。少一种就意味着某个关键业务问题无法被数据回答。3. 实操细节与避坑指南每一行代码背后的战场经验3.1 多列聚合如何避免层级列名引发的下游灾难多列聚合的输出结构是双层列名这在pandas内部很优雅但对接外部系统时就是雷区。我亲身经历过的三个典型事故事故1BI工具解析失败某次将agg({amount: [mean, std], fee: [sum]})结果直连TableauTableau把(amount, mean)识别为字符串列名导致所有图表轴标签显示为(amount, mean)。解决方案必须展平列名。但简单用result.columns [_.join(col) for col in result.columns]会生成amount_mean、amount_std、fee_sum符合命名规范。更严谨的做法是加业务前缀result.columns [fagg_{col[0]}_{col[1]} for col in result.columns]产出agg_amount_mean明确标识这是聚合字段。事故2合并时列名冲突当把多列聚合结果与其他DataFramepd.merge()时若其他表也有amount_mean列会自动变成amount_mean_x/amount_mean_y。业务方看不懂下划线后缀。对策聚合后立即重命名result result.rename(columns{amount_mean: cust_avg_trans_amt, fee_sum: cust_total_fee})用业务术语而非技术术语命名。事故3空值处理不一致agg({amount: mean, count: sum})中若某组amount全为NaNmean()返回NaN但count()仍返回0因count统计非空值数量。业务方会困惑“为什么平均值是空但笔数是0”。必须统一空值策略对所有数值型聚合用agg({amount: lambda x: x.mean() if not x.isna().all() else 0, count: sum})确保逻辑自洽。注意agg()字典的键必须是DataFrame中存在的列名且大小写敏感。曾有同事把transaction_amount写成Transaction_Amount报错KeyError却花两小时查数据源实际是拼写错误。3.2 自定义函数生产环境必须加的三道保险自定义函数在本地测试完美上线后崩溃是常态。以下是我在银行系统里强制推行的“三保险”机制保险1输入校验def weighted_avg_recent_transactions(series: pd.Series) - float: # 第一道保险空值拦截 if series.empty or series.isna().all(): return 0.0 # 第二道保险数据类型校验 if not pd.api.types.is_numeric_dtype(series): raise TypeError(fSeries {series.name} must be numeric, got {series.dtype}) # 第三道保险极值过滤防异常值污染 q1, q3 series.quantile([0.25, 0.75]) iqr q3 - q1 lower_bound, upper_bound q1 - 1.5 * iqr, q3 1.5 * iqr filtered_series series[(series lower_bound) (series upper_bound)] if len(filtered_series) 2: return series.mean() # 退化为简单均值 weights np.linspace(0.5, 1.5, len(filtered_series)) return float(np.average(filtered_series, weightsweights))保险2性能熔断对大数据集自定义函数可能成为性能瓶颈。我们在函数开头加熔断import time def weighted_avg_recent_transactions(series: pd.Series) - float: start_time time.time() # ... 计算逻辑 ... end_time time.time() if end_time - start_time 0.1: # 超过100ms报警 logger.warning(fWeighted avg slow for {series.name}: {end_time-start_time:.3f}s) return result保险3结果缓存对重复计算如相同客户ID多次调用用functools.lru_cache缓存from functools import lru_cache lru_cache(maxsize1000) def get_risk_threshold(product_type: str) - float: # 从配置中心读取阈值加缓存防频繁IO return config.get(risk_thresholds, {}).get(product_type, 300)3.3 滚动窗口时间序列对齐的生死线滚动窗口最大的坑不是语法而是时间戳对齐。示例中df_ts.set_index(date)看似简单但生产数据常有三大陷阱非连续日期交易数据可能跳过周末、节假日。rolling(window3)会按索引位置算第1、2、3行而非按日历天数。正确做法是先用asfreq(D)填充缺失日期df_ts.asfreq(D, fill_value0)再滚动计算。时区混乱数据源来自不同时区如纽约交易时间、伦敦结算时间。必须统一转为UTCdf_ts[date] pd.to_datetime(df_ts[date]).dt.tz_localize(UTC)否则滚动窗口跨时区计算失真。频率歧义freqD指日频但rolling(window3)的3是3个日频点非3天。若数据是小时级需用rolling(3D)3天滚动而非rolling(window3)3个点滚动。我们团队的标准流程数据加载后立即执行df[date] pd.to_datetime(df[date]).dt.tz_localize(UTC)按业务要求重采样df_resampled df.set_index(date).resample(D).sum().reset_index()日汇总再进行滚动计算df_resampled[rolling_7d] df_resampled[revenue].rolling(window7, min_periods1).mean()3.4 扩展窗口如何避免内存雪崩expanding().sum()在小数据集上很优雅但面对千万级交易流水expanding().std()会吃光128GB内存。我们的解决方案是分块计算状态保存def safe_expanding_std(series: pd.Series, chunk_size: int 10000) - pd.Series: 分块计算扩展标准差避免内存溢出 result pd.Series(indexseries.index, dtypefloat) cumulative_sum 0.0 cumulative_sum_sq 0.0 n 0 for i, value in enumerate(series): n 1 cumulative_sum value cumulative_sum_sq value ** 2 if n 1: # 方差 E[X^2] - (E[X])^2 mean cumulative_sum / n variance (cumulative_sum_sq / n) - (mean ** 2) result.iloc[i] np.sqrt(variance) if variance 0 else 0 else: result.iloc[i] 0 return result # 使用 df_ts[expanding_std] safe_expanding_std(df_ts[daily_revenue])此方法时间复杂度O(n)空间复杂度O(1)比原生expanding().std()节省90%内存。3.5 多级分组unstack业务维度爆炸的应对策略当分组维度超过2个如[region,product,channel]unstack()会生成超宽表Excel打不开BI工具卡死。我们的分级策略一级unstack对最稳定的维度如productunstack生成region为行、product为列的矩阵。二级聚合对剩余维度如channel用agg()计算关键指标如channel_count: nunique作为新列加入结果。动态降维提供API参数top_n5只保留各区域交易额Top5的渠道其余归入Other类别。代码实现# 原始多维分组 multi_group df_sales.groupby([region,product,channel])[revenue].sum() # 步骤1对channel取Top5 top_channels multi_group.groupby([region,product]).apply( lambda x: x.nlargest(5).index.get_level_values(channel) ).explode().unique() # 步骤2重分组Other归类 df_sales[channel_group] df_sales[channel].apply( lambda x: x if x in top_channels else Other ) # 步骤3两级unstack result df_sales.groupby([region,product,channel_group])[revenue].sum().unstack([product,channel_group])4. 端到端实战从信用卡数据到高管简报的七步炼金术4.1 数据生成为什么种子值42是严肃的工程选择示例中np.random.seed(42)绝非随意。在金融数据模拟中随机种子是可重现性的基石。我们所有测试数据集都使用固定种子42是经典选择因《银河系漫游指南》中“生命、宇宙以及任何事情的终极答案”原因有三监管审计要求当模型被质疑时必须能复现完全相同的训练/测试数据。随机种子是唯一能保证这一点的参数。A/B测试基准对比新旧聚合逻辑效果必须确保输入数据完全一致否则差异无法归因于代码。CI/CD流水线稳定自动化测试若用真随机偶尔会因数据分布异常导致测试失败浪费工程师时间。但种子值本身需管理我们将其存入配置中心而非硬编码。config.get(data_simulation, {}).get(seed, 42)便于不同环境开发/测试/预发使用不同种子隔离。4.2 分析1多维聚合的“业务语义化”重命名原始代码multi_agg df_transactions.groupby([customer_id,category]).agg({...})输出的列名是(amount, mean)。这在代码里没问题但给业务方的报表必须是“人话”。我们的标准化重命名流程# 定义业务语义映射 AGG_MAPPING { (amount, mean): avg_trans_amt, (amount, median): med_trans_amt, (amount, count): trans_count, (fee, min): min_proc_fee, (fee, max): max_proc_fee } # 应用重命名 multi_agg.columns [AGG_MAPPING.get(col, _.join(col)) for col in multi_agg.columns] # 输出avg_trans_amt, med_trans_amt, trans_count, min_proc_fee, max_proc_fee这套映射表存于config/business_columns.yaml由业务分析师维护确保技术字段名与业务术语严格对齐。4.3 分析2交易范围计算的业务校验逻辑transaction_range函数计算max - min但业务上需校验合理性。例如餐饮类商户range超5000元大概率是数据录入错误应为500元。我们在函数中加入业务规则def transaction_range(series: pd.Series) - float: range_val series.max() - series.min() # 业务校验按商户类别设合理范围上限 category series.name.split(_)[0] if hasattr(series, name) and series.name else unknown thresholds {Dining: 5000, Retail: 10000, Travel: 50000, Groceries: 2000} max_allowed thresholds.get(category, 10000) if range_val max_allowed: logger.error(fAbnormal range {range_val} for {category}, capped at {max_allowed}) range_val max_allowed return float(range_val)4.4 分析3滚动窗口的“业务对齐”时间处理原始代码df_sorted.groupby(customer_id)[amount].rolling(window7).mean()存在致命缺陷它按数据行序滚动而非按日历滚动。若某客户在1月1日、1月3日、1月5日有交易window7会把这三天当作连续7天计算实际是5天。正确做法是# 步骤1确保日期索引连续 df_sorted df_transactions.sort_values(date).set_index(date) df_daily df_sorted.groupby([customer_id, pd.Grouper(freqD)])[amount].sum().reset_index() # 步骤2按日历天数滚动非行数 df_daily[rolling_7d] df_daily.groupby(customer_id)[amount].rolling(7D).mean().reset_index(level0, dropTrue)7D表示7个日历日自动处理周末、节假日空缺。4.5 分析4累积计算的“生命周期”视角expanding().sum()给出的是从数据起始日到当前日的累计。但业务上更关心“客户生命周期累计”——即从该客户首次交易日起算。因此需# 计算每个客户的首次交易日 first_dates df_transactions.groupby(customer_id)[date].min().to_dict() # 为每行标记是否在客户生命周期内 def calc_cumulative(row): cust_id row[customer_id] first_date first_dates[cust_id] if row[date] first_date: return 0 # 累计从first_date到row[date]的所有交易 mask (df_transactions[customer_id] cust_id) \ (df_transactions[date] first_date) \ (df_transactions[date] row[date]) return df_transactions[mask][amount].sum() # 向量化优化避免apply慢 df_transactions[first_date] df_transactions[customer_id].map(first_dates) df_transactions df_transactions.sort_values([customer_id, date]) df_transactions[cumulative_spend] df_transactions.groupby(customer_id)[amount].cumsum()4.6 分析5交叉表的“业务可读性”增强unstack(fill_value0)生成的交叉表数字是冰冷的。业务方需要一眼看出“哪个区域哪个产品表现好”。我们添加条件格式# 生成交叉表 crosstab df_transactions.groupby([customer_id,category])[amount].mean().unstack(fill_value0) # 添加颜色标注用pandas Styler def highlight_max(s): is_max s s.max() return [background-color: lightgreen if v else for v in is_max] crosstab_styled crosstab.style.apply(highlight_max, axis1) # 导出为HTML报表时绿色高亮每行最大值4.7 分析6高管简报的“决策友好型”指标工程summary表中的avg_fee_percent计算total_fees / total_spend * 100但业务上需区分“正常费率”和“异常费率”。我们增加业务标签# 基于行业基准费率如信用卡0.25%-2.5%设区间 def fee_category(rate): if rate 0.5: return Below Benchmark elif rate 2.0: return Within Benchmark else: return Above Benchmark summary[fee_category] summary[avg_fee_percent].apply(fee_category) # 输出C001 - Within BenchmarkC002 - Above Benchmark这比单纯数字更能驱动业务动作——“Above Benchmark”的客户需立即核查手续费协议。5. 生产环境排障实录那些让你凌晨三点爬起来的报错5.1 常见报错速查表报错信息根本原因解决方案发生频率ValueError: No numeric types to aggregate分组列含非数值类型如NaN、字符串df[col] pd.to_numeric(df[col], errorscoerce)强制转数值NaN置0★★★★☆KeyError: column_nameagg字典键名与DataFrame列名不匹配大小写/空格/下划线print(df.columns.tolist())查看真实列名用df.columns df.columns.str.strip().str.lower()标准化★★★★★MemoryErroronexpanding().std()大数据集扩展计算内存溢出改用safe_expanding_std()分块计算或改用ewm(span7).std()指数加权近似★★★☆☆NaNin rolling result for first N rowsmin_periods未设置默认要求窗口满员rolling(window7, min_periods1).mean()但需评估首日值业务意义★★★★☆unstack()后列名含(col,func)导致下游失败未展平MultiIndex列名result.columns [_.join(col) for col in result.columns]★★★★★5.2 高危场景深度排障场景滚动窗口计算结果与业务方预期严重不符现象业务方说“7日均值应该平滑但你们的图锯齿状”。排查路径检查数据频率df[date].diff().value_counts()发现大量1 days和3 days间隔因跳过周末证明数据非连续。检查索引df.index.freq返回None说明未设频率。验证df.set_index(date).asfreq(D).rolling(7D).mean()后锯齿消失。根因未对时间序列做asfreq()填充rolling(window7)按行数滚动非按日历滚动。场景自定义函数在分布式环境Dask中报NameError现象本地pandas正常Dask集群报NameError: name weighted_average is not defined。原因Dask worker进程未导入该函数。解决方案from dask.distributed import Client client Client() # 显式分发函数到所有worker client.run(lambda: exec(from my_module import weighted_average)) # 或用client.submit()包装场景unstack()后出现ValueError: Index contains duplicate entries现象多级分组后unstack()报错。根因分组键组合不唯一如[region,product]中存在重复(North, Widget)。诊断df_sales.groupby([region,product]).size().sort_values(ascendingFalse).head(10)查看重复最多的组合。修复添加唯一标识如df_sales[id] range(len(df_sales))分组时加入id或用agg(sum)聚合重复项。5.3 我踩过的三个深坑坑1agg()中混用lambda与named function导致结果错位曾写agg({amount: [mean, lambda x: x.std()], fee: sum})结果amount的lambda结果被塞进fee_sum列。教训agg字典中同一列的所有聚合必须同构——要么全用列表[mean,std]要么全用字典{mean: mean, std: lambda x: x.std()}。坑2rolling().mean()在时序数据中忽略时区导致跨时区计算错误某次为亚太区客户计算7日均值数据源是UTC时间但业务方按本地时间解读。结果东京客户周一的交易被计入周日滚动窗口。教训所有时间序列操作前必须df[date] df[date].dt.tz_convert(Asia/Tokyo)转换到业务时区。坑3unstack()后fillna(0)掩盖了真实的数据缺失为南区生成报表时unstack(fill_value0)让所有“无交易”产品显示0业务方误以为是“亏损”实际是“未开展业务”。教训用unstack().where(pd.notna(unstacked), No Data)用字符串明确标识缺失原因。6. 进阶实战如何把这五种模式组装成银行级风控管道6.1 构建实时交易监控仪表盘将五种模式组合构建一个每5分钟刷新的风控看板# 数据流Kafka - Spark Streaming - Pandas Batch def build_realtime_dashboard(batch_df: pd.DataFrame) - dict: 输入5分钟批次交易数据 输出JSON格式仪表盘数据 # 步骤1多维聚合商户地区时间窗 hourly_agg batch_df.groupby([ merchant_id, region, pd.Grouper(keytimestamp, freqH) ]).agg({ amount: [count, sum, mean], fee: [sum] }) # 步骤2自定义风险分段高价值交易占比 def high_value_ratio(series): return (series 5000).sum() / len(series) if len(series) 0 else 0 risk_ratio batch_df.groupby([merchant_id, region])[amount].apply(high_value_ratio) # 步骤3滚动窗口过去24小时均值 vs 当前小时均值 # 先按小时聚合 hourly_sum batch_df.groupby([ merchant_id, region, pd.Grouper(keytimestamp, freqH) ])[amount].sum() # 计算滚动24小时均值 rolling_24h hourly_sum.groupby([merchant_id, region]).rolling(24H).mean() # 步骤4合并结果 dashboard_data { hourly_stats: hourly_agg.to_dict(), risk_ratios: risk_ratio.to_dict(), anomaly_score: (hourly_sum / rolling_24h).to_dict(), # 当前小时/24小时均值 last_updated: batch_df[timestamp].max().isoformat() } return dashboard_data # 使用 # dashboard_json build_realtime_dashboard(kafka_batch_df)6.2 构建监管报送自动化流水线满足《巴塞尔协议III》对交易对手风险暴露的报送要求def generate_basel_report(df_transactions: pd.DataFrame) - pd.DataFrame: 生成监管报送格式按对手方、产品、期限、币种四维聚合 # 四级分组监管硬性要求 basel_group df_transactions.groupby([ counterparty_id, # 对手方 product_type, # 产品贷款/债券/衍生品 maturity_bucket, # 期限1Y, 1-5Y, 5Y currency # 币种 ]) # 多列聚合监管指定指标 report basel_group.agg({ exposure_amount: [sum, max