本文详解如何在 pandas 中基于 id、hrz 和 tenor 多列分组，对数值列（如 1–4）并行应用多个自定义函数（如 KS 检验、CVM 检验），并规范输出为长格式结果表。
本文详解如何在 pandas 中基于 `id`、`hrz` 和 `tenor` 多列分组，对数值列（如 `1`–`4`）并行应用多个自定义函数（如 ks 检验、cvm 检验），并规范输出为长格式结果表。
在金融、风控或统计建模场景中，常需对分组后的时序或截面数据批量执行分布检验（如 Kolmogorov-Smirnov 或 Cramér-von Mises 检验）。pandas 提供了灵活高效的 groupby.agg() 机制，支持对多列同时应用多个自定义函数，并可精确控制输出结构。
以下以实际示例演示两种推荐方案——二者均保持原始分组逻辑清晰、结果可读性强，且适配任意数量的数值列与自定义函数。
✅ 方案一：pd.concat + agg()（推荐用于函数命名明确、需精细控制层级）
import pandas as pd
import numpy as np
from scipy import stats
# 定义自定义检验函数（注意：ks_test 返回统计量，非 p 值）
def ks_test(x):
# 对 x 排序后与 Uniform(0,1) 分布比较（假设输入已归一化或为[0,1]区间）
return stats.kstest(np.sort(x), 'uniform')[0]
def cvm_test(x):
n = len(x)
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
# 假设 df 为原始 DataFrame
group_cols = ['id', 'hrz', 'tenor']
value_cols = df.columns.difference(group_cols + ['date']) # 自动排除分组列和 date
# 分组并分别聚合
g = df.groupby(group_cols)[value_cols]
result_dict = {
'ks_test': g.agg(ks_test),
'cvm_test': g.agg(cvm_test)
}
# 合并为 MultiIndex DataFrame，test 作为最外层索引
out = pd.concat(result_dict, names=['test']).reset_index()
✅ 输出结构：test 列位于最左，分组列（id, hrz, tenor）紧随其后，各数值列（1, 2, 3, 4）保持原名，便于后续 pivot 或绘图。
✅ 方案二：agg([f1, f2]) + stack()（简洁一行式，函数名自动作为 level）
out = (
df.groupby(group_cols)[value_cols]
.agg([ks_test, cvm_test])
.rename_axis([None, 'test'], axis=1) # 将列二级索引命名为 'test'
.stack(level='test') # 展开 test 级别 → 转为行索引
.reset_index() # 转为标准 DataFrame
)
⚠️ 注意：此方式要求函数对象具有 __name__ 属性（默认满足），若使用 lambda 或重命名函数，建议改用方案一或显式传入元组 ('ks', ks_test)。
? 关键注意事项
-
输入数据预处理：KS 和 CVM 检验通常要求输入变量服从
[0,1]均匀分布（或经概率积分变换后）。若原始列非概率值，请先通过rank(method='min') / len(x)或scipy.stats.probplot等方式标准化。 -
缺失值处理：
agg()默认跳过 NaN；如需保留或报错，可在groupby(..., dropna=False)或函数内添加x = x.dropna()显式控制。 -
性能优化：对大数据集，避免在函数内重复调用
np.sort()；可提前对value_cols整体排序缓存，或改用numba加速 CVM 计算。 -
扩展性：新增函数只需在
result_dict或agg([...])列表中追加，无需修改主逻辑。
通过上述任一方法，即可高效生成符合分析需求的宽→长结构检验结果表，无缝对接下游可视化（如 seaborn.catplot）或模型特征工程流程。
