1. 项目概述：为什么多维聚合不是“加总求平均”那么简单

我在银行数据团队干了八年，从刚毕业写SQL跑日报，到后来带三个人的分析小组做风控模型和客户分群。这八年里，我最常被业务方堵在茶水间问的一句话是：“能不能按地区+产品线+客户等级，把上季度的交易额、平均单笔、最大单笔、还有最近30天滚动均值都拉出来？再把高价值客户占比也带上？”——每次听到这种需求，我心里都先叹一口气：又得重写一遍groupby，又得手动merge七八个中间表，又得调窗口函数参数调到凌晨两点。

但其实，这些需求背后藏着一个被严重低估的核心能力： 多维聚合的工程化思维 。它不是pandas文档里那几行示例代码能概括的，而是数据分析师从“能出数”跃升到“能驱动决策”的分水岭。你看到的是一张Excel里的交叉表，背后却是对业务逻辑、数据质量、计算效率、结果可解释性的四重校验。

比如，当风险经理说“我要看餐饮类商户的交易金额范围（max-min）”，他真正在意的不是数学差值，而是这个数字是否超过该类商户历史波动阈值的1.8倍——一旦超标，系统要自动触发人工复核流程。这时候，“range”就不是一个agg函数，而是一个业务规则的量化出口。再比如，运营总监要看“近7天滚动平均交易额”，他真正想判断的是：某客户最近消费节奏是否明显加快？是否该推送分期优惠？这里的“7天”不是随便拍的，而是基于该客群消费周期的统计显著性测试结果——我们实测过3/5/7/14天窗口，只有7天能稳定区分“临时促销刺激”和“真实行为转变”。

这篇文章讲的，就是我在真实生产环境里反复打磨出来的七套组合拳。它们不是教科书里的理想案例，而是我在某股份制银行信用卡中心上线反欺诈模型时，为处理日均2300万笔交易而沉淀下来的实战方法论。没有花哨的算法，全是用pandas原生功能搭出来的、经得起审计、扛得住并发、业务方能看懂的硬核操作。如果你还在用for循环遍历分组、还在为列名嵌套层级头疼、还在为滚动窗口的NaN值手动fillna，那接下来的内容，会直接帮你省下每周至少6小时的重复劳动。

2. 多维聚合的整体设计思路：从“堆砌函数”到“构建分析流水线”

2.1 为什么不能只用基础groupby？三个血泪教训

刚入行时，我也觉得 df.groupby('col').sum() 够用了。直到第一次做季度经营分析，被财务部打回来三次：

• 第一次 ：他们要“各区域各产品线的收入、毛利、订单量、客单价、复购率”，我写了5个独立groupby，merge时因索引对不齐导致某区域数据错位，最终报表里华东区零售收入虚高27%；
• 第二次 ：风控部要“近30天滚动逾期率”，我用 rolling(30).mean() ，但没考虑周末无交易导致窗口内有效数据不足，结果周一的指标突然跳变，被质疑模型失灵；
• 第三次 ：高管要看“客户生命周期价值（LTV）”，我按月groupby后cumsum，但没处理新老客户混排问题——刚注册的用户和五年老客挤在同一时间轴上，曲线完全失真。

这三次翻车让我彻底明白： 多维聚合的本质，是把业务问题翻译成数据操作语言的过程，而pandas只是工具，不是答案本身。 真正的难点在于三件事：

1. 维度解耦 ：把“地区+产品线+客户等级”这种天然耦合的业务概念，拆解成可独立验证、可组合叠加的数据结构；
2. 时序对齐 ：确保所有时间窗口计算（滚动、扩展、同比）都基于同一时间基准，且能识别并处理数据断点；
3. 结果塑形 ：让输出格式直接匹配下游使用场景——给BI工具的要扁平列名，给Excel的要行列分明，给API的要JSON友好。

所以我的设计原则很朴素： 每个聚合操作必须回答三个问题——它解决什么业务问题？输入数据是否满足前提条件？输出结果能否被非技术人员直接理解？ 比如 unstack() 从来不只是为了“好看”，而是为了让销售总监打开表格时，第一眼就能看出“华南区Widget产品比华北区高12%，但Gadget低8%”——这种信息密度，是MultiIndex Series永远做不到的。

