Python Pandas：按多字段分组并对多列应用自定义统计函数

现有如下示例DataFrame：

id  date        hrz tenor   1       2       3       4
AAA 16/03/2010  2   6m      0.54    0.54    0.78    0.19
AAA 30/03/2010  2   6m      0.05    0.67    0.20    0.03
AAA 13/04/2010  2   6m      0.64    0.32    0.13    0.20
AAA 27/04/2010  2   6m      0.99    0.53    0.38    0.97
AAA 11/05/2010  2   6m      0.46    0.90    0.11    0.14
AAA 25/05/2010  2   6m      0.41    0.06    0.96    0.31
AAA 08/06/2010  2   6m      0.19    0.73    0.58    0.80
AAA 22/06/2010  2   6m      0.40    0.95    0.14    0.56
AAA 06/07/2010  2   6m      0.22    0.74    0.85    0.94
AAA 20/07/2010  2   6m      0.34    0.17    0.03    0.77
AAA 03/08/2010  2   6m      0.13    0.32    0.39    0.95
AAA 16/03/2010  2   1y      0.54    0.54    0.78    0.19
AAA 30/03/2010  2   1y      0.05    0.67    0.20    0.03
AAA 13/04/2010  2   1y      0.64    0.32    0.13    0.20
AAA 27/04/2010  2   1y      0.99    0.53    0.38    0.97
AAA 11/05/2010  2   1y      0.46    0.90    0.11    0.14
AAA 25/05/2010  2   1y      0.41    0.06    0.96    0.31
AAA 08/06/2010  2   1y      0.19    0.73    0.58    0.80
AAA 22/06/2010  2   1y      0.40    0.95    0.14    0.56
AAA 06/07/2010  2   1y      0.22    0.74    0.85    0.94
AAA 20/07/2010  2   1y      0.34    0.17    0.03    0.77
AAA 03/08/2010  2   1y      0.13    0.32    0.39    0.95

需要按id、hrz、tenor字段分组，对列1、2、3、4应用以下两个自定义函数：

import scipy.stats
import numpy as np

def ks_test(x):
    return scipy.stats.kstest(np.sort(x), 'uniform')[0]

def cvm_test(x):
    n = len(x)
    i = np.arange(1, n + 1)
    x = np.sort(x)
    w2 = (1 / (12 * n)) + np.sum((x - ((2 * i - 1) / (2 * n))) ** 2)
    return w2

最终希望得到如下格式的输出DataFrame（数值为示例）：

id   hrz    tenor   test        1       2       3       4
AAA  2      6m      ks_test     0.04    0.06    0.02    0.03
AAA  2      6m      cvm_test    0.09    0.17    0.03    0.05
AAA  2      1y      ks_test     0.04    0.06    0.02    0.03
AAA  2      1y      cvm_test    0.09    0.17    0.03    0.05

可以通过以下步骤实现需求：

• 导入必要库并定义函数
  确保先导入pandas、numpy和scipy.stats，并定义好两个测试函数：

import pandas as pd
import numpy as np
import scipy.stats

def ks_test(x):
    return scipy.stats.kstest(np.sort(x), 'uniform')[0]

def cvm_test(x):
    n = len(x)
    i = np.arange(1, n + 1)
    x = np.sort(x)
    w2 = (1 / (12 * n)) + np.sum((x - ((2 * i - 1) / (2 * n))) ** 2)
    return w2

• 分组并应用函数
  使用groupby按指定字段分组，然后对目标列同时应用两个函数：

# 假设原始数据存储在df中
grouped = df.groupby(['id', 'hrz', 'tenor'])[[1,2,3,4]].agg([ks_test, cvm_test])

• 调整列结构匹配目标格式
  上述操作会生成多层列索引，需要重排结构并添加test列：

# 重排索引，将测试名称转为行维度
grouped = grouped.unstack(level=1).stack(level=0).reset_index()
# 重命名列名
grouped.columns = ['id', 'hrz', 'tenor', 'test', 1, 2, 3, 4]
# 排序确保顺序与示例一致
grouped = grouped.sort_values(['id', 'hrz', 'tenor', 'test']).reset_index(drop=True)

或者用更直观的分两次计算再合并的方式：

# 分别计算两个测试的结果
ks_result = df.groupby(['id', 'hrz', 'tenor'])[[1,2,3,4]].apply(ks_test).reset_index()
ks_result['test'] = 'ks_test'

cvm_result = df.groupby(['id', 'hrz', 'tenor'])[[1,2,3,4]].apply(cvm_test).reset_index()
cvm_result['test'] = 'cvm_test'

# 合并结果并调整列顺序
final_df = pd.concat([ks_result, cvm_result], ignore_index=True)
final_df = final_df[['id', 'hrz', 'tenor', 'test', 1, 2, 3, 4]]
final_df = final_df.sort_values(['id', 'hrz', 'tenor', 'test']).reset_index(drop=True)

执行后，final_df即为目标格式的输出结果。

内容的提问来源于stack exchange，提问作者Whitebeard13