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

我在银行数据平台组干了八年，从最早用SQL写几十行嵌套子查询做客户分层，到后来带团队重构整个风险指标计算引擎，踩过的坑比写的代码还多。今天聊的这个主题——“多维聚合中的数据操作”，听起来像教科书里的一个章节标题，但实际在生产环境里，它直接决定着风控模型能不能按时上线、月度经营分析报告能不能准时发出、甚至监管报送数据有没有偏差。我见过太多团队把 df.groupby().agg() 当成万能胶水，结果在千万级交易流水里跑出内存溢出，在跨季度滚动计算时发现时间窗口对不上，在给高管做区域-产品交叉报表时被财务部打回来三次——就因为“North Region的Gadget销售额和我们ERP系统差了0.3%”。

核心关键词就三个： 多维聚合、生产级、业务语义 。不是教你语法，而是告诉你为什么 agg({'amount': ['mean', 'std']}) 要写成这样，而不是先 mean() 再 std() ；为什么 unstack() 之后必须加 fill_value=0 ，否则下游BI工具会报错；为什么滚动窗口的 min_periods=3 比默认的 None 更安全，哪怕你文档里没写这一句。这些细节背后全是血泪教训：某次大促期间，因未处理滚动平均的前两行NaN，导致实时欺诈预警漏判三笔高危交易；另一次季度结账，因 unstack() 未设填充值，Excel导出后空单元格被自动转成0，让区域经理误判了新品类渗透率。

适合谁看？如果你正在用pandas做以下任何一件事，这篇文章就是为你写的：

• 每天手动合并5张不同维度的groupby结果表，然后用VLOOKUP对齐；
• 写完自定义函数后不敢交给同事维护，因为docstring里写着“此处逻辑需结合2023年Q4风控策略文档第7页”；
• 看到 rolling(window=7).mean() 就抄，但从没想过为什么是7不是5或10；
• 被业务方问“上个月南区零售类客户的中位数消费额是多少”时，得现查PPT翻历史口径定义。

这不是语法速查手册，而是一份从银行核心系统、风控引擎、运营看板里榨出来的实战笔记。接下来我会拆解五个真实场景：多列异构聚合如何避免结果错位、自定义函数怎样写才经得起审计、滚动窗口的边界陷阱怎么填、扩展窗口在YTD报表里的致命细节、多级分组后unstack的变形术。每个部分都附带我在线上环境抓包的真实日志、参数选择的推演过程，以及——最关键的是，那些从来不会写在官方文档里的“不能这么干”的禁忌。

2. 多列异构聚合：为什么一次agg胜过三次merge

2.1 问题本质：维度对齐的隐形战争

先说个扎心事实：90%的pandas聚合性能问题，根源不在数据量，而在 结果结构失控 。你试过这样写吗？

# ❌ 危险示范：三次独立groupby + merge
mean_amt = df.groupby('merchant_category')['transaction_amount'].mean()
std_amt = df.groupby('merchant_category')['transaction_amount'].std()
min_fee = df.groupby('merchant_category')['processing_fee'].min()
result = mean_amt.to_frame('mean_amt').join(std_amt.to_frame('std_amt')).join(min_fee.to_frame('min_fee'))

表面看结果没错，但暗藏三重危机：

1. 索引错位风险 ：如果某类商户在 std_amt 计算中因全空值返回NaN行，而 mean_amt 有值， join 后该行数据就错位了；
2. 重复计算开销 ：三次遍历DataFrame，对千万级数据就是三倍CPU和内存；
3. 维护地狱 ：当业务要求增加 median_amt 时，你得改四行代码，且无法保证新列和其他列的索引顺序绝对一致。

真正的生产级写法，是让pandas在单次扫描中完成所有计算。原文示例中这行代码才是关键：

result = df.groupby('merchant_category').agg({
    'transaction_amount': ['mean', 'median'], 
    'processing_fee': ['min', 'max']
})

但注意！这里藏着一个新手必踩的坑： 列名映射必须严格对应原始字段名 。我曾遇到一个案例，原始数据中 processing_fee 列实际叫 proc_fee （上游ETL脚本缩写），结果 agg 后该列直接消失，而错误提示是 KeyError: 'processing_fee' ——但你在DataFrame里 df.columns 根本看不到这个错，因为缩写列名在 print(df.head()) 里显示正常，只有 agg 时才暴露。

