企业官网建设流程全解析

1. 项目概述：为什么多维聚合不是“加个groupby”就能搞定的事

我在银行数据平台组干了八年，从最早用SQL写几十行嵌套子查询做客户分层，到现在带团队设计日均处理20亿条交易的实时聚合管道。说实话，刚看到“多维聚合”这个词时，我也以为就是GROUP BY region, product, category再套几个SUM()和AVG()——直到某天凌晨三点，风控系统因为一个没处理好的滚动窗口计算，把三家正常商户标成了高风险，触发了自动冻结流程。那次事故后我们复盘发现：90%的数据问题不来自算法逻辑错误，而源于对聚合行为边界的误判。比如你用rolling(window=7).mean()算客户周均消费，但没意识到pandas默认会按原始索引顺序滚动，而真实交易数据是按时间戳排序的；又比如用unstack()生成区域-产品矩阵时，漏掉了fill_value=0参数，导致下游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(drop=True)确保顺序，再检查rolling(window=7)的首尾索引是否连续；
• 边界验证：对业务敏感字段（如手续费），聚合后立即执行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_dict=SEMANTIC_DICT)，该函数会检查当前数据中实际出现的组合数是否符合cardinality预期，并根据sparsity_rule决定是否允许fill_value参数生效。这种设计让技术实现和业务规则彻底解耦——当市场部新增一个销售区域时，只需更新SEMANTIC_DICT，所有聚合逻辑自动适配。

2.3 滚动与扩展窗口不是时间函数，而是状态机

原文把rolling()和expanding()归为时间序列操作，这在教学场景没问题，但在生产环境会埋雷。关键区别在于：滚动窗口是局部状态，扩展窗口是全局状态。以滚动7日均值为例，它的计算只依赖当前行及前6行，内存占用恒定；而扩展窗口需要从数据起点累积计算，当处理10亿行数据时，内存会随行数线性增长。更危险的是状态一致性——如果数据按customer_id分组后做滚动计算，但分组内时间戳不连续（比如客户A在1月1日和1月10日有交易，中间缺8天），rolling(window=7)仍会强行计算最近7条记录的均值，结果完全失真。

我的做法是把窗口计算重构为显式状态机。核心原则：所有窗口操作必须声明state_key（状态标识）和time_col（时间基准）。例如：

# 错误示范：隐式状态，无法处理时间断点 df.groupby('customer_id')['amount'].rolling(window=7).mean() # 正确实践：显式状态机，自动处理时间断点 def rolling_window_stateful(df, state_key='customer_id', time_col='trans_time', window_days=7, agg_func='mean'): # 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( window=window_days, min_periods=1).agg(agg_func).reset_index(level=[0,1], drop=True) 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, strategy='semantic', max_length=30): """ 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, strategy='semantic') 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(nopython=True) 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, weights=weights) 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(nopython=True)强制编译，否则会回退到Python解释模式。且传入参数必须是numpy数组，不能是pandas Series——这是很多教程忽略的关键点。

3.3 滚动窗口的NaN处理：不是补0那么简单

原文提到“前两行显示NaN值是因为3日窗口需要三个数据点”，但没说清楚为什么必须是NaN，以及如何科学处理它。在风控场景中，用fillna(method='ffill')向前填充会制造虚假趋势（比如把第1天的异常高交易额延续到第2、3天），而dropna()又会丢失关键起始点。我的经验是：NaN是系统发出的健康信号，应该分级响应而非粗暴填充。

具体策略：

• Level 1（诊断层）：在计算滚动窗口后，立即统计NaN比例。若某分组NaN_ratio > 0.3，触发数据质量告警；
• Level 2（业务层）：对不同业务场景配置差异化填充策略：
  • 欺诈检测：用bfill()向后填充（假设首日异常是设备故障，次日数据可信）；
  • 营收预测：用interpolate(method='linear')线性插值（假设营收呈平滑变化）；
  • 合规报送：严格保留NaN，并在元数据中标记data_gap_flag=True；
