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

我在银行数据团队干了八年，从最早用Excel手搓报表，到后来写SQL跑T+1任务，再到如今用Pandas搭实时分析流水线——最常被低估、也最容易在关键时刻掉链子的，就是聚合操作。很多人觉得 groupby().sum() 和 groupby().mean() 就是全部了，直到某天风控同事深夜打电话问：“上个月南区高端客户在旅游类商户的单日交易波动率突然飙升300%，系统没告警，是不是聚合逻辑漏了什么？”——查了一宿代码，发现根本不是算法问题，而是原始聚合只按“客户ID”分组，把“时间窗口”“地域层级”“商户类型”三个维度全压扁了，波动率计算失去了时间锚点，自然失效。

这就是Part 20要解决的核心问题： 真实业务场景中的聚合，从来不是单维度的静态切片，而是多维度、有时序、带业务逻辑的动态建模过程 。你看到的“平均交易额”，背后可能是“近30天滚动均值 vs 历史基线标准差”的比值；你导出的“区域销售额”，实际是“按省-市-区三级行政编码展开后，再按产品线交叉透视”的矩阵；你提交给高管的“高风险客户清单”，底层逻辑是“单笔超5万元交易占比 + 近7天交易频次增速 + 同类客群偏离度”的复合打分。这些都不是 agg('mean') 能覆盖的，它们需要你像搭乐高一样，把基础聚合块（sum/mean）、自定义逻辑块（range/weighted_avg）、时间窗口块（rolling/expanding）、结构重塑块（unstack/melt）严丝合缝地拼接起来。

关键词里反复出现的“Towards AI”，其实暗示了这类内容的受众本质：不是教初学者怎么安装pandas，而是帮已在生产环境跑着千万级数据的分析师、数据工程师、风控建模师，把那些藏在日报背后的“隐性假设”显性化、可复现、可审计。比如财务部要的“各产品线毛利率”，表面看是 sum(revenue)/sum(cost) ，但实际必须区分：是按订单粒度聚合再计算？还是先算每单毛利率再取均值？前者受大额订单扰动小，后者更能反映单笔交易健康度——这个选择直接决定是否误判某个产品线正在亏损。我见过太多团队因为没厘清这点，在季度复盘会上被业务方当场质疑数据口径，最后发现根源就在一行 agg({'revenue': 'sum', 'cost': 'sum'}) 的写法上。

所以这篇内容的价值，不在于教你新函数，而在于帮你建立一套 聚合决策框架 ：当业务需求抛来一句“我要看XX维度下的YY指标”，你能立刻拆解出四个关键判断点——
第一，维度是否多层嵌套？（比如“华东大区→江苏省→南京市→玄武区”）
第二，指标是否需时序对比？（比如“本周 vs 上周 vs 去年同期”）
第三，计算逻辑是否含业务规则？（比如“手续费按阶梯费率计算，非简单百分比”）
第四，结果是否需特定呈现形态？（比如“供BI拖拽的宽表” or “供算法模型输入的长表”）
这四个问题的答案，直接决定你该用 agg() 字典映射、 rolling().apply() 、 expanding().agg() 还是 groupby().unstack() 。接下来我会用银行真实场景的七层分析链，把每个决策点的坑和解法，掰开揉碎讲透。

2. 核心设计思路：从“能跑通”到“能扛住生产压力”的四重跃迁

很多同学学完pandas聚合，写个demo脚本跑通就以为掌握了。但当我接手第一个银行核心报表系统时，运维同事指着监控图说：“你这个聚合脚本，每天凌晨跑3小时，CPU峰值98%，还把下游ETL卡死——它不是‘能用’，是‘有毒’。” 这句话让我彻底反思： 生产环境的聚合，本质是工程问题，不是语法问题 。我把多年踩坑经验总结为四重跃迁，这也是本文所有案例的设计底层逻辑。

2.1 第一重跃迁：从“单列聚合”到“跨列协同聚合”

新手常犯的错误，是把不同业务含义的指标硬塞进同一个 agg() 调用。比如财务要求同时输出“交易金额均值”和“手续费最小值”，直觉写法是：

df.groupby('category').agg({'amount': 'mean', 'fee': 'min'})

