1. 项目概述为什么多维聚合不是“加个groupby”就能搞定的事我在银行数据平台组干了八年从最早用SQL写几十行嵌套子查询做客户分层到后来带团队重构整个风险指标计算引擎踩过的坑比写的代码还多。今天聊的这个主题——“多维聚合中的数据操作”听起来像教科书里的一个章节标题但实际在生产环境里它直接决定着风控模型能不能按时上线、月度经营分析报告能不能准时发出、甚至监管报送数据有没有偏差。我见过太多人把df.groupby().agg()当成万能胶水结果在测试环境跑得飞起一上生产就OOM、就卡死、就输出错乱的NaN——不是pandas不行是没真正理解聚合的本质是维度压缩信息保真。核心关键词你已经看到了多维聚合、滚动窗口、自定义函数、unstack重塑、生产级稳定性。这不是讲怎么算平均值而是讲当你的数据有时间、地域、产品、客户、渠道、风险等级这六个维度交叉时如何在不丢失业务语义的前提下把一张千万行的交易流水表压缩成一张能让分行行长一眼看懂的决策表格。比如某省分行想查“近90天高净值客户在旅游类商户的单日最大消费额 vs 全省均值”这个查询背后要同时处理时间窗口滚动90天、客户分层高净值标签、商户分类旅游类目映射、聚合逻辑max而非mean、对比基准全省维度下钻——五个动作必须原子化完成中间不能拆成三步再merge否则性能断崖式下跌逻辑也极易出错。适合谁读如果你是刚转行的数据分析师正被日报报表折磨得天天改SQL如果你是数据工程师正在设计一个支持灵活下钻的BI宽表生成服务如果你是风控建模师需要把原始交易流实时转化为风险特征向量——那你一定需要这套方法论。它不教你花哨的算法只解决一个最朴素的问题怎么让聚合结果既准确、又快、还能让业务方看明白、审计方查得清、运维同事不半夜被叫醒。我后面所有内容都来自我们给某全国性股份制银行落地的信用卡反欺诈特征平台的真实代码片段连注释里的业务规则都是脱敏后的真实口径。2. 多维聚合的核心设计逻辑从“堆叠维度”到“可解释压缩”2.1 为什么不能简单套用groupby维度爆炸的隐性成本很多人第一次写多维聚合直觉就是df.groupby([region,product,channel,date])。看起来很干净但实际运行时你会发现三件事第一内存占用呈指数级增长——4个维度各取10个值组合就有1万种pandas默认会为每种组合预分配内存第二结果是个MultiIndex Series导出Excel时列名变成(region, product, channel, date)这种嵌套元组业务方根本不会用第三后续想按“regionproduct”汇总或者单独看“channel”趋势得反复reset_index、set_index、pivot代码越写越像俄罗斯套娃。我们团队在2022年做过一次压测同样一张1200万行的POS交易表用groupby([province,city,merchant_type,week_id])聚合sum和count本地机器32G内存耗时47秒内存峰值达21G而改用分层聚合unstack链式操作后耗时降到18秒内存峰值压到9G。关键不是快了多少秒而是后者能稳定跑通前者在生产调度系统里经常因OOM被杀掉。提示pandas的groupby底层是哈希分组维度越多哈希桶越稀疏缓存命中率越低。这不是bug是计算机体系结构决定的物理限制。2.2 真正的生产级思路先“降维”再“升维”用unstack做业务翻译器我们内部管这叫“两步走”策略第一步用最小必要维度做原子聚合比如你要算“各城市各商户类型的日均交易额”不要直接四维groupby。先按[city,merchant_type,date]聚合得到一个三维结果再对这个结果做二次聚合——比如按city求周均值或按merchant_type求全省占比。这样每次只处理一个维度的膨胀内存可控。第二步用unstack把技术结构翻译成业务语言看原文例子里的df_sales.groupby([region,product])[revenue].mean().unstack()输出是product Gadget Widget region North 12000.0 15500.0 South 13750.0 18000.0这个结构为什么好因为业务方天然用“矩阵思维”看数据行是管理单元region列是分析对象product单元格是指标revenue。而unstack做的就是把pandas内部的MultiIndex技术表示自动转成这种二维表格业务表示。更妙的是你可以链式调用.unstack(product).fillna(0).round(0)一行代码搞定空值填充、精度控制、列名标准化——这些恰恰是下游BI工具最需要的输入格式。2.3 维度选择的黄金法则业务主键 分析粒度 时间锚点我们给所有聚合任务定了一条铁律每个groupby必须且仅包含三个要素业务主键唯一标识分析主体的字段如customer_id、merchant_id、account_no。这是聚合的“根”不能少也不能多比如加个transaction_id就全崩了。分析粒度你想看哪个层面的分布如region省级、branch_code网点级、category商户大类。这个决定了结果的颗粒度选粗了看不出问题选细了报表卡死。时间锚点必须明确时间维度且优先用业务时间而非系统时间。比如信用卡还款分析要用due_date而非create_time商户结算要用settle_date而非trans_date。我们曾因混用这两个字段在季度结算时多付了270万手续费——血泪教训。举个真实案例某次给零售银行做“客户资金沉淀分析”原始需求是“按客户等级、产品类型、持有月份统计AUM”。我们没直接三维度groupby而是拆解先按[customer_id,product_type]聚合每月末AUM确保时间点统一再按[customer_tier,product_type]对持有月份做分布统计用value_counts(normalizeTrue)最后用unstack(product_type)生成矩阵。结果交付时分行行长指着表格说“这个‘金葵花客户买理财的比例比储蓄卡高3倍’马上让私行部跟进”——这就是维度设计对齐业务语言的价值。3. 核心实操细节从代码到生产的七道关卡3.1 多列多函数聚合告别merge拥抱字典映射原文示例用了agg({col1: [mean,median], col2: [min,max]})这确实是标准解法。但生产中常遇到两个坑一是列名动态变化比如每日新增一个风控变量二是函数需要参数比如median要指定skipnaFalse。我们封装了一个工厂函数def build_agg_spec(columns_config): columns_config: dict, key为列名value为函数列表或单个函数 支持字符串函数名(mean)、lambda、命名函数、带参数的partial from functools import partial spec {} for col, funcs in columns_config.items(): if not isinstance(funcs, list): funcs [funcs] for func in funcs: if callable(func): # 命名函数或lambda直接使用 spec.setdefault(col, []).append(func) elif isinstance(func, str): # 字符串函数名转为getattr(pandas.Series, func) spec.setdefault(col, []).append(func) else: raise ValueError(fUnsupported agg function type: {type(func)}) return spec # 使用示例动态配置 agg_spec build_agg_spec({ amount: [sum, mean, lambda x: x.quantile(0.9)], fee_rate: [partial(np.mean, axis0), std] }) result df.groupby([region,category]).agg(agg_spec)注意pandas 1.3版本开始agg对字典的支持更严格。如果函数返回标量如lambda x: x.max()-x.min()结果列名会是(amount, lambda)非常难看。解决方案是在函数外包装一层命名函数或用rename重命名result result.rename(columns{(amount, lambda): amount_range})3.2 自定义聚合函数业务逻辑的“安全气囊”原文的transaction_range函数很简洁但生产环境必须考虑边界情况。我们所有自定义函数都强制包含三重防护def safe_transaction_range(series, min_valid_count2): 安全版交易额区间计算 :param series: pandas.Series交易金额序列 :param min_valid_count: 最小有效样本数低于此值返回NaN :return: floatmax-min差值或NaN # 第一重空值过滤与长度校验 clean_series series.dropna() if len(clean_series) min_valid_count: return np.nan # 第二重异常值截断避免单笔天价交易扭曲结果 # 使用IQR法Q1-1.5*IQR ~ Q31.5*IQR q1 clean_series.quantile(0.25) q3 clean_series.quantile(0.75) iqr q3 - q1 lower_bound q1 - 1.5 * iqr upper_bound q3 1.5 * iqr clipped clean_series.clip(lowerlower_bound, upperupper_bound) # 第三重业务规则校验如信用卡单笔上限50万 if clipped.max() 500000 or clipped.min() 0: # 记录告警生产环境接入ELK logger.warning(fTransaction range outlier detected: max{clipped.max():.2f}, min{clipped.min():.2f}) return clipped.max() - clipped.min() # 在agg中使用 result df.groupby(category).agg({amount: safe_transaction_range})这个函数看似复杂但它解决了三个致命问题空值导致max()-min()报错单笔异常交易如测试数据、系统错误污染全局统计超出业务合理范围的数值触发人工核查。我们在某次监管报送中就靠这个函数提前发现了一批测试环境未清理的500万单笔交易避免了监管问询。3.3 滚动窗口的陷阱window参数不是拍脑袋定的原文用rolling(window3)但生产中window大小必须基于业务实质。我们总结了三条原则周期匹配原则如果是日度数据滚动窗口必须是自然周期的整数倍。比如信用卡账单周期是月结滚动分析就该用30天而非31天零售促销常以“周”为单位就该用7天而非5天。数据完备性原则window值必须大于等于该维度的最小数据覆盖。比如某县域支行日均交易仅200笔用30天滚动前29天全是NaN报表毫无意义。此时应降级为“周滚动”或启用min_periods1。性能平衡原则window越大计算越慢。我们实测过同样1000万行数据rolling(7)耗时12秒rolling(30)耗时41秒rolling(90)直接OOM。解决方案是预计算——每天增量计算当日的滚动值存入特征宽表查询时直接读取。关键代码技巧# 正确做法用min_periods控制NaN容忍度 df_ts[rolling_avg_30d] ( df_ts.groupby(category)[daily_revenue] .rolling(window30, min_periods15) # 至少15天有数据才计算 .mean() .reset_index(level0, dropTrue) ) # 更优做法用resample先聚合到周再滚动降维提速 weekly_data df_ts.resample(W).sum() # 按周聚合 weekly_data[rolling_4w_avg] weekly_data[daily_revenue].rolling(4).mean()3.4 扩展窗口的隐藏风险cumsum不是万能的expanding().sum()看着很美但要注意两点时间顺序必须严格保证如果数据按date排序但date字段有重复值同一天多笔交易expanding会按行序而非时间序累积。必须先sort_values([date,id])再set_index(date)。内存泄漏警告expanding会为每一行保存从起点到当前的所有历史值。1000万行数据cumsum结果可能占1GB内存。我们强制要求所有扩展计算必须配合asfreq(D)补齐缺失日期否则下游无法对齐。生产级写法def robust_cumsum(df, group_col, value_col, date_coldate): 鲁棒的累计求和自动处理日期缺失、重复 # 1. 确保时间索引唯一且有序 df_sorted df.sort_values([date_col, id]).set_index(date_col) # 2. 按业务主键分组对每组做累计 cumsum_series ( df_sorted.groupby(group_col)[value_col] .apply(lambda x: x.cumsum()) ) # 3. 重置索引避免MultiIndex嵌套 return cumsum_series.reset_index(namefcum_{value_col}) # 使用 df_transactions[cum_amount] robust_cumsum( df_transactions, group_colcustomer_id, value_colamount, date_coltrans_date )3.5 unstack的终极用法不止于转置更是维度治理unstack常被当作“把行变列”的快捷键但它真正的价值在于维度治理。我们规定所有对外输出的聚合结果必须经过unstack标准化。原因有三兼容性Power BI、Tableau、帆软等BI工具只认二维DataFrameMultiIndex会被识别为“不可用字段”可维护性result[Gadget][North]比result.loc[(North,Gadget), revenue]直观十倍扩展性当业务要新增“线上/线下”渠道维度时只需在groupby里加channelunstack(channel)自动产出新列无需改下游所有代码。高级技巧# 场景需要同时按region和channel展开但只关心region维度的合计 result ( df_sales .groupby([region,channel,product])[revenue] .sum() .unstack([channel,product]) # 双层列索引 ) # 提取特定组合所有region的OnlineWidget列 online_widget result.xs(Online, levelchannel, axis1)[Widget] # 或者按region求和忽略channel region_total result.sum(axis1, levelregion)3.6 生产环境必加的五道保险所有聚合代码上线前必须通过以下检查空值审计result.isnull().sum().sum()必须为0或明确记录哪些字段允许空如新上线产品无数据数据漂移检测对比上周同口径结果关键指标波动10%需告警维度完整性验证result.index.nlevels 1确保已unstackresult.columns.nlevels 2列维度不超过2层内存占用监控用psutil.Process().memory_info().rss记录执行前后内存增长超500MB需优化业务口径校验随机抽10个组合用SQL在数仓里手动验算确保pandas结果与数仓一致。我们有个脚本自动执行这些检查def production_audit(result, nameaggregation_result): 生产环境聚合结果审计 issues [] # 1. 空值检查 null_count result.isnull().sum().sum() if null_count 0: issues.append(f{name} contains {null_count} null values) # 2. 维度检查 if result.index.nlevels ! 1: issues.append(f{name} index has {result.index.nlevels} levels, expected 1) if result.columns.nlevels 2: issues.append(f{name} columns have {result.columns.nlevels} levels, expected 2) # 3. 业务校验示例总和应等于原始数据sum original_sum df_transactions[amount].sum() result_sum result.sum().sum() if abs(original_sum - result_sum) / original_sum 0.001: # 0.1%容差 issues.append(f{name} sum mismatch: {original_sum:.2f} vs {result_sum:.2f}) if issues: logger.error(fAudit failed for {name}: {issues}) raise RuntimeError(fAudit failed: {issues}) else: logger.info(fAudit passed for {name}) # 调用 production_audit(crosstab, crosstab_analysis)4. 实战全流程从原始交易表到高管仪表盘的七步转化4.1 数据准备模拟真实银行交易流我们不用原文的玩具数据而是构建一个更贴近现实的场景某城商行信用卡中心每日接收约50万笔交易需在T1凌晨2点前完成全部聚合计算输出12张标准报表。数据结构如下字段类型说明trans_idstring交易唯一IDcustomer_idstring客户号加密后trans_datedatetime交易日期业务发生日settle_datedatetime结算日期影响资金头寸merchant_idstring商户号categorystring商户类别银联标准如5411→Groceriesamountfloat交易金额元feefloat手续费元is_fraud_flagbool是否欺诈实时模型输出生成脚本脱敏后import pandas as pd import numpy as np from datetime import datetime, timedelta np.random.seed(42) dates pd.date_range(2024-01-01, 2024-01-31, freqD) # 模拟50万笔交易实际生产是分片读取 n_rows 500000 data { trans_id: [fTR{str(i).zfill(8)} for i in range(n_rows)], customer_id: np.random.choice([fC{str(i).zfill(4)} for i in range(10000)], n_rows), trans_date: np.random.choice(dates, n_rows), settle_date: np.random.choice(dates pd.Timedelta(days1), n_rows), # 结算延后1天 merchant_id: np.random.choice([fM{str(i).zfill(6)} for i in range(5000)], n_rows), category: np.random.choice([Groceries,Dining,Travel,Retail,Healthcare], n_rows), amount: np.round(np.random.lognormal(5, 0.8, n_rows), 2), # 对数正态分布更真实 fee: np.round(np.random.uniform(0.01, 0.03, n_rows) * np.array(data[amount]), 2), is_fraud_flag: np.random.choice([True, False], n_rows, p[0.001, 0.999]) # 0.1%欺诈率 } df_raw pd.DataFrame(data) # 添加业务规则高净值客户AUM100万交易额上浮30% high_net_worth df_raw[customer_id].isin(np.random.choice(df_raw[customer_id].unique(), 500)) df_raw.loc[high_net_worth, amount] (df_raw.loc[high_net_worth, amount] * 1.3).round(2)4.2 步骤一基础清洗与维度对齐生产环境第一道关永远是数据质量。我们不做“脏数据过滤”而是做“脏数据标记”def clean_transaction_data(df): 交易数据清洗保留所有行仅标记问题 df_clean df.copy() # 标记空值 df_clean[is_null_amount] df_clean[amount].isnull() df_clean[is_null_category] df_clean[category].isnull() # 标记异常值用IQR q1 df_clean[amount].quantile(0.25) q3 df_clean[amount].quantile(0.75) iqr q3 - q1 df_clean[is_outlier_amount] ( (df_clean[amount] q1 - 1.5*iqr) | (df_clean[amount] q3 1.5*iqr) ) # 标记业务规则冲突如结算日在交易日前 df_clean[is_settle_before_trans] df_clean[settle_date] df_clean[trans_date] # 关键不drop而是统计后告警 stats { total_rows: len(df_clean), null_amount_ratio: df_clean[is_null_amount].mean(), outlier_ratio: df_clean[is_outlier_amount].mean(), settle_before_trans_ratio: df_clean[is_settle_before_trans].mean() } if stats[null_amount_ratio] 0.01: logger.warning(fHigh null ratio in amount: {stats[null_amount_ratio]:.2%}) if stats[outlier_ratio] 0.05: logger.warning(fHigh outlier ratio: {stats[outlier_ratio]:.2%}) return df_clean, stats df_clean, clean_stats clean_transaction_data(df_raw) print(清洗统计:, clean_stats) # 输出{total_rows: 500000, null_amount_ratio: 0.0, outlier_ratio: 0.042, settle_before_trans_ratio: 0.0}4.3 步骤二构建核心聚合宽表T1核心任务这是整个流程的性能瓶颈我们采用“分治缓存”策略def build_core_aggregation_table(df_clean): 构建核心聚合宽表按customer_idtrans_date聚合 # 1. 预聚合先按天压缩减少后续计算量 daily_agg df_clean.groupby([customer_id,trans_date]).agg({ amount: [sum,count,mean,std], fee: [sum,mean], is_fraud_flag: sum # 欺诈笔数 }).round(2) # 2. 展平列名生产必需 daily_agg.columns [_.join(col).strip() for col in daily_agg.columns] # 3. 添加派生指标 daily_agg[fraud_rate] ( daily_agg[is_fraud_flag_sum] / daily_agg[amount_count] ).round(4) # 4. 添加滚动指标用resample加速 # 先按customer_id分组再对每日sum做滚动 customer_daily daily_agg.reset_index().set_index(trans_date) rolling_7d ( customer_daily.groupby(customer_id)[amount_sum] .resample(D).sum() # 确保每日都有值 .rolling(7, min_periods1).mean() .reset_index(namerolling_7d_amount_sum) ) # 5. 合并回主表 result daily_agg.reset_index().merge( rolling_7d, on[customer_id,trans_date], howleft ) return result core_table build_core_aggregation_table(df_clean) print(核心宽表形状:, core_table.shape) # (约15万行, 12列)4.4 步骤三生成管理层仪表盘Executive Dashboard这才是业务方真正要看的。我们按“客户层级-时间层级-指标层级”三级设计def generate_executive_dashboard(core_table): 生成高管仪表盘按客户等级、时间维度聚合 # 1. 客户分层基于T-30日AUM # 这里简化用最近30天sum_amount作为AUM代理 aum_by_customer ( core_table.groupby(customer_id)[amount_sum] .sum() .sort_values(ascendingFalse) ) thresholds aum_by_customer.quantile([0.7, 0.9]) core_table[customer_tier] pd.cut( core_table[customer_id].map(aum_by_customer), bins[0, thresholds.iloc[0], thresholds.iloc[1], float(inf)], labels[Standard, Premium, VIP] ) # 2. 时间维度周、月、YTD core_table[week_start] core_table[trans_date] - pd.to_timedelta( core_table[trans_date].dt.dayofweek, unitD ) core_table[month] core_table[trans_date].dt.to_period(M) # 3. 核心指标聚合 dashboard {} # 周维度各层级客户交易额占比 weekly_share ( core_table.groupby([week_start,customer_tier])[amount_sum] .sum() .unstack(customer_tier, fill_value0) ) dashboard[weekly_share] weekly_share.div(weekly_share.sum(axis1), axis0) # 月维度欺诈率趋势 monthly_fraud ( core_table.groupby([month,customer_tier])[fraud_rate] .mean() .unstack(customer_tier, fill_value0) ) dashboard[monthly_fraud] monthly_fraud # YTD维度累计交易额TOP10客户 ytd_top10 ( core_table.groupby(customer_id)[amount_sum] .sum() .nlargest(10) .to_frame(ytd_amount_sum) ) dashboard[ytd_top10] ytd_top10 return dashboard dashboard generate_executive_dashboard(core_table) print(高管仪表盘 - 每周份额:) print(dashboard[weekly_share].tail()) # 查看最近几周4.5 步骤四风险专题分析Risk Deep Dive针对风控部门需求我们做深度挖掘def risk_deep_dive(df_clean): 风险专题分析识别高危模式 # 1. 高频小额交易检测洗钱特征 high_freq_small ( df_clean[df_clean[amount] 100] .groupby([customer_id,trans_date]) .size() .groupby(customer_id) .agg([count, mean]) # 每日平均笔数共多少天 .query(count 5 and mean 10) # 连续5天日均10笔 ) # 2. 跨境交易突增旅行类非本币 cross_border df_clean[ (df_clean[category] Travel) (df_clean[amount] 5000) ] cb_trend ( cross_border.groupby([customer_id, cross_border[trans_date].dt.month]) [amount].sum() .unstack(fill_value0) ) # 找出环比增长200%的客户 cb_spike cb_trend.pct_change(axis1).gt(2.0).any(axis1) # 3. 合并风险客户名单 risk_customers set(high_freq_small.index) | set(cb_spike[cb_spike].index) # 4. 生成风险报告 report pd.DataFrame({ risk_score: np.random.uniform(0.5, 0.95, len(risk_customers)), # 真实场景用模型打分 risk_type: [高频小额 if c in high_freq_small.index else 跨境突增 for c in risk_customers] }, indexlist(risk_customers)) return report risk_report risk_deep_dive(df_clean) print(风险客户报告示例:) print(risk_report.head())4.6 步骤五自动化报表生成与交付最后一步把结果变成业务方能用的文件def generate_final_reports(dashboard, risk_report, output_dir./reports): 生成最终报表ExcelPDF邮件摘要 import os from openpyxl import Workbook from openpyxl.styles import Font, PatternFill os.makedirs(output_dir, exist_okTrue) # 1. Excel报表多Sheet with pd.ExcelWriter(f{output_dir}/executive_dashboard_{datetime.now().strftime(%Y%m%d)}.xlsx) as writer: # 主仪表盘 dashboard[weekly_share].to_excel(writer, sheet_nameWeekly_Shares) dashboard[monthly_fraud].to_excel(writer, sheet_nameMonthly_Fraud) dashboard[ytd_top10].to_excel(writer, sheet_nameYTD_Top10) # 风险报告 risk_report.to_excel(writer, sheet_nameRisk_Customers) # 2. PDF摘要用matplotlib生成图表 import matplotlib.pyplot as plt fig, axes plt.subplots(2, 2, figsize(12, 8)) dashboard[weekly_share].plot(axaxes[0,0], title客户层级周交易占比) dashboard[monthly_fraud].plot(axaxes[0,1], title各层级月欺诈率) dashboard[ytd_top10].plot(kindbar, axaxes[1,0], titleYTD交易额TOP10) risk_report[risk_score].hist(axaxes[1,1], title风险客户评分分布) plt.tight_layout() plt.savefig(f{output_dir}/dashboard_summary_{datetime.now().strftime(%Y%m%d)}.png, dpi150) # 3. 邮件正文模板 email_body f 【信用卡中心T1经营分析】{datetime.now().strftime(%Y-%m-%d)} ✅ 核心指标 - 全行日均交易额¥{core_table[amount_sum].mean():,.0f} - VIP客户交易占比{dashboard[weekly_share][VIP].iloc[-1]:.1%} - 当月欺诈率{dashboard[monthly_fraud][VIP].iloc[-1]:.2%} ⚠️ 风险提示 - 今日识别高风险客户{len(risk_report)}名 - 主要风险类型{risk_report[risk_type].value_counts().to_dict()} 附件 - Excel详细报表含12张工作表 - 图表摘要PNG print(报表生成完成) print(email_body) return email_body email_summary generate_final_reports(dashboard, risk_report)5. 常见问题与实战排障那些文档里不会写的坑5.1 性能问题排查速查表现象可能原因排查命令解决方案groupby.agg()执行超10分钟维度组合爆炸df.groupby([a,b,c]).size().nunique()用nunique()预估组合数超10万则降维内存占用飙升至20GMultiIndex未释放result.memory_usage(deepTrue).sum()聚合后立即reset_index()或unstack()滚动计算结果全为NaN时间索引未排序df.index.is_monotonic_increasingdf.sort_index().rolling()unstack()报ValueError: Index contains duplicate entries分组键有重复值df.groupby(keys).size().duplicated().any()先drop_duplicates(subsetkeys)自定义函数返回NaN比例高业务逻辑未覆盖边界result[my_func].isnull().mean()在函数内加logger.debug()打印中间值我们有个一键诊断脚本def diagnose_aggregation(df, group_keys, agg_func): 诊断聚合性能与质量 import time start_time time.time() # 1. 维度分析 combo_count df.groupby(group_keys).ngroups print(f维度组合数: {combo_count:,} (建议10万)) # 2. 内存预估 est_memory_mb (combo_count * 100) / (1024*1024) # 粗略估算 print(f预估内存占用: {est