1. 项目概述为什么多维聚合不是“加个groupby”就能搞定的事我在银行数据平台组干了八年从最早用SQL写几十行嵌套子查询做客户分层到现在带团队设计日均处理20亿条交易的实时聚合管道踩过的坑比别人走过的路还多。今天聊的这个主题——“多维聚合中的数据操作”听起来像教科书里的一个章节标题但实际在生产环境里它直接决定着风控模型能不能及时拦截一笔可疑交易、运营活动预算要不要紧急调整、甚至监管报送文件能不能准时提交。你手里的pandas.groupby()不是语法糖而是一把双刃剑用对了三行代码顶三个月ETL开发用错了凌晨两点被电话叫醒查内存溢出还是找不到问题在哪。核心关键词就三个多维聚合、生产级、业务语义。注意不是“多维分析”——那是BI工具的事也不是“聚合函数”——那是数据库基础课内容我们聚焦的是当真实业务问题同时横跨时间、空间、产品、客户四个维度且每个维度都带着非标准计算逻辑时怎么让代码既跑得稳、又看得懂、还能经得起审计。比如某次反洗钱系统升级要求对“近90天内单日交易超5笔且单笔超3万元”的客户按其常消费的TOP3商户类别分别计算滚动30天的金额变异系数标准差/均值。这个需求拆开看是时间窗口频次过滤金额阈值商户聚类统计指标归一化——没有一个现成的agg()能直接套用。我见过太多人把这当成纯技术问题查文档、试参数、调API。结果呢代码跑通了输出看着也像那么回事但业务方一问“为什么这个客户的变异系数是0.87而不是0.92”立马哑火。因为没人把“变异系数在这里代表资金流动稳定性”这个业务含义和pandas的.std().mean()计算链真正对齐。所以这篇不是讲语法而是讲怎么把银行信贷经理的口头需求翻译成机器可执行、人可验证、审计可追溯的数据操作流。后面所有实操细节都来自我们去年上线的“零售客户价值动态评估系统”真实代码库——已稳定运行427天日均处理1.2亿条聚合任务零生产事故。2. 多维聚合的核心设计逻辑从“算得出来”到“算得明白”2.1 为什么必须放弃“先group再agg”的线性思维新手最容易犯的错误就是把多维聚合当成单维度的叠加。比如要算“各地区各产品线的销售额中位数退货率”第一反应是df.groupby([region,product]).agg({sales:median, returns:sum, orders:sum})看起来没问题错。这里埋了三个雷雷1中位数的业务陷阱median()在pandas里默认对每列独立计算但“地区A的电子产品中位数”和“地区B的服装中位数”根本不可比——前者样本量可能5000后者才200。业务上真正需要的是“剔除异常值后的稳健中位数”而pandas原生median不支持winsorize缩尾预处理。我们最终方案是先用scipy.stats.mstats.winsorize对sales列做5%缩尾再groupby计算否则某地突发一场大促就会扭曲整个区域基准。雷2退货率的分母陷阱returns.sum()/orders.sum()看似合理但订单数为0时会触发除零警告更致命的是如果某产品在某地区当月没订单这个组合在groupby结果里直接消失导致下游报表显示“该地区无此产品”而实际是“该产品未销售”。正确做法是强制保留全组合用pd.MultiIndex.from_product生成完整索引再用reindex(fill_value0)补零。雷3聚合粒度漂移上面代码实际计算的是“每个地区,产品组合的退货率”但业务需求其实是“每个地区的退货率按产品线加权”。这就涉及聚合层级的嵌套先按地区分组再在组内按产品计算权重最后加权平均。pandas的agg()不支持这种嵌套逻辑必须用apply()配合自定义函数。提示真正的多维聚合不是“一次groupby搞定”而是“分层计算跨层校验”。我们团队内部有个铁律任何agg()调用前必须手写三行注释说明——① 这个聚合的业务定义是什么引用业务手册条款号② 分母是否包含所有可能场景特别是零值、空值③ 结果是否满足下游系统字段精度要求比如监管报送要求小数点后4位但float64计算可能有1e-15误差。2.2 生产环境必须考虑的四大刚性约束在银行系统里聚合不是学术练习而是受多重硬约束的工程任务。我拿去年一个真实案例说明为央行《金融机构客户尽职调查指引》做数据准备要求输出“高风险客户近6个月交易对手地域分布热力图”。约束1内存墙原始交易表120GB客户ID哈希后仍有800万唯一值。如果直接df.groupby([customer_id,counterparty_region])pandas会尝试构建800万×200地域数的稀疏矩阵内存瞬间飙到48GB。解决方案是分块聚合用pd.read_csv(chunksize50000)流式读取每块单独groupby后用pd.concat()合并中间结果再全局聚合。实测内存峰值压到6.2GB耗时只增加17%。约束2时间一致性“近6个月”不是简单df[df.date pd.Timestamp.now() - pd.DateOffset(months6)]。因为交易数据入库有延迟T1日才完整而监管要求“截至报告日T的数据快照”。我们建了专用的时间锚点表每日报送时先查reporting_calendar表获取当前有效截止日期如2024-03-15再用该日期计算6个月前日期。避免了因系统时钟误差导致的报送偏差。约束3审计可追溯性所有聚合结果必须附带溯源信息。我们在最终DataFrame里强制添加三列_agg_version聚合逻辑版本号、_data_snapshot_date原始数据抽取时间、_agg_timestamp聚合完成时间戳。更重要的是对每个custom agg函数用inspect.getsource()捕获源码存入元数据表。去年审计时监管员随机抽查了3个指标我们5分钟内就提供了从原始SQL抽取→清洗规则→聚合函数→结果校验的全链路证据。约束4业务逻辑隔离风控、财务、运营部门对同一数据源的聚合逻辑完全不同。比如“客户活跃度”风控部定义为“近30天登录次数≥3且有交易”财务部定义为“近30天产生手续费收入≥100元”。如果混写在一个agg()里改一个需求就要全量回归测试。我们的解法是用策略模式封装聚合逻辑每个部门对应一个继承自BaseAggregator的类通过配置文件切换实现。上线后财务部修改活跃度定义只需更新finance_aggregator.py其他模块完全不受影响。2.3 多维聚合的拓扑结构为什么“unstack”不是格式美化而是架构选择很多人把unstack()当成Excel透视表的替代品这是巨大误解。在我们系统里unstack是聚合结果的物理存储格式契约。举个例子客户价值评分模型需要输入一个宽表列为[customer_id, region_north_revenue, region_south_revenue, product_widget_count, product_gadget_count...]。如果不用unstackgroupby结果是MultiIndex Series转成宽表要写十几行reset_indexpivot且列名动态生成难维护。但unstack的威力不止于此。我们发现当聚合维度超过3个时如[customer_segment, region, product_category, quarter]直接unstack会导致列爆炸。这时我们采用“分层unstack”策略# 先按前两个维度groupby得到MultiIndex DataFrame result df.groupby([segment,region]).agg({...}) # 对region维度unstack生成宽表 wide_by_region result.unstack(region, fill_value0) # 再对product_category维度在每个region列下继续unstack # 这里用pd.concat拼接各region的product维度结果 product_wide {} for col in wide_by_region.columns.levels[0]: if col ! segment: # 跳过索引列 temp df[df.region col].groupby([segment,product_category]).agg({...}) product_wide[col] temp.unstack(product_category, fill_value0) final_result pd.concat(product_wide, axis1)这个操作看似复杂但它解决了生产中最痛的痛点下游系统对接成本。BI工具、监管报送接口、机器学习特征工程平台都要求固定列名结构。unstack生成的列名如revenue__north__widget可直接映射到目标系统的字段无需额外的列名映射配置。去年接入新监管报送系统时仅靠unstack的命名规范就节省了3人日的字段对齐工作。3. 核心实操要点七种必须掌握的聚合模式详解3.1 混合聚合不同列用不同函数但必须解决“列名地狱”原文示例中df.groupby(merchant_category).agg({transaction_amount: [mean,median], processing_fee: [min,max]})输出的列名是(transaction_amount, mean)这样的元组这在Jupyter里看着清爽但进生产就灾难——下游Java服务解析不了嵌套列名Excel导出后列名变成transaction_amount,mean带逗号连不上数据库字段。我们的标准化解法是列名扁平化业务前缀def flatten_agg_columns(df, prefix): 将MultiIndex列名扁平化为prefix_colname_func格式 如 (transaction_amount, mean) - ta_mean if not isinstance(df.columns, pd.MultiIndex): return df new_cols [] for col in df.columns: # 取列名首字母缩写 函数名缩写 base_name col[0][:2].lower() # transaction_amount - ta func_name col[1].lower() # mean - mean, std - std # 特殊处理避免冲突如amount和fee都缩写为am if col[0] transaction_amount: base_name ta elif col[0] processing_fee: base_name pf elif col[0] transaction_count: base_name tc # 函数名缩写 if func_name mean: func_abbr avg elif func_name median: func_abbr med elif func_name std: func_abbr std else: func_abbr func_name[:3] new_col f{prefix}{base_name}_{func_abbr} new_cols.append(new_col) df.columns new_cols return df # 使用示例 result df.groupby(merchant_category).agg({ transaction_amount: [mean,median], processing_fee: [min,max] }) result flatten_agg_columns(result, prefixbiz_) # 输出列名biz_ta_avg, biz_ta_med, biz_pf_min, biz_pf_max实操心得这个函数我们放在common/agg_utils.py里所有聚合任务强制导入。上线半年来因列名不一致导致的下游解析失败从每月12次降到0次。关键技巧是缩写规则必须写死在文档里且禁止开发人员手动改。我们甚至用pytest写了校验用例确保flatten_agg_columns()输出的列名符合正则^biz_[a-z]{2}_[a-z]{3}$。3.2 自定义聚合lambda够用吗不它正在杀死你的可维护性原文用lambda x: x.max() - x.min()计算范围简洁是真简洁但问题也真严重问题1无法序列化lambda函数不能被pickle序列化意味着无法用Dask或Spark分布式执行。我们曾把一个lambda聚合任务迁移到Dask集群结果报错Cant pickle function lambda at 0x...折腾两天才发现根源。问题2无调试入口当计算结果异常如某商户范围值为负数你没法在lambda里加断点或print。而named function可以轻松插入logging.debug(fInput series: {x.tolist()})。问题3业务逻辑黑箱lambda x: x.max() - x.min()只告诉你“算差值”但业务上需要知道“这个差值是否剔除了退款订单”、“是否排除了测试交易merchant_id以TEST_开头”我们强制推行的自定义聚合规范import logging from typing import Union, Callable def transaction_range(series: pd.Series, exclude_refunds: bool True, exclude_test_merchants: bool True, refund_flag_col: str is_refund, merchant_id_col: str merchant_id) - float: 计算交易金额范围最大值-最小值 business_rule: 根据《反欺诈操作手册》第3.2条范围计算需排除退款及测试商户交易 audit_trail: 此函数版本v2.12024-03-15由风控部张工确认逻辑 logger logging.getLogger(__name__) # 1. 创建工作副本避免修改原始数据 work_series series.copy() # 2. 排除退款如果提供退款标识列 if exclude_refunds and refund_flag_col in series.index: # 注意这里假设series是DataFrame的一列需从原始df获取上下文 # 实际中我们传入完整df用mask筛选 pass # 3. 排除测试商户需完整df上下文故此函数不单独使用 # 真实场景中此函数作为apply的参数接收完整group # 因此我们重写为class-based aggregator return work_series.max() - work_series.min() # 更优解面向对象聚合器推荐 class BusinessAggregator: def __init__(self, config: dict None): self.config config or {} self.logger logging.getLogger(self.__class__.__name__) def range_excluding_refunds(self, group_df: pd.DataFrame) - float: 专用于groupby.apply的范围计算 # group_df是当前分组的完整DataFrame可访问所有列 valid_mask ~group_df.get(is_refund, pd.Series([False]*len(group_df))) if merchant_id in group_df.columns: valid_mask ~group_df[merchant_id].str.startswith(TEST_) amounts group_df.loc[valid_mask, amount] if len(amounts) 2: self.logger.warning(fGroup has 2 valid transactions, returning NaN. Group size: {len(group_df)}) return np.nan return amounts.max() - amounts.min() # 使用方式 aggregator BusinessAggregator() result df.groupby(merchant_category).apply(aggregator.range_excluding_refunds)注意事项在银行系统里所有自定义聚合函数必须带business_rule注释明确引用业务文档条款。去年审计时监管员抽查了17个自定义函数全部凭此注释快速验证了合规性没花一分钟查代码逻辑。3.3 滚动窗口聚合为什么window3不是魔法数字而是业务心跳原文示例用rolling(window3).mean()计算3日均值但没说清为什么是3不是5这个数字背后是业务节奏。在我们信用卡中心3日滚动均值对应“欺诈检测的黄金响应窗口”——根据历史数据92%的盗刷行为在首次异常交易后48小时内发生第二次所以3日窗口能覆盖99%的早期预警场景。但window参数只是冰山一角。生产中必须处理的五个现实问题问题1时间对齐陷阱df.set_index(date).rolling(3D)按日历日滚动但交易系统只在工作日生成数据。如果遇到周末rolling(3D)会包含空值导致均值失真。正确做法是用rolling(3, min_periods2)并指定closedright确保只包含当前日及之前数据。问题2数据新鲜度控制滚动计算必须基于“已确认数据”。我们给每条交易打上data_status标签raw/confirmed/adjusted滚动聚合只取data_statusconfirmed的记录。这个过滤必须在rolling前完成否则rolling().apply()里再过滤会导致窗口大小不一致。问题3性能优化对10亿行数据做rolling(30).mean()pandas默认用O(n*w)算法w为窗口大小耗时不可接受。我们改用numba加速from numba import jit import numpy as np jit(nopythonTrue) def rolling_mean_numba(arr: np.ndarray, window: int) - np.ndarray: Numba加速的滚动均值比pandas快8倍 n len(arr) result np.full(n, np.nan) if n window: return result # 首先计算第一个窗口 window_sum np.sum(arr[:window]) result[window-1] window_sum / window # 滑动窗口减去离开的加上进入的 for i in range(window, n): window_sum window_sum - arr[i-window] arr[i] result[i] window_sum / window return result # 应用到分组 def fast_rolling_mean(group: pd.Series, window: int 30) - pd.Series: arr group.values result rolling_mean_numba(arr, window) return pd.Series(result, indexgroup.index)问题4边界值处理策略开头几行的NaN不能简单fillna(0)那会扭曲趋势。我们采用三种策略forward_fill: 适用于平稳业务如基础存款利率min_periods1: 适用于启动期业务如新上线产品interpolate(methodlinear): 适用于连续型指标如账户余额问题5多列协同滚动有时需要“交易金额滚动均值”和“交易笔数滚动均值”的比值。如果分开计算再相除会因NaN对齐问题出错。必须用rolling().apply()一次性计算def amount_per_transaction_rolling(group_df: pd.DataFrame) - float: 计算滚动期内单笔交易均值 amounts group_df[amount].rolling(30, min_periods10).sum() counts group_df[transaction_count].rolling(30, min_periods10).sum() return (amounts / counts).iloc[-1] # 返回最后一个值 # 注意这里group_df必须按时间排序且索引为datetime result df.sort_values(date).groupby(customer_id).apply(amount_per_transaction_rolling)3.4 扩展窗口聚合cumsum不是累加而是业务生命周期刻度expanding().sum()看似简单但“从开始累计”这个定义在金融场景里极其危险。比如计算客户累计交易额如果从2010年开户日起算到2024年数据量巨大且早期数据质量差当时没采集merchant_id直接cumsum会污染整个指标。我们的“扩展窗口”实践规范规范1锚点动态化不用expanding()而用rolling(windowlen(df), min_periods1)但起始点由业务规则决定。例如# 客户价值累计从首次交易日开始而非开户日 first_txn_date df.groupby(customer_id)[date].min() # 构建每个客户的扩展窗口起始点 df[expanding_start] df[customer_id].map(first_txn_date) # 然后用自定义函数计算规范2状态感知累计累计不是机械加法要识别业务状态变化。例如“客户忠诚度积分”需在以下情况重置客户降级为普通会员status从gold变basic账户休眠超90天last_active_date距今90主动注销is_closedTrue我们用状态机实现def loyalty_points_expanding(group_df: pd.DataFrame) - pd.Series: points [] current_total 0 last_active None for idx, row in group_df.sort_values(date).iterrows(): # 检查是否需重置 if (row[status] basic or (last_active and (row[date] - last_active).days 90) or row[is_closed]): current_total 0 current_total row[points_earned] points.append(current_total) last_active row[date] return pd.Series(points, indexgroup_df.index)规范3监管合规累计某些指标需按监管周期累计如“季度累计投诉量”。这时扩展窗口必须与日历对齐# 按自然季度累计 df[quarter] df[date].dt.to_period(Q) # 对每个季度内数据cumsum但跨季度不延续 df[qtr_cumsum] df.groupby([customer_id,quarter])[complaints].cumsum()3.5 多级分组与重塑unstack不是终点而是新聚合的起点原文df.groupby([region,product]).mean().unstack()生成宽表但生产中这往往是下一个聚合的输入。比如我们生成了region × product的销售额矩阵后还要计算各地区销售额占全国比重需先sum(axis1)再广播除法各产品线在各地区的增长率需与上期矩阵做diffTOP3产品地区集中度需对每行排序取前3这些操作在unstack后的DataFrame上进行比在MultiIndex上容易得多。但要注意三个坑坑1缺失组合的填充策略unstack(fill_value0)填0看似合理但“某地区无某产品销售”和“某地区该产品销售额为0”业务含义不同。我们用fill_valuenp.nan再用业务规则填充# 规则新设地区开业30天的缺失产品用全国均值填充 new_regions df[df[open_date] (pd.Timestamp.now() - pd.DateOffset(days30))][region].unique() national_avg df[revenue].mean() wide_df.loc[new_regions] wide_df.loc[new_regions].fillna(national_avg)坑2列名语义化unstack()后列名是(product, Widget)但下游需要product_widget_revenue。我们用add_prefix()和add_suffix()组合# 先unstack product维度 wide result.unstack(product, fill_value0) # 列名变为 MultiIndex: (revenue, Widget), (revenue, Gadget) # 展平并添加业务前缀 wide.columns [frev_{col[1].lower()} for col in wide.columns] # 结果rev_widget, rev_gadget坑3内存爆炸预防当分组维度多如[region, product, channel, month]unstack后列数region数×product数×channel数×month数可能超百万列。这时必须用分块unstack# 按region分块处理 all_results [] for region in df[region].unique(): region_df df[df[region] region] # 对每个region只unstack product和channel regional_wide region_df.groupby([product,channel]).mean().unstack([product,channel]) regional_wide.columns [f{regional_wide.columns.names[0]}_{c[0]}_{c[1]} for c in regional_wide.columns] regional_wide[region] region all_results.append(regional_wide) final_wide pd.concat(all_results, ignore_indexTrue)3.6 综合实战银行信用卡客户价值动态评估系统现在把前面所有模式串起来还原我们真实的“客户价值动态评估”系统。这不是玩具数据而是每天处理1200万条交易的真实流水。业务需求全景为营销部提供客户价值热力图维度包括[客户等级, 地区, 商户类别, 时间窗口]指标包括[近30天交易额, 近30天交易频次, 近30天单笔均值, 近30天金额变异系数, 近30天高价值交易占比]且所有指标需支持“滚动更新”每小时刷新和“快照回溯”可查任意历史时点。系统架构分层层级技术选型关键设计数据接入层Apache Kafka Flink实时交易流每条消息含txn_id, customer_id, amount, category, timestamp, is_refund清洗聚合层Pandas Numba Dask用Dask DataFrame分片处理Numba加速滚动计算存储层ClickHouse按customer_id分片timestamp排序支持毫秒级聚合查询服务层FastAPI提供REST APIGET /value/{customer_id}?as_of2024-03-15核心聚合代码精简版import pandas as pd import numpy as np from numba import jit from typing import Dict, List, Tuple class CustomerValueAggregator: def __init__(self, config: Dict): self.config config self.window_days config.get(rolling_window_days, 30) self.high_value_threshold config.get(high_value_threshold, 300.0) staticmethod jit(nopythonTrue) def _rolling_std_numba(arr: np.ndarray, window: int) - np.ndarray: # Numba加速的标准差计算略同前文rolling_mean_numba pass def calculate_metrics(self, df: pd.DataFrame) - pd.DataFrame: 输入按customer_id分组的交易DataFrame已按时间排序 输出单客户多指标DataFrame索引为时间列为指标 # 1. 基础统计滚动30天 df_sorted df.sort_values(timestamp) df_sorted df_sorted.set_index(timestamp) # 交易额滚动统计 rev_rolling df_sorted[amount].rolling( f{self.window_days}D, min_periodsmax(10, self.window_days//3), closedright ) metrics pd.DataFrame(indexdf_sorted.index) metrics[rolling_revenue] rev_rolling.sum() metrics[rolling_count] rev_rolling.count() metrics[rolling_avg] rev_rolling.mean() # 2. 变异系数标准差/均值需处理均值为0 std_vals self._rolling_std_numba( df_sorted[amount].values, self.window_days ) # 将std数组对齐到metrics索引 metrics[rolling_std] pd.Series(std_vals, indexmetrics.index) metrics[coefficient_of_variation] np.where( metrics[rolling_avg] ! 0, metrics[rolling_std] / metrics[rolling_avg], np.nan ) # 3. 高价值交易占比需在rolling窗口内计算 # 这里用apply因为需在每个窗口内做布尔运算 def high_value_ratio(series): if len(series) 0: return np.nan high_val_count (series self.high_value_threshold).sum() return high_val_count / len(series) # 注意rolling().apply()对每个窗口应用函数 metrics[high_value_ratio] rev_rolling.apply(high_value_ratio) # 4. 业务维度扩展地区、商户类别 # 这里用groupbyunstack生成交叉维度 # 为节省篇幅展示核心逻辑 region_cat_pivot df_sorted.groupby([ pd.Grouper(freqf{self.window_days}D), region, category ])[amount].sum().unstack([region,category], fill_value0) # 5. 最终整合将所有指标合并为宽表 # metrics是时间序列region_cat_pivot是周期聚合需对齐时间 # 实际中用resample(D).ffill()对齐 return metrics # 使用示例对单客户聚合 aggregator CustomerValueAggregator({ rolling_window_days: 30, high_value_threshold: 300.0 }) # 假设df_customer是某客户的100万条交易 customer_metrics aggregator.calculate_metrics(df_customer) # 输出DataFrame索引为datetime列包括rolling_revenue等避坑经验实录经验1时间窗口对齐是生死线我们曾因rolling(30D)和resample(D)的时区处理不一致导致凌晨1点的聚合结果包含当日0点前数据被风控系统误判为“夜间异常交易”。解决方案所有时间操作统一用UTC时区入库时即转换避免本地时区干扰。经验2内存监控必须前置在Dask集群上我们给每个worker加了内存钩子from dask.distributed import Client client Client(memory_limit4GB) # 注册内存使用回调 def log_memory_usage(future): if future.status finished: mem future.result().memory_usage(deepTrue).sum() if mem 3e9: # 超3GB报警 logging.warning(fHigh memory usage: {mem/1e9:.2f} GB) client.register_worker_plugin(log_memory_usage)经验3结果校验自动化每次聚合后强制运行校验def validate_aggregation_result(result_df: pd.DataFrame) - bool: 聚合结果业务校验 # 1. 交易额不应为负 if (result_df[rolling_revenue] 0).any(): return False # 2. 变异系数应在合理范围0-5过高说明数据异常 cv result_df[coefficient_of_variation] if cv.dropna().max() 5: return False # 3. 高价值占比应在0-1之间 if not ((result_df[high_value_ratio] 0) (result_df[high_value_ratio] 1)).all(): return False return True3.7 高级模式条件聚合与分层聚合的嵌套艺术最后分享一个压箱底技巧当业务需求出现“如果A则X否则Y”的嵌套逻辑时如何避免写满屏if-else。场景某理财产品销售系统需计算“客户购买力指数”规则如下如果客户近30天交易额 10万元且至少有3笔交易则指数 交易额 × 1.2否则如果客户有VIP等级则指数 VIP基础分 × 交易频次否则指数 交易额 × 0.8错误做法用apply()遍历每行判断# ❌ 千万别这么写性能极差 def calc_power_index(row): if row[rolling_revenue] 100000 and row[rolling_count] 3: return row[rolling_revenue] * 1.2 elif row[vip_level] 0: return row[vip_base_score] * row[rolling_count] else: return row[rolling_revenue] * 0.8 df[power_index] df.apply(calc_power_index, axis1) # 100万行要12分钟正确解法向量化条件聚合def vectorized_power_index(df: pd.DataFrame) - pd.Series: 向量化计算购买力指数100万行0.5秒 # 初始化结果Series result pd.Series(np.nan, indexdf.index) # 条件1高价值客户 cond1 (df[rolling_revenue] 100000) (df[rolling_count] 3) result.loc[cond1] df.loc[cond1, rolling_revenue] * 1.2 # 条件2VIP客户注意需排除已满足条件1的 cond2 (~cond1) (df[vip_level] 0) result.loc[cond2] df.loc[cond2, vip_base_score] * df.loc[cond2, rolling_count] # 条件3普通客户 cond3 ~(cond1 | cond2) result.loc