2.2 结构解析：分层列名的生存指南

运行上述代码后，输出是这样的：

transaction_amount     processing_fee
mean      median      min      max
Dining     55.10       52.30     1.36     2.03
Retail    150.78      125.50     2.68     6.31
Travel    221.78      189.60     5.69     9.60

这个双层列结构（MultiIndex）不是设计缺陷，而是pandas为防止歧义做的精密防护。想象一下，如果所有结果压平成 ['mean', 'median', 'min', 'max'] ，当你后续要取“零售类商户的手续费最小值”时，代码是 result['min'] 还是 result['processing_fee_min'] ？前者会同时拿到 transaction_amount_min （不存在）和 processing_fee_min （存在），引发 KeyError 。

实操心得 ：

• 取特定组合值永远用元组索引： result[('transaction_amount', 'mean')] 或 result['transaction_amount']['mean'] ；
• 扁平化列名时禁用 reset_index() 简单粗暴： result.columns = ['_'.join(col).strip() for col in result.columns.values] ，否则 ('transaction_amount', 'mean') 变成 transaction_amount_mean ，而 ('processing_fee', 'min') 变成 processing_fee_min ，命名风格不统一；
• 最佳实践是用 droplevel(0) 降级外层： result.droplevel(0, axis=1) ，得到纯 ['mean','median','min','max'] 列，前提是确认所有聚合列都来自同一业务域（如全是金额类指标）。

提示：当需要将结果喂给Tableau或Power BI时，务必提前扁平化列名。这些工具无法识别pandas的MultiIndex，会把整行当做一个字段值，导致透视表完全失效。

2.3 生产级增强：缺失值与类型安全

原文示例没提一个致命细节： 当某组数据全为空时， agg 会返回NaN，但类型可能突变 。比如 transaction_amount 是float64，但 agg(['mean','count']) 中 count 返回int64，混合后整列变成object类型，后续 fillna(0) 会失败。

我的解决方案是在agg前强制类型校验：

# ✅ 生产级写法：预处理+类型锁定
def safe_agg(df):
    # 步骤1：填充数值型空值（业务规则：空交易额按0计）
    numeric_cols = ['transaction_amount', 'processing_fee']
    for col in numeric_cols:
        if df[col].isnull().sum() > 0:
            df[col] = df[col].fillna(0)  # 注意：此处0是业务约定，非技术默认值
    
    # 步骤2：执行聚合
    result = df.groupby('merchant_category').agg({
        'transaction_amount': ['mean', 'median'],
        'processing_fee': ['min', 'max']
    })
    
    # 步骤3：强制转换为float64（避免int/float混合）
    result = result.astype(np.float64)
    
    return result

这个 safe_agg 函数在我们银行的每日结算任务中跑了三年，零事故。关键点在于： 所有空值填充必须基于业务规则，而非技术便利 。曾有团队为图省事用 df.fillna(method='ffill') ，结果把某类商户的手续费空值填充成上一类的值，导致风控阈值计算偏差，被监管问询。

3. 自定义聚合函数：业务逻辑的代码化封装

3.1 Lambda的甜蜜陷阱与命名函数的审计价值

原文用lambda演示了范围计算： lambda x: x.max() - x.min() 。这很简洁，但在我经历的两次监管检查中，这种写法都被要求整改。原因很简单： lambda无法被审计追踪 。当合规部门问“这个范围值的计算逻辑依据哪条风控政策？”时，你没法指着一行lambda说“看这里”。而命名函数可以：

def transaction_range(series):
    """
    计算交易金额范围（最大值-最小值）
    依据《2023年反欺诈操作指引》第4.2条：
    "商户类别交易波动性评估须采用极差指标，阈值设定为：零售类≤150元，餐饮类≤200元"
    """
    return series.max() - series.min()

这个docstring不是形式主义。去年某次现场检查，检查员随机抽了3个指标，其中就包括 transaction_range 。他对照docstring里的政策条款号，当场调出PDF原文核对，全程耗时不到2分钟。而隔壁组用lambda写的类似函数，被要求补写说明文档并重新走审批流程，延误了两周上线。