这看似正确，但埋下两个隐患：
隐患一：语义断裂 。 amount.mean() 反映客户支付能力， fee.min() 暴露渠道成本下限，二者无业务关联性。当某类商户 fee.min() 异常低时，你无法判断是渠道让利还是数据污染——因为聚合结果把两个独立信号强行耦合在同一个DataFrame里。
隐患二：性能陷阱 。pandas对字典式聚合会为每个键值对单独扫描数据， {'amount': 'mean', 'fee': 'min'} 实际触发两次全量遍历。当数据量超千万行，耗时翻倍。

我的解决方案是 按业务域分组聚合 ：

# 按客户价值域聚合（金额相关）
value_agg = df.groupby('category')['amount'].agg(['mean', 'median', 'std'])

# 按成本控制域聚合（手续费相关）  
cost_agg = df.groupby('category')['fee'].agg(['min', 'max', 'mean'])

# 最终用concat横向拼接，保留语义隔离
result = pd.concat([value_agg, cost_agg], axis=1)

这样做的好处：

• 可解释性强 ： value_agg 和 cost_agg 变量名自带业务上下文，半年后回看代码无需查文档；
• 调试友好 ：若 cost_agg 结果异常，可单独重跑该段，不影响 value_agg ；
• 性能提升 ：单次扫描完成同域内所有计算，实测千万行数据提速40%。

提示：银行内部审计要求所有指标计算过程可追溯。我们强制规定——任何聚合结果必须能反向定位到其所属的业务域变量（如 value_agg , risk_agg ），禁止出现 temp_result = df.groupby(...).agg({...}) 这类无意义命名。

2.2 第二重跃迁：从“内置函数”到“业务逻辑即代码”

lambda x: x.max() - x.min() 这种写法很酷，但上线前必须回答三个问题：

1. 当 x 为空时返回什么？（空Series调用 max() 抛 ValueError ）
2. 当 x 含缺失值时如何处理？（ max() 默认跳过NaN，但业务可能要求“有任一NaN则整个指标置空”）
3. 这个范围值是否需单位校验？（比如交易金额范围超100万，应触发人工复核而非直接入库）

我坚持用 命名函数替代lambda ，且函数体必须包含防御性编程：

def transaction_range(series):
    """计算交易金额区间，业务规则：空序列返回None，含NaN则整体置空"""
    if series.empty:
        return None
    if series.isna().any():
        return None
    return series.max() - series.min()

# 调用时明确传递参数，避免歧义
result = df.groupby('category').agg({'amount': transaction_range})

更进一步，我们封装了 银行专用聚合器类 ，把常见业务逻辑预置为方法：

class BankingAgg:
    @staticmethod
    def fee_ratio(amount_series, fee_series):
        """手续费率 = 总手续费 / 总交易额，要求双序列长度一致且无空值"""
        if len(amount_series) != len(fee_series):
            raise ValueError("金额与手续费序列长度不匹配")
        if amount_series.sum() == 0:
            return 0.0
        return (fee_series.sum() / amount_series.sum()) * 100
    
    @staticmethod
    def high_value_ratio(series, threshold=300):
        """高价值交易占比，threshold可配置化"""
        return (series > threshold).sum() / len(series) * 100

# 使用时清晰表明业务意图
result = df.groupby('customer_id').agg({
    'amount': ['sum', 'mean'],
    ('amount', 'fee'): lambda x: BankingAgg.fee_ratio(x['amount'], x['fee'])
})

这种设计让代码成为活的业务文档——当新同事看到 BankingAgg.fee_ratio ，立刻明白这是“总手续费率”，而非某个模糊的“比率计算”。

2.3 第三重跃迁：从“静态切片”到“动态时间窗口”

rolling(window=7).mean() 看着简单，但生产环境必须处理三类边界情况：

• 数据稀疏性 ：某客户连续5天无交易，第6天突然有3笔，此时7日窗口只有3个有效点；
• 业务时效性 ：风控要求“最近7个自然日”，而非“最近7条记录”，需确保日期字段已补全；
• 计算一致性 ：下游系统要求所有指标统一用“前推填充”（ffill），但pandas默认返回NaN。

我们的标准做法是 预处理+参数化 ：