2.2 七种核心模式的选型逻辑：什么场景用什么武器

我把生产环境里高频使用的聚合模式，按业务目标归为七类。选择依据不是“哪个函数更炫”，而是“哪个最不容易出错、最容易维护、最贴近业务直觉”：

模式类型	典型业务场景	为什么选它而不是其他方案	我的实操经验
多列多函数聚合	财务报表需同时展示收入均值、中位数、标准差	避免多次groupby+merge带来的索引错乱风险；pandas内置字典映射天然保证分组键一致性	曾用此法将某银行月报生成时间从47分钟压到92秒，关键在避免中间表IO
自定义函数聚合	风控规则中的“交易金额变异系数=标准差/均值”	lambda无法调试、不可复用；命名函数可加docstring说明业务含义，且支持断点调试	所有自定义函数必加 @lru_cache(maxsize=128) ，防止单一客户重复计算耗尽内存
滚动窗口聚合	实时反欺诈中的“近1小时交易频次突增300%”	rolling().apply() 比 resample() 更精准控制窗口边界，且支持非时间索引（如按交易序号滚动）	窗口大小必须业务定义：信用卡用“最近5笔”而非“最近24小时”，因夜间交易少会稀释信号
扩展窗口聚合	客户LTV计算、员工累计业绩追踪	expanding().sum() 比手动cumsum更安全——自动处理分组内排序，且支持 min_periods=1 防首行NaN	必须配合 sort_values() 预处理，否则扩展结果完全错误（亲测！）
多级分组+unstack	销售看板中的“区域×产品矩阵”	unstack() 生成DataFrame比 pivot_table() 更可控，且可链式调用 .fillna(0) 避免空值干扰BI渲染	列名层级必须扁平化： result.columns = ['_'.join(col).strip() for col in result.columns]
复合指标聚合	“高价值客户占比=单笔>5000元交易数/总交易数”	agg() 不支持跨列计算，必须用 apply() + pd.Series 返回多指标	函数内务必用 try/except 包裹，防止单一客户数据异常导致全量失败
分层聚合+重采样	跨年同比分析中的“Q1 vs Q1”	先 groupby(pd.Grouper(freq='Q')) 再 pct_change(periods=4) ，比手动切片更鲁棒	时间频率必须显式声明： freq='QS' （季初）而非 'Q' ，避免季度起始日偏差

这个选型表不是理论推导，而是我在三家金融机构上线12个分析模块后，用故障工单倒逼出来的。比如“滚动窗口”那条经验——某次大促期间，因未设 min_periods=3 ，大量新注册用户因前3天无交易导致滚动均值为NaN，风控策略误判为“零活跃用户”，差点关停支付通道。从此我所有滚动计算必加此参数。

3. 核心细节解析与实操要点：那些文档里不会写的坑

3.1 多列多函数聚合：如何驯服嵌套列名这只怪兽

当你执行 df.groupby(['region','product']).agg({'revenue':['sum','mean'],'cost':['min','max']}) ，pandas会返回一个Column MultiIndex：外层是原始列名（revenue/cost），内层是函数名（sum/mean/min/max）。这个结构在Jupyter里看着清爽，但往生产环境一扔就露馅：

• BI工具对接失败 ：Tableau/Tableau Prep读取时把 ('revenue','sum') 当字符串，列名变成 ("revenue", "sum") ，根本没法拖拽；
• Excel导出错乱 ： to_excel() 默认保留层级，打开后列名显示为两行，业务方直接懵圈；
• 后续计算崩溃 ：想算毛利率 revenue_sum / cost_max ，但 result[('revenue','sum')] 这种写法既难读又易错。

我的标准化解法（已封装为团队内部函数）：

def flatten_agg_columns(df, sep='_'):
    """
    将agg产生的MultiIndex列名扁平化
    示例: ('revenue', 'sum') -> 'revenue_sum'
    """
    if not isinstance(df.columns, pd.MultiIndex):
        return df
    
    # 关键：用列表推导式替代map，避免空格/特殊字符问题
    new_columns = []
    for col in df.columns:
        # 过滤None值（如agg中某列未指定函数）
        clean_parts = [str(c) for c in col if c is not None]
        new_columns.append(sep.join(clean_parts))
    
    df.columns = new_columns
    return df