实操心得 ：

• 命名函数必须包含 政策依据、业务场景、阈值说明 三要素；
• 函数名要体现业务含义，而非技术动作。 calc_merchant_volatility() 比 range_calc() 好十倍；
• 避免在lambda中写复杂逻辑。曾见有人写 lambda x: (x > x.quantile(0.95)).sum() / len(x) if len(x)>10 else 0 ，这种代码连作者三个月后都看不懂。

3.2 加权平均的工程实现：不只是np.average

原文的 weighted_average 函数用了 np.linspace(0.5,1.5,len(series)) 生成权重。这在教学示例中没问题，但在生产环境会出大事—— 权重和必须为1 。 np.linspace(0.5,1.5,10) 生成的权重和是10，而 np.average 要求权重和为1，否则结果会放大10倍。

正确写法是显式归一化：

def weighted_average(series):
    """计算加权平均（近期交易权重更高）"""
    if len(series) < 2:
        return series.mean()
    
    # 生成线性递增权重：[1,2,3,...,n]
    weights = np.arange(1, len(series) + 1)
    # 归一化：确保权重和为1
    weights = weights / weights.sum()
    
    return np.average(series, weights=weights)

更关键的是， 权重策略必须可配置 。在我们银行，不同业务线权重规则不同：信用卡中心用线性权重，而对公业务部要求指数衰减（ weights = np.exp(-0.1 * np.arange(len(series)-1, -1, -1)) ）。所以最终方案是：

def weighted_average(series, weight_type='linear', decay_factor=0.1):
    """支持多种权重策略的加权平均"""
    if len(series) < 2:
        return series.mean()
    
    if weight_type == 'linear':
        weights = np.arange(1, len(series) + 1)
    elif weight_type == 'exponential':
        # 指数衰减：最近交易权重最高
        weights = np.exp(-decay_factor * np.arange(len(series)-1, -1, -1))
    else:
        raise ValueError(f"Unsupported weight_type: {weight_type}")
    
    weights = weights / weights.sum()
    return np.average(series, weights=weights)

# 使用时明确指定策略
result = df.groupby('customer_id').agg({
    'amount': lambda x: weighted_average(x, weight_type='exponential', decay_factor=0.05)
})

这样写，业务方随时可以调整 decay_factor ，而无需动核心函数。去年Q3，风控部将衰减因子从0.1调至0.05以降低敏感度，我们只改了一个参数，没动一行聚合逻辑。

3.3 高阶自定义：多条件风险分段的落地

原文Analysis 7的 risk_metrics 函数展示了多输出能力，但生产环境需要更强健的版本。真实场景中，“高价值交易”阈值不是固定300，而是动态的：

def risk_segmentation(series, high_value_threshold=None, percentile=95):
    """
    客户风险分段：高价值交易占比 & 常规交易均值
    动态阈值：若未指定threshold，则取当前客户交易额的95分位数
    """
    if high_value_threshold is None:
        high_value_threshold = series.quantile(percentile / 100)
    
    high_value_mask = series > high_value_threshold
    high_value_count = high_value_mask.sum()
    high_value_pct = (high_value_count / len(series) * 100) if len(series) > 0 else 0
    
    # 常规交易均值：排除高价值交易后的均值
    regular_avg = series[~high_value_mask].mean() if high_value_mask.sum() < len(series) else np.nan
    
    return pd.Series({
        'high_value_threshold': high_value_threshold,
        'high_value_count': high_value_count,
        'high_value_pct': round(high_value_pct, 1),
        'regular_avg': round(regular_avg, 2) if not np.isnan(regular_avg) else 0
    })

# 应用时传入动态阈值
risk_analysis = df_transactions.groupby('customer_id').apply(
    lambda x: risk_segmentation(x['amount'], percentile=90)
)

这个版本解决了三个痛点：

• 阈值可配置（固定值或分位数）；
• 常规交易均值计算时，若全为高价值交易则返回0而非NaN，避免下游计算中断；
• 返回 high_value_threshold 本身，方便业务方验证阈值合理性。

注意： apply 在大数据集上较慢，若性能敏感，可用 transform 替代。但 transform 只能返回同长度序列，无法返回多指标Series，需权衡。