def safe_rolling_mean(series, window=7, min_periods=3, fill_method='ffill'):
    """
    银行级滚动均值：支持最小有效期、填充策略、自动对齐日期
    """
    # 步骤1：确保索引为DatetimeIndex且无重复
    if not isinstance(series.index, pd.DatetimeIndex):
        raise ValueError("序列索引必须为日期类型")
    
    # 步骤2：按自然日补全缺失日期（用0填充交易额，用NaN填充其他指标）
    full_index = pd.date_range(series.index.min(), series.index.max(), freq='D')
    series_full = series.reindex(full_index, fill_value=0)
    
    # 步骤3：执行滚动计算，指定最小有效期
    rolling_result = series_full.rolling(
        window=window, 
        min_periods=min_periods
    ).mean()
    
    # 步骤4：按需填充
    if fill_method == 'ffill':
        rolling_result = rolling_result.fillna(method='ffill')
    elif fill_method == 'drop':
        rolling_result = rolling_result.dropna()
    
    return rolling_result

# 实际调用（业务含义一目了然）
df['7day_spend_avg'] = safe_rolling_mean(df['amount'], window=7, min_periods=3)

这套逻辑已沉淀为团队共享包 banking_pandas ，所有成员调用同一接口，杜绝“张三用 min_periods=1 ，李四用 min_periods=5 ”导致的指标不一致。

2.4 第四重跃迁：从“结果可用”到“结构即服务”

unstack() 常被当作格式美化工具，但在银行系统里，它是 数据契约的物理载体 。例如，监管报送要求“按机构-产品-币种三维交叉表”格式，任何维度顺序错位都会被退回。我们制定铁律：

• unstack前必验证索引层级 ： groupby(['org', 'product', 'currency']) 生成的MultiIndex，必须用 verify_integrity=True 检查是否有重复组合；
• unstack后必强类型转换 ： result = result.unstack(['product', 'currency']).astype('float32') ，避免因object类型导致BI工具解析失败；
• 缺失值必业务赋值 ： unstack(fill_value=0) 中的 0 不是随意选的，而是监管明文规定的“未发生交易”占位符。

更关键的是，我们把 unstack 操作封装进 数据契约校验器 ：

def validate_cross_tab(df, index_cols, column_cols, expected_shape):
    """
    校验交叉表是否符合监管/业务契约
    expected_shape: 如 (32, 15) 表示32家机构 × 15个产品
    """
    # 检查索引唯一性
    if df.index.duplicated().any():
        raise ValueError(f"索引存在重复组合：{df.index[df.index.duplicated()]}")
    
    # 检查维度完整性
    actual_shape = (len(df.index.unique(level=0)), len(df.columns))
    if actual_shape != expected_shape:
        raise ValueError(f"维度不匹配：期望{expected_shape}，实际{actual_shape}")
    
    # 检查数值类型
    if not np.issubdtype(df.values.dtype, np.number):
        raise TypeError(f"交叉表必须为数值类型，当前为{df.values.dtype}")
    
    return True

# 使用示例
cross_tab = df.groupby(['org', 'product'])['amount'].sum().unstack(fill_value=0)
validate_cross_tab(cross_tab, ['org'], ['product'], (32, 15))

这套机制让 unstack 从“锦上添花”变成“合规刚需”，每次报表生成都自动通过契约校验，而不是等监管反馈才返工。

3. 实操细节拆解：银行七层分析链的逐层实现

现在我们进入最硬核的部分——用一个贯穿始终的银行信用卡分析场景，把前述设计思想落地为可直接抄作业的代码。注意，这里所有数据生成、参数设定、异常处理，都来自我经手的真实项目（已脱敏）。别跳过任何细节，那些看似“多此一举”的写法，都是血泪教训换来的。

3.1 数据生成：模拟真实业务约束的合成数据

很多教程用 np.random 生成数据，但银行数据有强业务约束：

• 交易金额服从对数正态分布（小额高频，大额低频）；
• 手续费是阶梯费率（如≤100元收1.5%，＞100元收2.0%）；
• 客户活跃度有周期性（工作日交易量比周末高35%）；
• 区域编码需符合国家标准（GB/T 2260）。

我写的合成器严格遵循这些规则：

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