• Level 3（工程层）：用min_periods参数控制最小有效点数。例如rolling(window=7, min_periods=4)表示只要有4个有效点就计算均值，避免全空。

代码实现：

def robust_rolling(df, time_col, value_col, window_days=7, fill_strategy='business', min_periods=1): """ 健壮的滚动计算，内置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( window=f'{window_days}D', # 用字符串形式支持日期偏移 min_periods=min_periods ).mean() # 分级填充 if fill_strategy == 'none': pass elif fill_strategy == 'forward': rolling_result = rolling_result.fillna(method='ffill') elif fill_strategy == 'backward': rolling_result = rolling_result.fillna(method='bfill') elif fill_strategy == 'interpolate': rolling_result = rolling_result.interpolate(method='linear') elif fill_strategy == 'business': # 业务策略：欺诈检测用bfill，营收用interpolate if 'fraud' in value_col.lower(): rolling_result = rolling_result.fillna(method='bfill') else: rolling_result = rolling_result.interpolate(method='linear') return rolling_result.reset_index(name=f'{value_col}_rolling_{window_days}d') # 使用示例：欺诈检测场景 fraud_rolling = robust_rolling( df_transactions, time_col='date', value_col='amount', window_days=7, fill_strategy='business' )

4. 完整实操流程：从原始交易数据到高管决策看板

4.1 数据准备与质量基线校验

所有聚合操作前，必须完成三道质量防火墙。这不是可选项，而是我所在团队的上线红线。以信用卡交易数据为例，原始数据来自核心银行系统，包含trans_id,customer_id,trans_time,merchant_category,amount,fee等字段。防火墙设计如下：

防火墙1：Schema一致性校验
检查字段类型、空值率、唯一性约束是否符合数据字典。例如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(f"fee/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(f"Schema校验失败：{'; '.join(issues)}") return True # 执行校验 validate_schema(df_transactions)

防火墙2：时间序列完整性校验
检查时间戳是否连续、是否存在重复或乱序。这对滚动窗口至关重要：

def validate_time_series(df, time_col='trans_time', freq='D'): """ freq: 'D'=日频, 'H'=小时频, 'T'=分钟频 """ df_sorted = df.sort_values(time_col).copy() # 检查重复时间戳 dup_times = df_sorted.duplicated(subset=[time_col], keep=False) 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_col='date', freq='D')

防火墙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, on='customer_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( window='7D', # 使用日期偏移，自动处理不规则时间间隔 min_periods=3 # 至少3个点才计算，避免噪声 ).mean().reset_index(level=0, drop=True) # 合并回原数据 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(level=0, drop=True).values # 计算累计交易笔数 cumulative_count = df_enhanced.sort_values(['customer_id', 'date']).groupby('customer_id')['amount'].expanding().count() df_enhanced['cumulative_count'] = cumulative_count.reset_index(level=0, drop=True).values

Step 5：多维交叉分析（回答“结构如何”）
生成客户-商户类别矩阵，支持交叉销售分析：

# 创建交叉表：客户为行，商户类别为列，值为平均交易额 crosstab_avg = df_enhanced.groupby(['customer_id', 'category'])['amount'].mean().unstack(fill_value=0) # 添加行列总计 crosstab_avg['Total'] = crosstab_avg.sum(axis=1) crosstab_avg.loc['Total'] = crosstab_avg.sum(axis=0) # 计算占比矩阵（每行百分比） crosstab_pct = crosstab_avg.div(crosstab_avg['Total'], axis=0).multiply(100).round(1) crosstab_pct = crosstab_pct.drop('Total', axis=1) # 移除总计列，只保留占比

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(days=7)) ] 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, axis=1) 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(normalize=True).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(days=30)] recent_revenue = recent_data['amount'].sum() prior_data = df_enhanced[ (df_enhanced['date'] >= df_enhanced['date'].max() - pd.Timedelta(days=60)) & (df_enhanced['date'] < df_enhanced['date'].max() - pd.Timedelta(days=30)) ] 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_time=None): """生成聚合结果元数据""" if execution_time is None: execution_time = datetime.now() metadata = {