4. 滚动窗口计算：时间序列分析的边界艺术

4.1 窗口大小的业务决策树

原文用 window=3 演示滚动平均，但没解释为什么是3。在银行场景，窗口大小是严肃的业务决策：

业务场景	典型窗口	决策依据
实时欺诈监控	1-3天	检测突发性异常，如单日交易量激增300%
月度经营分析	7天	平滑周末效应，捕捉周度趋势（如周五餐饮消费高峰）
季度风险评估	30天	覆盖完整月度周期，消除节假日扰动
年度客户价值预测	90天	匹配季度财报周期，与营收确认节奏对齐

我们曾因窗口选错付出代价：某次将欺诈监控窗口设为7天，导致一笔连续7天每天小额测试的盗刷交易未被触发（每天金额略低于阈值），直到第8天才爆发大额盗刷。复盘后，将高风险商户的窗口改为1天，低风险商户保持7天，用 groupby('merchant_risk_level') 分层计算。

实操参数选择法 ：

1. 先用 df['date'].diff().dt.days.describe() 看数据时间间隔分布；
2. 若间隔不均（如有些天无交易），用 rolling('7D') （按日历天）而非 rolling(7) （按行数）；
3. 对高频交易（如支付网关），窗口必须用 '30T' （30分钟）而非天。

4.2 NaN陷阱的七种填法与业务含义

原文提到“前两行NaN是预期行为”，但生产中必须主动管理。 rolling 的 min_periods 参数是关键：

# ❌ 危险：默认min_periods=None → 前n-1行全NaN
df['rolling_avg'] = df.groupby('category')['daily_revenue'].rolling(window=3).mean()

# ✅ 生产级：min_periods=1 → 至少1个值就计算
df['rolling_avg_safe'] = df.groupby('category')['daily_revenue'].rolling(
    window=3, min_periods=1
).mean()

但 min_periods=1 只是开始，后续填充策略必须匹配业务：

NaN位置	业务含义	推荐填充法	示例代码
开头N行	数据不足，无历史参考	用当日值填充（最保守）	.fillna(method='bfill')
中间行	系统故障导致数据丢失	用前后均值线性插值	.interpolate(method='linear')
结尾行	预测场景需延续趋势	用最后有效值前向填充	.fillna(method='ffill')
全列NaN	分组内数据不足（如新商户）	填-1并标记“数据不可用”	.fillna(-1).astype(int) + 新增状态列

我们线上系统采用组合策略：

def safe_rolling_avg(series, window=7, min_periods=3):
    """带多重NaN处理的滚动平均"""
    rolling_result = series.rolling(window=window, min_periods=min_periods).mean()
    
    # 步骤1：对min_periods仍不足的行，用当日值填充（最安全）
    rolling_result = rolling_result.fillna(series)
    
    # 步骤2：若仍有NaN（如series本身为NaN），标记为-999
    rolling_result = rolling_result.fillna(-999)
    
    return rolling_result

# 应用
df_ts['rolling_avg'] = df_ts.groupby('category')['daily_revenue'].apply(
    lambda x: safe_rolling_avg(x, window=7)
)

4.3 分组滚动的隐藏雷区：索引对齐

原文代码 df_ts.groupby('category')['daily_revenue'].rolling(window=3).mean().reset_index(level=0, drop=True) 看似正确，但有个致命隐患： 当分组内数据量不同时， reset_index 可能导致索引错乱 。

真实案例：某次处理跨境支付数据， category 分组中“Retail”有1000行，“Dining”仅50行。 rolling 后 reset_index(level=0, drop=True) 将 category 索引丢弃，但 rolling 结果的索引顺序与原始DataFrame不一致，导致 rolling_avg 值被错配到其他日期。

正确解法是用 transform 保持索引对齐：

# ✅ 绝对安全：transform确保索引1:1对应
df_ts['rolling_avg'] = df_ts.groupby('category')['daily_revenue'].transform(
    lambda x: x.rolling(window=3, min_periods=1).mean()
)

transform 返回与原DataFrame等长的Series，索引完全对齐，彻底规避错配风险。虽然 transform 比 agg 稍慢，但在生产环境， 数据正确性永远优先于毫秒级性能 。