# 使用示例
result = df.groupby(['region','product']).agg({
    'revenue': ['sum', 'mean'],
    'cost': ['min', 'max']
})
result = flatten_agg_columns(result)  # 输出列名：revenue_sum, revenue_mean, cost_min, cost_max

提示：千万别用 df.columns.map('_'.join) ！当某列只有一层索引（如 agg({'revenue':'sum'}) ）时， '_'.join('revenue') 会返回 'r_e_v_e_n_u_e' ——这是我在某城商行踩过的坑，debug了3小时才发现是字符串被迭代了。

更狠的实战技巧：动态列名注入业务语义
财务部要求“所有金额列单位为万元，保留两位小数”，我直接在扁平化时注入：

def smart_flatten(df, unit='万元', decimal=2):
    new_cols = []
    for col in df.columns:
        base_name = '_'.join([str(c) for c in col if c is not None])
        # 对含'revenue'/'cost'的列自动加单位
        if any(kw in base_name.lower() for kw in ['revenue', 'cost', 'amount']):
            new_cols.append(f"{base_name}_{unit}")
        else:
            new_cols.append(base_name)
    df.columns = new_cols
    
    # 自动数值格式化（仅对金额列）
    amount_cols = [c for c in df.columns if '万元' in c]
    for col in amount_cols:
        df[col] = (df[col] / 10000).round(decimal)
    return df

这样输出的列名直接是 revenue_sum_万元 ，业务方一眼懂，且数据已按需缩放——省去他们自己除10000的步骤，减少人为错误。

3.2 自定义函数聚合：如何让业务逻辑“活”在代码里

很多教程教你写 lambda x: x.max()-x.min() ，但生产环境里，lambda是定时炸弹：

• 无法调试 ：PyCharm断点打不进去，出错时只显示 <lambda> ，你得猜是哪一行；
• 不可审计 ：合规检查时，风控官问“这个变异系数阈值3.5怎么来的？”，你总不能说“我lambda里写的”；
• 难复用 ：同样计算“交易集中度”，A团队用基尼系数，B团队用赫芬达尔指数，C团队用Top3占比——全写lambda，代码库就成垃圾场。

我的解决方案：业务函数工厂模式

from functools import partial
import numpy as np

class BusinessAgg:
    """承载所有业务聚合逻辑的类，强制文档化"""
    
    @staticmethod
    def transaction_range(series, threshold_pct=0.05):
        """
        计算交易金额范围，但排除异常值（默认剔除上下5%）
        业务依据：银保监《反洗钱交易监测指引》第12条
        """
        if len(series) < 3:
            return series.max() - series.min()
        
        # 业务要求：剔除极端值后再计算
        lower = series.quantile(threshold_pct)
        upper = series.quantile(1 - threshold_pct)
        filtered = series[(series >= lower) & (series <= upper)]
        return filtered.max() - filtered.min()
    
    @staticmethod
    def weighted_avg_by_time(series, time_weights=None):
        """
        按时间衰减加权平均（越近的交易权重越高）
        业务依据：客户行为学中"近因效应"，权重衰减系数α=0.92
        """
        if time_weights is None:
            # 自动生成时间权重：最近一笔权重1.0，向前每笔*0.92
            n = len(series)
            weights = np.array([0.92**(n-i) for i in range(n)])
        else:
            weights = np.array(time_weights)
        return np.average(series, weights=weights)

# 注册为pandas agg函数
df.groupby('category')['amount'].agg(
    range_no_outliers=partial(BusinessAgg.transaction_range, threshold_pct=0.03),
    weighted_avg=BusinessAgg.weighted_avg_by_time
)

注意： partial 比lambda安全得多——它保留了函数名和docstring，IDE能跳转，日志能打印，审计时直接截图函数文档就行。

实操心得：所有自定义函数必须带“熔断机制”
某次线上事故：某客户单日交易127笔， weighted_avg_by_time 计算时 np.average 因权重和不为1报错，导致整批数据中断。现在我的所有函数开头必加：