def generate_bank_transactions(n_samples=10000, seed=42):
    """
    生成符合银行业务特征的信用卡交易数据
    """
    np.random.seed(seed)
    
    # 1. 客户池（模拟真实分布：80%普通客户，15%金卡，5%白金）
    customers = ['C' + str(i).zfill(3) for i in range(1, 1001)]
    customer_tiers = np.random.choice(
        ['Standard', 'Gold', 'Platinum'], 
        size=len(customers), 
        p=[0.8, 0.15, 0.05]
    )
    customer_df = pd.DataFrame({'customer_id': customers, 'tier': customer_tiers})
    
    # 2. 商户类别（按银联分类标准）
    categories = ['Groceries', 'Dining', 'Travel', 'Retail', 'Healthcare', 'Education']
    
    # 3. 交易日期（工作日权重更高）
    start_date = datetime(2024, 1, 1)
    end_date = datetime(2024, 3, 31)
    date_range = pd.date_range(start_date, end_date, freq='D')
    # 工作日权重1.35，周末权重0.65
    weekday_weights = [1.35 if d.weekday() < 5 else 0.65 for d in date_range]
    dates = np.random.choice(date_range, size=n_samples, p=weekday_weights/np.sum(weekday_weights))
    
    # 4. 交易金额（对数正态分布，模拟长尾特性）
    # 普通客户：均值150，金卡：均值300，白金：均值800
    tier_means = {'Standard': 150, 'Gold': 300, 'Platinum': 800}
    amounts = []
    for cid in np.random.choice(customers, size=n_samples):
        tier = customer_df[customer_df['customer_id']==cid]['tier'].iloc[0]
        # 对数正态参数：mu=ln(mean)-sigma²/2, sigma根据tier调整
        sigma = 0.8 if tier == 'Standard' else 1.0 if tier == 'Gold' else 1.2
        mu = np.log(tier_means[tier]) - sigma**2 / 2
        amount = np.random.lognormal(mu, sigma)
        # 截断至合理范围（0-50000）
        amount = max(1, min(50000, amount))
        amounts.append(round(amount, 2))
    
    # 5. 手续费（阶梯费率）
    fees = []
    for amt in amounts:
        if amt <= 100:
            fee = round(amt * 0.015, 2)
        elif amt <= 1000:
            fee = round(100*0.015 + (amt-100)*0.02, 2)
        else:
            fee = round(100*0.015 + 900*0.02 + (amt-1000)*0.025, 2)
        fees.append(fee)
    
    # 6. 组装DataFrame
    data = {
        'date': dates,
        'customer_id': np.random.choice(customers, size=n_samples),
        'category': np.random.choice(categories, size=n_samples),
        'amount': amounts,
        'fee': fees
    }
    return pd.DataFrame(data)

# 生成10万行数据（模拟单月交易量）
df = generate_bank_transactions(100000)
print(f"生成数据形状：{df.shape}")
print(f"日期范围：{df['date'].min()} 至 {df['date'].max()}")
print(f"客户分层：{df.merge(pd.DataFrame({'customer_id': customers, 'tier': customer_tiers}), on='customer_id')['tier'].value_counts()}")

这段代码的关键价值在于： 它生成的数据天然携带业务约束 。当你后续做 rolling() 计算时，不必担心“为什么周末交易量突降”——因为合成器已按真实规律注入；当你做 unstack() 时，也不用手动补全缺失机构——因为客户池按监管要求的32家分行生成。这才是生产级数据准备该有的样子。

3.2 多维聚合实战：七层分析链的代码实现

现在，我们用这份真实感十足的数据，执行银行每日必做的七层分析。每层都标注了 业务目标、技术要点、避坑提示 ，你可以直接复制到Jupyter中运行。

分析层1：客户-品类双维度统计（解决“谁在买什么”）

# 业务目标：识别高价值客户在各品类的消费特征
# 技术要点：多列不同聚合函数 + 层级列名扁平化
print("=== 分析层1：客户-品类双维度统计 ===")

# 执行聚合（按客户ID和品类分组）
multi_agg = df.groupby(['customer_id', 'category']).agg({
    'amount': ['sum', 'mean', 'count', 'std'],  # 金额相关指标
    'fee': ['sum', 'mean']                        # 手续费相关指标
})

# 扁平化列名（避免MultiIndex带来的后续处理麻烦）
multi_agg.columns = ['_'.join(col).strip() for col in multi_agg.columns]
multi_agg = multi_agg.reset_index()

# 关键避坑：处理std为NaN的情况（当某客户在某品类仅1笔交易）
multi_agg['amount_std'] = multi_agg['amount_std'].fillna(0)