5. 扩展窗口与多级分组：从累计求和到决策矩阵

5.1 扩展窗口的YTD陷阱：时序排序是生死线

原文 expanding().sum() 代码隐含一个致命假设： 数据已按时间排序 。但现实是，ETL任务可能乱序写入，或上游系统时间戳有误差。

我亲眼见过一次事故：某日批量任务中，一条2024-03-15的交易数据因网络延迟，晚于2024-03-20的数据写入数据库。 expanding().sum() 按入库顺序计算，导致3月15日的累计值包含了3月20日的交易，YTD报表整整高估了230万元。

生产级防护三步法 ：

1. 强制按时间排序： df = df.sort_values(['category', 'date']).reset_index(drop=True) ；
2. 验证时间单调性： if not df.groupby('category')['date'].is_monotonic_increasing.all(): raise ValueError("Time series not sorted!") ；
3. 使用 expanding 时指定 method='table' （pandas 1.4+），它会按索引顺序而非分组顺序计算。

最终代码：

def safe_expanding_sum(df, date_col='date', group_col='category', value_col='daily_revenue'):
    """防乱序的扩展累计和"""
    # 步骤1：按分组和时间双重排序
    df_sorted = df.sort_values([group_col, date_col]).reset_index(drop=True)
    
    # 步骤2：验证时间单调性
    for name, group in df_sorted.groupby(group_col):
        if not group[date_col].is_monotonic_increasing:
            raise ValueError(f"Group {name} has non-monotonic dates!")
    
    # 步骤3：计算扩展累计和
    df_sorted['cumulative_sum'] = df_sorted.groupby(group_col)[value_col].expanding(
        method='table'
    ).sum().reset_index(level=0, drop=True)
    
    return df_sorted

5.2 多级分组的unstack变形术：从MultiIndex到决策矩阵

原文 unstack() 示例简单直接，但生产中常需处理更复杂场景。比如业务方要求：“按地区、产品、月份三级分组，展示各月销售额，并补零”。

# 原始三级分组
result = df_sales.groupby(['region', 'product', 'month'])['revenue'].sum()

# ❌ 直接unstack会报错：无法unstack多层
# result.unstack() → ValueError: Index has 3 levels, not 2

# ✅ 正确解法：逐层unstack
result_2d = result.unstack('month', fill_value=0)  # 先unstack月份
result_matrix = result_2d.unstack('product', fill_value=0)  # 再unstack产品

但更常见的是 缺失组合的补全问题 。比如“North”地区没有“Travel”产品， unstack 后该单元格为空，而业务要求显示0。

标准解法是 reindex ：

# 获取所有可能的组合
all_regions = ['North', 'South', 'East', 'West']
all_products = ['Widget', 'Gadget', 'Travel', 'Retail']
all_months = ['Jan', 'Feb', 'Mar']

# 创建完整索引
full_index = pd.MultiIndex.from_product(
    [all_regions, all_products, all_months], 
    names=['region', 'product', 'month']
)

# 重新索引并填充0
result_full = result.reindex(full_index, fill_value=0).unstack('month').unstack('product')

实操心得 ：

• unstack 前务必确认索引层级： result.index.names ；
• 补零必须用 reindex 而非 fillna(0) ，后者只填NaN，不补缺失组合；
• 对于超大维度（如100个地区×50个产品）， reindex 内存爆炸，此时改用 pivot_table ：

  result_pivot = df_sales.pivot_table(
      values='revenue', 
      index='region', 
      columns=['product', 'month'], 
      aggfunc='sum',
      fill_value=0
  )
  

5.3 交叉分析的终极形态：crosstab与agg的协同

原文Analysis 5用 unstack() 生成客户-品类矩阵，但实际中常需叠加聚合。比如：“每个客户在各品类的平均交易额，但只统计交易次数≥5的客户”。

# 步骤1：先过滤满足条件的客户
customer_stats = df_transactions.groupby('customer_id').agg({
    'amount': 'mean',
    'category': 'count'
}).rename(columns={'category': 'transaction_count'})

valid_customers = customer_stats[customer_stats['transaction_count'] >= 5].index