def safe_weighted_avg(series, **kwargs):
    try:
        # 原有逻辑...
        return np.average(series, weights=weights)
    except Exception as e:
        # 业务兜底：降级为简单均值，并记录告警
        logger.warning(f"Weighted avg failed for {len(series)} items: {e}, fallback to mean")
        return series.mean()

这就是生产代码和教学代码的本质区别：前者必须思考“当一切出错时，系统还能不能走”。

3.3 滚动窗口聚合：时间陷阱比想象中更深

rolling(window=7) 看着简单，但实际部署时，90%的问题出在时间对齐上。举个真实案例：某基金公司要做“近5个交易日收益率滚动标准差”，但他们的交易日历不是自然日——节假日、周末、交易所休市日都要剔除。如果直接用 df.rolling(5) ，会把休市日当成0收益，标准差直接失真。

我的四步校准法：

1. 确认数据频率 ：先用 df.index.inferred_freq 检查是否为 'D' （日频），若返回 None ，说明时间戳不规律，必须重采样；
2. 业务日历对齐 ：用 pandas_market_calendars 库获取真实交易日：

   import pandas_market_calendars as mcal
   nyse = mcal.get_calendar('NYSE')
   # 获取2024年所有交易日
   schedule = nyse.schedule(start_date='2024-01-01', end_date='2024-12-31')
   trading_days = schedule.index.date
   

3. 重采样填充 ：将原始数据按交易日重采样，缺失日用前向填充（业务逻辑：休市日无新数据，沿用上一日状态）：

   df_daily = df.set_index('date').reindex(trading_days, method='ffill')
   

4. 滚动计算 ：此时 df_daily.rolling(5) 才真正代表“最近5个交易日”。

更隐蔽的坑：分组内滚动窗口的排序依赖
你以为 df.groupby('customer_id')['amount'].rolling(7).mean() 会自动按时间排序？错！它只按原始DataFrame顺序滚动。如果数据是按客户ID排序的（常见ETL输出），那滚动窗口会跨客户计算！正确姿势：

# 必须显式排序！且排序字段必须是时间列
df_sorted = df.sort_values(['customer_id', 'transaction_time'])
result = df_sorted.groupby('customer_id')['amount'].rolling(7).mean()
# 但注意：rolling()返回的是MultiIndex Series，需reset_index
result = result.reset_index(level=0, drop=True)  # 保留原索引

我在某互联网银行做实时风控时，就因漏了这一步，导致A客户的第7笔交易，被算进了B客户的滚动均值——整整3小时的误报，技术复盘报告写了17页。

4. 实操过程与核心环节实现：一个银行级客户分析流水线

4.1 数据准备：模拟真实信用卡交易流

我们不用玩具数据，直接复刻某股份制银行信用卡中心的典型数据结构。关键特征：

• 时间非均匀 ：交易集中在工作日10-12点、19-21点，周末下午有小高峰；
• 客户分层明显 ：VIP客户单笔均值¥3200，普通客户¥280，但VIP客户数仅占0.7%；
• 商户类别强相关 ：Travel类交易金额方差是Groceries的4.2倍（业务事实：机票价格波动远大于买菜）；
• 数据质量挑战 ：约0.3%交易时间戳为空，0.1%金额为负（退款），需预处理。

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

# 设置随机种子确保可复现
np.random.seed(42)

# 构建真实感时间序列：模拟工作日高峰
def generate_transaction_times(n_days=60, base_date='2024-01-01'):
    dates = pd.date_range(base_date, periods=n_days, freq='D')
    times = []
    for date in dates:
        # 工作日（周一至周五）添加高峰时段
        if date.weekday() < 5:
            # 上午高峰：10-12点，生成30%交易
            morning = pd.date_range(date + pd.Timedelta(hours=10), 
                                  date + pd.Timedelta(hours=12), 
                                  freq='5T')
            # 晚上高峰：19-21点，生成40%交易
            evening = pd.date_range(date + pd.Timedelta(hours=19), 
                                  date + pd.Timedelta(hours=21), 
                                  freq='3T')
            # 随机采样
            all_slots = list(morning) + list(evening)
            sampled = np.random.choice(all_slots, size=int(n_days*0.7), replace=True)
        else:
            # 周末：下午14-16点小高峰
            weekend = pd.date_range(date + pd.Timedelta(hours=14), 
                                  date + pd.Timedelta(hours=16), 
                                  freq='10T')
            sampled = np.random.choice(weekend, size=int(n_days*0.3), replace=True)
        times.extend(sampled)
    return np.random.choice(times, size=n_days*20, replace=True)  # 总计约1200笔

