本文详解如何在 pandas 中对指定列（如 id、hrz、tenor）进行分组，并同时对多个数值列（如 1–4）批量应用多个自定义统计检验函数（如 ks 检验、cvm 检验），最终生成结构清晰、可直接分析的长格式结果表。
本文详解如何在 pandas 中对指定列（如 id、hrz、tenor）进行分组，并同时对多个数值列（如 1–4）批量应用多个自定义统计检验函数（如 ks 检验、cvm 检验），最终生成结构清晰、可直接分析的长格式结果表。
在金融、风控或计量分析场景中，常需对分组后的时序数据（如不同期限、评级、产品维度下的指标序列）执行分布一致性检验。例如，对每组 id+hrz+tenor 下的各列（1, 2, 3, 4）分别计算 Kolmogorov-Smirnov（KS）统计量和 Cramér–von Mises（CVM）统计量，以评估其经验分布是否接近均匀分布。
以下为完整、可复现的实现方案（需提前导入依赖）：
import pandas as pd
import numpy as np
from scipy import stats
# 示例自定义函数（注意：ks_test 返回统计量，非 p-value）
def ks_test(x):
# 使用 uniform 分布进行单样本 KS 检验，返回 D 统计量
return stats.kstest(np.sort(x), 'uniform')[0]
def cvm_test(x):
n = len(x)
if n == 0:
return np.nan
i = np.arange(1, n + 1)
x_sorted = np.sort(x)
w2 = (1 / (12 * n)) + np.sum((x_sorted - ((2 * i - 1) / (2 * n))) ** 2)
return w2
✅ 推荐方法一：使用 agg() + pd.concat()（结构可控、命名明确）
该方法显式构建字典映射函数名 → 聚合结果，便于调试与扩展：
group_cols = ['id', 'hrz', 'tenor']
value_cols = df.columns.difference(group_cols + ['date']) # 自动排除分组列和 date
g = df.groupby(group_cols)[value_cols]
result = pd.concat({
'ks_test': g.agg(ks_test),
'cvm_test': g.agg(cvm_test)
}, names=['test']).reset_index()
✅ 输出特点：
- 新增
test列，值为'ks_test'或'cvm_test'； - 原始分组列（
id,hrz,tenor）与数值列（1,2,3,4）保持原名； - 行顺序默认按字典序排列
test，可通过.sort_values(['id','hrz','tenor','test'])显式排序。
✅ 推荐方法二：使用 agg([f1, f2]) + stack()（代码更紧凑）
适用于函数数量适中、且希望函数名自动作为层级索引的场景：
result = (
df.groupby(group_cols)[value_cols]
.agg([ks_test, cvm_test])
.rename_axis([None, 'test'], axis=1) # 将列二级索引命名为 'test'
.stack('test') # 展平为长格式
.reset_index()
)
⚠️ 注意事项：
-
agg([f1,f2])会将函数名（如functools.wraps或为函数设置__name__，否则列名不可读。推荐显式命名函数或改用方法一； - 若某组数据为空或含缺失值，
ks_test/cvm_test需自行处理NaN（示例中cvm_test已加空值防护）； -
scipy.stats.kstest对小样本敏感，建议确保每组至少有 5 个有效观测值； - 如需同时输出统计量和 p 值，可修改函数返回
pd.Series({'statistic': ..., 'pvalue': ...})，再配合agg()的字典语法。
? 总结
两种方法均能高效实现「多列分组 + 多函数 + 多目标列」的聚合需求。
-
追求可读性与稳定性 → 选
concat({name: agg(func)})； -
追求简洁性且函数命名规范 → 选
agg([f1,f2]).stack()。
无论哪种方式，最终均可无缝接入后续分析流程（如 pivot 成宽表、可视化分组差异、导出报告等）。