# 步骤2：对有效客户做交叉分析
df_valid = df_transactions[df_transactions['customer_id'].isin(valid_customers)]
crosstab = pd.crosstab(
    df_valid['customer_id'], 
    df_valid['category'], 
    values=df_valid['amount'], 
    aggfunc='mean',
    margins=True  # 添加行列总计
).round(2)

print(crosstab)

pd.crosstab 比 unstack 更强大：

• 支持 margins=True 添加总计行/列；
• aggfunc 可传入自定义函数；
• normalize 参数可快速计算占比（ normalize='index' 为行占比， normalize='columns' 为列占比）。

我们风控日报就用 crosstab 生成“高风险客户地域分布热力图”，代码仅三行，却替代了原来20行SQL。

6. 端到端实战：银行信用卡分析流水线的七层防御

6.1 场景还原：从原始数据到高管简报

让我们把前述所有技术点，串成一条真实的银行流水线。输入是60万行信用卡交易日志，目标是生成一份《Q2客户价值分析简报》，包含7个模块：

模块	技术要点	业务价值
1. 客户分层统计	多列异构聚合	快速识别VIP客户（高交易频次+高均值）
2. 品类波动分析	自定义range函数	定位需加强监控的高波动品类（如Travel）
3. 消费趋势监测	滚动7日平均	发现某客户连续7天餐饮消费激增，触发尽调
4. YTD累计消费	扩展窗口求和	计算客户生命周期价值（CLV）
5. 品类偏好矩阵	unstack + crosstab	向营销部推送“南区客户偏爱Groceries”
6. 高管摘要	列名扁平化+四舍五入	一页纸呈现核心指标，避免小数点后三位
7. 风险客户清单	多条件自定义函数	输出高价值交易占比>60%的客户ID

这条流水线每天凌晨2点自动运行，输出Excel和PDF双格式。下面是我亲手写的生产代码，已脱敏但保留全部工程细节。

6.2 代码实现：七层防御的完整流水线

import pandas as pd
import numpy as np
from datetime import datetime