# 生成数据
n_records = 1200
transaction_times = generate_transaction_times()

# 客户分层：VIP(0.7%), Gold(12%), Standard(87.3%)
customer_types = np.random.choice(
    ['VIP', 'Gold', 'Standard'], 
    size=n_records, 
    p=[0.007, 0.12, 0.873]
)

# 按客户类型生成金额（体现业务差异）
amounts = []
for ctype in customer_types:
    if ctype == 'VIP':
        amounts.append(np.random.lognormal(mean=8.2, sigma=0.6))  # 均值≈3200
    elif ctype == 'Gold':
        amounts.append(np.random.lognormal(mean=5.8, sigma=0.5))  # 均值≈320
    else:
        amounts.append(np.random.lognormal(mean=5.2, sigma=0.4))  # 均值≈180

# 商户类别分布（业务真实比例）
categories = np.random.choice(
    ['Groceries', 'Dining', 'Travel', 'Retail', 'Utilities'], 
    size=n_records, 
    p=[0.25, 0.22, 0.18, 0.20, 0.15]
)

# 构建DataFrame
df = pd.DataFrame({
    'transaction_id': [f'TX{str(i).zfill(6)}' for i in range(n_records)],
    'customer_id': np.random.choice([f'C{str(i).zfill(3)}' for i in range(1, 501)], n_records),
    'customer_type': customer_types,
    'category': categories,
    'amount': np.round(amounts, 2),
    'fee': np.round(np.array(amounts) * 0.025, 2),  # 固定费率2.5%
    'transaction_time': transaction_times
})

# 添加数据质量问题：0.3%时间戳为空，0.1%金额为负（退款）
null_time_idx = np.random.choice(n_records, size=int(n_records*0.003), replace=False)
df.loc[null_time_idx, 'transaction_time'] = pd.NaT

refund_idx = np.random.choice(n_records, size=int(n_records*0.001), replace=False)
df.loc[refund_idx, 'amount'] = -df.loc[refund_idx, 'amount']

print("信用卡交易数据概览：")
print(f"总记录数：{len(df)}")
print(f"客户数：{df['customer_id'].nunique()}")
print(f"时间范围：{df['transaction_time'].min()} 至 {df['transaction_time'].max()}")
print(f"数据质量：空时间戳{df['transaction_time'].isna().sum()}条，退款{len(refund_idx)}笔")

这段代码生成的不是均匀分布的随机数，而是带着业务指纹的数据——这才是你每天真实面对的战场。

4.2 分析流水线：七步构建银行级洞察

步骤1：多维聚合——客户分层×商户类别的全景视图

# 关键：先处理数据质量，再聚合
df_clean = df.dropna(subset=['transaction_time'])  # 删除空时间戳
df_clean = df_clean[df_clean['amount'] > 0]  # 排除退款（或单独标记）

# 七维聚合：客户类型、商户类别、时间粒度（周）、金额统计
# 业务要求：VIP客户看绝对值，普通客户看占比
weekly_stats = (
    df_clean
    .assign(week=lambda x: x['transaction_time'].dt.to_period('W'))
    .groupby(['customer_type', 'category', 'week'])
    .agg({
        'amount': ['sum', 'mean', 'count', 'std'],
        'fee': 'sum'
    })
    .round(2)
)

# 扁平化列名
weekly_stats = flatten_agg_columns(weekly_stats)

# 计算各客户类型在各类别中的交易占比（业务核心指标）
category_share = (
    df_clean
    .groupby(['customer_type', 'category'])
    .size()
    .groupby(level=0)  # 按customer_type分组
    .apply(lambda x: x / x.sum() * 100)  # 计算百分比
    .round(1)
    .rename('category_share_pct')
    .reset_index()
)

print("【步骤1输出】客户分层×商户类别周度统计（示例）：")
print(weekly_stats.head(10))
print("\n【步骤1输出】各客户类型商户偏好占比：")
print(category_share.head(10))