# 输出前5行（观察数据形态）
print(multi_agg.head())
print(f"聚合后行数：{len(multi_agg)}，原始行数：{len(df)}")
print(f"平均每客户-品类组合交易笔数：{len(df)/len(multi_agg):.1f}")

注意：这里 fillna(0) 不是随意为之。银行业务规则明确——当标准差无法计算时，视为“无波动”，填0比填NaN更符合监管报送逻辑。我在某次审计中就因没填0，被要求重新跑全量数据。

分析层2：自定义风险指标（解决“哪些交易异常”）

# 业务目标：计算每个客户的“高价值交易占比”，用于反欺诈模型特征
# 技术要点：命名函数 + 业务阈值参数化 + 缺失值防御
print("\n=== 分析层2：自定义风险指标 ===")

def high_value_ratio(series, threshold=300, min_transactions=5):
    """
    计算高价值交易占比，含业务兜底逻辑
    threshold: 高价值门槛（可随监管政策调整）
    min_transactions: 最小交易笔数，避免样本过少导致指标失真
    """
    if len(series) < min_transactions:
        return np.nan  # 样本不足，不参与评分
    
    high_count = (series > threshold).sum()
    return round((high_count / len(series)) * 100, 2)

# 应用自定义函数
risk_df = df.groupby('customer_id').agg({
    'amount': lambda x: high_value_ratio(x, threshold=300),
    'fee': 'sum'
}).rename(columns={'amount': 'high_value_pct', 'fee': 'total_fee'})

# 关键避坑：过滤掉无效客户（如测试账户、休眠户）
# 银行规则：近90天无交易或总交易额<100元，视为无效
active_mask = (df.groupby('customer_id')['amount'].sum() >= 100) & \
              (df.groupby('customer_id')['date'].max() >= (datetime.now() - timedelta(days=90)))
risk_df = risk_df[risk_df.index.isin(active_mask[active_mask].index)]

print(risk_df.sort_values('high_value_pct', ascending=False).head(10))
print(f"有效客户数：{len(risk_df)}，总客户数：{df['customer_id'].nunique()}")

分析层3：滚动窗口分析（解决“趋势是否异常”）

# 业务目标：计算每个客户的7日滚动交易均值，识别突发性消费增长
# 技术要点：日期索引对齐 + 最小有效期 + 前推填充
print("\n=== 分析层3：滚动窗口分析 ===")

# 步骤1：按客户和日期排序，确保时间序列连续
df_sorted = df.sort_values(['customer_id', 'date']).set_index('date')

# 步骤2：为每个客户生成完整日期索引（补全无交易日）
def fill_customer_dates(group):
    min_date, max_date = group.index.min(), group.index.max()
    full_dates = pd.date_range(min_date, max_date, freq='D')
    return group.reindex(full_dates, fill_value=0)

# 步骤3：应用滚动计算（使用我们之前定义的safe_rolling_mean）
df_filled = df_sorted.groupby('customer_id')['amount'].apply(fill_customer_dates)
rolling_result = df_filled.rolling(window=7, min_periods=3).mean()

# 步骤4：合并回原数据（关键！避免索引错位）
df_with_rolling = df.set_index(['date', 'customer_id'])
df_with_rolling['7day_avg'] = rolling_result.reindex(df_with_rolling.index, method='ffill')

# 关键避坑：必须用reindex+method='ffill'，而非直接merge
# 因为原始df有重复日期-客户组合（同天多笔交易），merge会爆炸式膨胀
print(df_with_rolling.reset_index().head(10))

分析层4：扩展窗口分析（解决“长期价值评估”）

# 业务目标：计算每个客户的累计交易额，用于LTV（客户终身价值）模型
# 技术要点：按客户分组 + expanding() + 累计值业务校验
print("\n=== 分析层4：扩展窗口分析 ===")

# 按客户分组计算累计和
cumulative = df_sorted.groupby('customer_id')['amount'].expanding().sum()

# 重构为DataFrame（避免MultiIndex混乱）
cum_df = pd.DataFrame({
    'customer_id': cumulative.index.get_level_values(0),
    'date': cumulative.index.get_level_values(1),
    'cumulative_spend': cumulative.values
})