class CreditCardAnalyzer:
    def __init__(self, df_raw):
        self.df = df_raw.copy()
        # 步骤0：基础清洗（生产环境必做）
        self._clean_data()
    
    def _clean_data(self):
        """数据清洗：业务规则驱动"""
        # 交易额必须>0
        self.df = self.df[self.df['amount'] > 0]
        # 时间必须在合理范围（防测试数据污染）
        self.df = self.df[
            (self.df['date'] >= '2024-01-01') & 
            (self.df['date'] <= datetime.now().strftime('%Y-%m-%d'))
        ]
        # 填充空品类（业务规则：空值归为'Other'）
        self.df['category'] = self.df['category'].fillna('Other')
    
    def analysis_1_multi_agg(self):
        """模块1：客户分层统计"""
        # 多列异构聚合：金额类用mean/median/count，费用类用sum
        result = self.df.groupby('customer_id').agg({
            'amount': ['mean', 'median', 'count'],
            'fee': 'sum'
        })
        
        # 扁平化列名 + 四舍五入
        result.columns = ['avg_amount', 'median_amount', 'txn_count', 'total_fee']
        result = result.round({'avg_amount': 2, 'median_amount': 2, 'total_fee': 2})
        
        # 计算衍生指标
        result['avg_fee_rate'] = (result['total_fee'] / result['avg_amount']).round(4)
        
        return result
    
    def analysis_2_volatility(self):
        """模块2：品类波动分析"""
        def calc_range(series):
            return series.max() - series.min()
        
        result = self.df.groupby('category').agg({
            'amount': [calc_range, 'std', 'count']
        })
        result.columns = ['range', 'std_dev', 'txn_count']
        result = result.round(2)
        
        # 标记高波动品类（range > 300 或 std_dev > 100）
        result['high_volatility'] = (
            (result['range'] > 300) | (result['std_dev'] > 100)
        )
        
        return result
    
    def analysis_3_rolling_trend(self, window=7):
        """模块3：消费趋势监测"""
        # 按客户+日期排序（防乱序）
        df_sorted = self.df.sort_values(['customer_id', 'date'])
        
        # 计算滚动平均（min_periods=3防首日NaN）
        df_sorted['rolling_avg'] = df_sorted.groupby('customer_id')['amount'].transform(
            lambda x: x.rolling(window=window, min_periods=3).mean()
        )
        
        # 标记异常：滚动平均 > 整体均值150%
        overall_mean = self.df['amount'].mean()
        df_sorted['is_anomaly'] = df_sorted['rolling_avg'] > (overall_mean * 1.5)
        
        return df_sorted[['customer_id', 'date', 'amount', 'rolling_avg', 'is_anomaly']]
    
    def analysis_4_ytd_cumulative(self):
        """模块4：YTD累计消费"""
        df_sorted = self.df.sort_values(['customer_id', 'date'])
        df_sorted['cumulative_spend'] = df_sorted.groupby('customer_id')['amount'].expanding(
            method='table'
        ).sum().reset_index(level=0, drop=True)
        
        return df_sorted[['customer_id', 'date', 'amount', 'cumulative_spend']]
    
    def analysis_5_preference_matrix(self):
        """模块5：品类偏好矩阵"""
        # 用crosstab生成热力图数据
        matrix = pd.crosstab(
            self.df['customer_id'],
            self.df['category'],
            values=self.df['amount'],
            aggfunc='mean',
            normalize='index',  # 行占比：每个客户在各品类的消费占比
            margins=True
        ).round(3)
        
        # 过滤掉总占比<0.01的客户（减少噪声）
        matrix = matrix[matrix['All'] >= 0.01]
        
        return matrix
    
    def analysis_6_exec_summary(self):
        """模块6：高管摘要"""
        summary = self.df.groupby('customer_id').agg({
            'amount': ['sum', 'mean', 'count'],
            'fee': 'sum'
        })
        summary.columns = ['total_spend', 'avg_txn', 'txn_count', 'total_fee']
        summary = summary.round(2)
        
        # 添加业务标签
        summary['tier'] = pd.cut(
            summary['total_spend'], 
            bins=[0, 10000, 50000, float('inf')], 
            labels=['Bronze', 'Silver', 'Gold']
        )
        
        return summary
    
    def analysis_7_risk_customers(self, high_value_pct=60):
        """模块7：风险客户清单"""
        def risk_score(series):
            high_value_mask = series > series.quantile(0.9)
            high_pct = (high_value_mask.sum() / len(series)) * 100
            return pd.Series({
                'high_value_pct': round(high_pct, 1),
                'risk_score': 1 if high_pct > high_value_pct else 0
            })
        
        result = self.df.groupby('customer_id')['amount'].apply(risk_score)
        return result[result['risk_score'] == 1]
    
    def run_all(self):
        """执行全部分析，返回字典"""
        print("🚀 启动信用卡分析流水线...")
        results = {}
        
        try:
            print("✅ 模块1：客户分层统计")
            results['multi_agg'] = self.analysis_1_multi_agg()
            
            print("✅ 模块2：品类波动分析")
            results['volatility'] = self.analysis_2_volatility()
            
            print("✅ 模块3：消费趋势监测")
            results['rolling_trend'] = self.analysis_3_rolling_trend()
            
            print("✅ 模块4：YTD累计消费")
            results['ytd_cumulative'] = self.analysis_4_ytd_cumulative()
            
            print("✅ 模块5：品类偏好矩阵")
            results['preference_matrix'] = self.analysis_5_preference_matrix()
            
            print("✅ 模块6：高管摘要")
            results['exec_summary'] = self.analysis_6_exec_summary()
            
            print("✅ 模块7：风险客户清单")
            results['risk_customers'] = self.analysis_7_risk_customers()
            
            print("🎉 流水线执行完成！共7个模块")
            
        except Exception as e:
            print(f"❌ 流水线执行失败：{str(e)}")
            raise
        
        return results

# 使用示例
if __name__ == "__main__":
    # 加载数据（此处为模拟）
    np.random.seed(42)
    customers = [f'C{i:03d}' for i in range(1, 1001)]
    categories = ['Groceries', 'Dining', 'Travel', 'Retail', 'Other']
    dates = pd.date_range('2024-04-01', '2024-06-30', freq='D')
    
    data = {
        'date': np.random.choice(dates, 600000),
        'customer_id': np.random.choice(customers, 60