实操心得： 永远先做数据清洗再聚合 。某次我跳过这步，直接对含退款的数据 sum() ，结果VIP客户“净收入”为负，风控部连夜开会——其实只是退款没过滤。现在我的所有聚合前必加 df_clean = df.query('amount > 0') 。

步骤2：自定义聚合——构建“交易健康度”指标

银行业务痛点：单纯看交易额会忽略风险。我们需要一个综合指标，同时反映 规模、稳定性、增长性 。

def transaction_health_score(series):
    """
    交易健康度评分（0-100分）
    业务逻辑：规模（40%）+ 稳定性（30%，用变异系数）+ 增长性（30%，用近7天vs前7天增速）
    变异系数 = std/mean，值越小越稳定；增速为正向加分项
    """
    if len(series) < 14:  # 至少需要14天数据
        return np.nan
    
    # 规模得分：log10(均值)，映射到0-40分（避免大客户碾压小客户）
    mean_val = series.mean()
    scale_score = min(40, max(0, np.log10(mean_val + 1) * 10))
    
    # 稳定性得分：变异系数越小越好，1/（1+变异系数）映射到0-30分
    std_val = series.std()
    cv = std_val / mean_val if mean_val > 0 else np.inf
    stability_score = 30 / (1 + cv) if cv < np.inf else 0
    
    # 增长性得分：近7天均值 / 前7天均值，限制在0-30分
    recent = series.iloc[-7:].mean()
    prior = series.iloc[-14:-7].mean()
    growth_rate = recent / prior if prior > 0 else 0
    growth_score = min(30, max(0, (growth_rate - 1) * 100))
    
    return round(scale_score + stability_score + growth_score, 1)

# 应用到客户维度
health_scores = (
    df_clean
    .sort_values(['customer_id', 'transaction_time'])  # 必须排序！
    .groupby('customer_id')['amount']
    .apply(transaction_health_score)
    .rename('health_score')
    .reset_index()
)

print("【步骤2输出】客户交易健康度评分（Top10）：")
print(health_scores.nlargest(10, 'health_score'))

这个函数不是数学游戏，而是把风控经理口头说的“我们要找那些交易稳定、持续增长、但不过度集中的客户”翻译成了可计算的公式。 所有业务指标必须能被一句话解释清楚，否则就是伪需求。

步骤3：滚动窗口——实时捕捉行为突变

# 按客户ID分组，计算近7天滚动交易频次（检测刷单）
df_sorted = df_clean.sort_values(['customer_id', 'transaction_time'])
df_sorted['rolling_count_7d'] = (
    df_sorted
    .groupby('customer_id')['transaction_id']
    .rolling('7D', on='transaction_time')  # 关键：on参数指定时间列
    .count()
    .reset_index(level=0, drop=True)
)

# 计算突变比率：当前滚动频次 / 近30天均值滚动频次
# 先计算30天窗口的滚动均值
df_sorted['rolling_mean_30d'] = (
    df_sorted
    .groupby('customer_id')['rolling_count_7d']
    .rolling('30D', on='transaction_time')
    .mean()
    .reset_index(level=0, drop=True)
)

# 突变比率 = 当前滚动频次 / 30天均值，>2.5即预警
df_sorted['abnormal_ratio'] = (
    df_sorted['rolling_count_7d'] / df_sorted['rolling_mean_30d']
)

# 标记高风险客户（突变比率>2.5且频次>10）
abnormal_customers = (
    df_sorted
    .query('abnormal_ratio > 2.5 and rolling_count_7d > 10')
    .groupby('customer_id')
    .agg({
        'abnormal_ratio': 'max',
        'rolling_count_7d': 'last'
    })
    .sort_values('abnormal_ratio', ascending=False)
    .head(10)
)

print("【步骤3输出】高风险行为突变客户（Top10）：")
print(abnormal_customers)

注意： rolling('7D', on='transaction_time') 中的 on 参数是灵魂。不用它，pandas会按行号滚动（即最近7行），完全失去时间意义。这个参数在pandas 1.3+才稳定支持，旧版本必须先 set_index('transaction_time') 。

步骤4：扩展窗口——构建客户生命周期价值（LTV）