# 关键避坑：累计值必须单调不减（数学上成立，但需程序校验）
# 银行风控要求：若发现累计值下降，立即告警（数据污染信号）
if (cum_df.groupby('customer_id')['cumulative_spend'].diff() < 0).any():
    raise ValueError("检测到累计值下降，可能存在数据异常！")

print(cum_df.sort_values(['customer_id', 'date']).head(10))

分析层5：多级透视表（解决“管理驾驶舱展示”）

# 业务目标：生成供高管查看的“区域-产品”交叉表
# 技术要点：多级groupby + unstack + 契约校验
print("\n=== 分析层5：多级透视表 ===")

# 模拟区域字段（按GB/T 2260标准生成32家分行）
regions = [f"Region_{i}" for i in range(1, 33)]
df['region'] = np.random.choice(regions, size=len(df))

# 执行多级聚合
crosstab = df.groupby(['region', 'category'])['amount'].sum().unstack(fill_value=0)

# 关键避坑：强制类型转换 + 形状校验
crosstab = crosstab.astype('int64')  # 避免float导致BI显示小数
if crosstab.shape != (32, 6):
    raise ValueError(f"交叉表维度错误：期望(32,6)，实际{crosstab.shape}")

print("区域-品类交易额汇总（前5行）：")
print(crosstab.head())
print(f"\n总交易额验证：{crosstab.values.sum():,} vs 原始数据{df['amount'].sum():,}")

分析层6：高管摘要（解决“一句话结论”）

# 业务目标：生成一页纸的高管摘要，含核心KPI
# 技术要点：多指标聚合 + 列名语义化 + 百分比计算
print("\n=== 分析层6：高管摘要 ===")

summary = df.groupby('customer_id').agg({
    'amount': ['sum', 'mean', 'count'],
    'fee': 'sum'
}).round(2)

# 扁平化列名并重命名
summary.columns = ['total_spend', 'avg_transaction', 'transaction_count', 'total_fee']
summary = summary.reset_index()

# 计算关键比率（手续费率、客单价）
summary['fee_rate'] = ((summary['total_fee'] / summary['total_spend']) * 100).round(2)
summary['spend_per_transaction'] = (summary['total_spend'] / summary['transaction_count']).round(2)

# 关键避坑：手续费率分母为0的处理（银行规则：无交易客户手续费率=0）
summary['fee_rate'] = summary['fee_rate'].fillna(0)

# 输出Top10高价值客户
top_customers = summary.nlargest(10, 'total_spend')[[
    'customer_id', 'total_spend', 'transaction_count', 'fee_rate'
]]
print("Top10高价值客户：")
print(top_customers.to_string(index=False))

分析层7：复合风险评分（解决“精准干预”）

# 业务目标：构建客户风险评分卡，综合高价值占比、波动率、增速
# 技术要点：apply()传入多列 + 权重配置化 + 评分标准化
print("\n=== 分析层7：复合风险评分 ===")

def risk_score(row, weights={'high_value': 0.4, 'volatility': 0.3, 'growth': 0.3}):
    """
    综合风险评分（0-100分）
    high_value: 高价值交易占比（越高越风险）
    volatility: 交易金额标准差（越高越风险）  
    growth: 近30天交易额环比增速（越高越风险）
    """
    # 获取该客户数据
    cust_data = df[df['customer_id'] == row.name]
    
    # 计算各维度分项
    high_value_pct = (cust_data['amount'] > 300).sum() / len(cust_data) * 100 if len(cust_data) > 0 else 0
    volatility = cust_data['amount'].std() if len(cust_data) > 1 else 0
    # 计算30天环比（需确保有足够数据）
    if len(cust_data) < 30:
        growth = 0
    else:
        recent = cust_data.nlargest(30, 'date')['amount'].sum()
        prior = cust_data.nsmallest(30, 'date')['amount'].sum()
        growth = ((recent / prior) - 1) * 100 if prior > 0 else 0
    
    # 加权汇总（各维度先归一化到0-100）
    score = (
        min(100, high_value_pct * weights['high_value']) +
        min(100, (volatility / 500) * 100 * weights['volatility']) +  # 假设500为波动率上限
        min(100, growth * weights['growth'])
    )
    return round(score, 1)

# 应用评分函数（注意：传入的是分组后的Series，需用apply）
risk_scores = summary.set_index('customer_id').apply(risk_score, axis=1)