# 按客户ID和时间排序，计算累计消费
df_ltv = df_clean.sort_values(['customer_id', 'transaction_time'])
df_ltv['cumulative_spend'] = (
    df_ltv
    .groupby('customer_id')['amount']
    .expanding(min_periods=1)  # 至少1笔才计算
    .sum()
    .reset_index(level=0, drop=True)
)

# 计算LTV分位数（业务用于客户分层）
ltv_percentiles = (
    df_ltv
    .groupby('customer_id')['cumulative_spend']
    .last()  # 取最终累计值
    .quantile([0.25, 0.5, 0.75, 0.9])
    .round(0)
)

print("【步骤4输出】客户LTV分位数（元）：")
print(ltv_percentiles)

这里 min_periods=1 是关键。没有它，第一个交易的累计值就是NaN，整个链条断裂。 生产代码里，所有可能产生NaN的地方，都必须有明确的缺省策略。

步骤5：多级分组+unstack——生成管理层看板

# 业务需求：各区域（模拟）各产品线的平均交易额矩阵
# 先按时间分层：Q1/Q2/Q3/Q4
df_qtr = df_clean.assign(quarter=lambda x: x['transaction_time'].dt.to_period('Q'))

# 多级分组：区域（按客户ID前缀模拟）、产品线（category）、季度
# 区域映射：C001-C200→North, C201-C400→South, C401-C500→West
df_qtr['region'] = df_qtr['customer_id'].str[1:4].astype(int)
df_qtr['region'] = pd.cut(
    df_qtr['region'], 
    bins=[0, 200, 400, 500], 
    labels=['North', 'South', 'West'],
    include_lowest=True
)

# 生成交叉表
crosstab = (
    df_qtr
    .groupby(['region', 'category', 'quarter'])['amount']
    .mean()
    .unstack(['category', 'quarter'])  # 双层unstack
    .round(2)
)

# 扁平化双层列名
crosstab.columns = [
    f"{cat}_{qtr}" for cat, qtr in crosstab.columns
]

print("【步骤5输出】区域×产品线×季度平均交易额矩阵：")
print(crosstab.head())

unstack(['category', 'quarter']) 生成的是双层列，比单层更符合业务思维——“Dining_Q2024”比“Dining”更能说明问题。

步骤6：复合指标——高净值客户识别引擎

def high_value_segmentation(group):
    """
    高净值客户识别：同时满足三个条件
    1. 近30天交易总额 > ¥50,000
    2. 单笔>¥5,000交易占比 > 15%
    3. 交易商户类别数 ≥ 5（体现消费广度）
    """
    # 近30天数据
    recent_30 = group[group['transaction_time'] >= group['transaction_time'].max() - pd.Timedelta(days=30)]
    
    total_spend = recent_30['amount'].sum()
    high_value_ratio = (recent_30['amount'] > 5000).mean()
    merchant_diversity = recent_30['category'].nunique()
    
    return pd.Series({
        'total_spend_30d': round(total_spend, 2),
        'high_value_ratio': round(high_value_ratio * 100, 1),
        'merchant_diversity': merchant_diversity,
        'is_high_value': (
            total_spend > 50000 and 
            high_value_ratio > 0.15 and 
            merchant_diversity >= 5
        )
    })

hv_segments = (
    df_clean
    .groupby('customer_id')
    .apply(high_value_segmentation)
    .query('is_high_value')
    .sort_values('total_spend_30d', ascending=False)
)

print("【步骤6输出】高净值客户清单（Top5）：")
print(hv_segments.head())

这个函数体现了 业务规则的可组合性 。每个条件都是独立可验证的，组合起来就是风控策略。上线后，市场部立刻用这份名单做了高端信用卡定向营销，ROI提升22%。

步骤7：端到端整合——生成高管摘要报告

# 整合所有分析结果，生成一页纸摘要
summary_report = {}

# 1. 整体健康度
summary_report['overall_health'] = round(health_scores['health_score'].mean(), 1)

# 2. 风险客户数
summary_report['abnormal_customers'] = len(abnormal_customers)

# 3. 高净值客户数
summary_report['high_value_customers'] = len(hv_segments)

# 4. LTV分布
ltv_last = df_ltv.groupby('customer_id')['cumulative_s