# 关键避坑：评分必须业务可解释
# 我们约定：0-30低风险，31-70中风险，71-100高风险
risk_levels = pd.cut(risk_scores, bins=[0,30,70,100], labels=['Low', 'Medium', 'High'])

risk_report = pd.DataFrame({
    'risk_score': risk_scores,
    'risk_level': risk_levels
}).sort_values('risk_score', ascending=False)

print("高风险客户TOP5：")
print(risk_report.head(5))

4. 生产环境避坑指南：那些文档里不会写的血泪教训

以上代码在Jupyter里跑通只是第一步。真正上生产，还有无数细节决定成败。这些经验，是我和团队在三年间踩过至少27次坑后总结的，每一条都对应一次线上事故。

4.1 内存爆炸：你以为的“小数据”，其实是内存杀手

现象：某次部署后，服务器内存占用从40%飙升至99%，Prometheus告警疯狂推送。排查发现，一个看似简单的 df.groupby(['region','product','category']).agg({'amount':'sum'}) ，在32×15×6=2880个组合下，生成了2880行结果——但问题不在行数，而在 pandas的字符串索引内存开销 。

真相：当 region 字段是 'Region_01' 这类字符串时，pandas为每个唯一值分配独立内存块。2880个字符串，每个占64字节，光索引就吃掉184KB。但更致命的是，当后续做 unstack() 时，pandas会为每个缺失组合创建 NaN 占位符，而 NaN 在object类型中占更多内存。

解决方案：

• 索引数字化 ：将 region 映射为整数ID（ Region_01 → 1 ），用 category dtype存储；
• 预过滤 ： groupby() 前用 df = df[df['amount'] > 0] 剔除无效记录；
• 分块聚合 ：对超大数据集，用 pd.read_csv(chunksize=10000) 分批处理，再 pd.concat() 合并。

# 实战代码：安全的分块聚合
def safe_groupby_aggregate(df_path, chunk_size=50000, **kwargs):
    """
    内存安全的分块聚合
    """
    results = []
    for chunk in pd.read_csv(df_path, chunksize=chunk_size):
        # 对每块数据执行相同聚合
        chunk_result = chunk.groupby(kwargs['by'])[kwargs['agg_col']].agg(kwargs['agg_func'])
        results.append(chunk_result)
    
    # 合并结果并二次聚合（处理跨块边界）
    final_result = pd.concat(results).groupby(level=0).sum()  # 假设是sum聚合
    return final_result

# 使用示例
# final_sum = safe_groupby_aggregate('transactions.csv', by=['region'], agg_col='amount', agg_func='sum')

4.2 时间窗口漂移：你以为的“最近7天”，其实是“最近7条”

这是最隐蔽的坑。某次风控模型上线后，准确率暴跌，排查发现： rolling(window=7) 默认按 行序 而非 日期序 滑动！当数据未按日期排序时，“最近7天”变成“最近7条记录”，而那7条可能全是上周的。

解决方案：

• 强制日期排序 ：所有涉及时间窗口的操作，第一行必须是 df = df.sort_values('date') ；
• 索引转为DatetimeIndex ： df = df.set_index('date') ，这样 rolling() 自动按时间对齐；
• 添加时间校验 ：计算后检查 rolling_result.index.max() - rolling_result.index.min() 是否等于窗口大小。

# 时间校验函数
def validate_rolling_window(result_series, window_days=7, tolerance_days=1):
    """
    验证滚动窗口是否按时间对齐
    """
    if not isinstance(result_series.index, pd.DatetimeIndex):
        raise TypeError("索引必须为DatetimeIndex")
    
    # 计算实际时间跨度
    actual_span = (result_series.index.max() - result_series.index.min()).days
    if abs(actual_span - window_days) > tolerance_days:
        raise ValueError(f"滚动窗口时间跨度异常：期望{window_days}天，实际{actual_span}天")
    
    return True

# 使用
# validate_rolling_window(df['7day_avg'], window_days=7)

4.3 NaN传染：一个缺失值，毁掉整个指标链

现象：某日早报中，“华东区总交易额”显示为 NaN ，但各子区域数据都正常。追查发现，某家分行 category 字段有空值， groupby(['region','category']) 时，空值被归为一类， unstack() 后该