如何为sklearn GridSearchCV实现生存分析的Integrated Brier Score评分器

问题核心原因 #

你当前的错误来自两个核心问题：

1. make_scorer默认生成的评分函数仅能获取验证集真实标签y_true和评估器默认predict()方法的输出，无法拿到已拟合完成的评估器实例，也不能直接调用predict_survival_function这类自定义预测方法。
2. 你在评分函数内手动调用fit属于逻辑错误，GridSearch在每个折叠、每组参数下都会自动拟合评估器，重复拟合不仅逻辑错误还会导致数据泄露。

注意：你现有IB_time_interval函数中取时间的索引有误，sksurv的生存标签结构化数组每个元素的第一个维度是事件状态（索引0），第二个维度是生存时间（索引1），请把i[2]改为i[1]，否则会触发索引越界报错。

解决方案 #

提供两种适配不同需求的实现方案：

方案1：使用sklearn原生自定义评分器（改动最小）

sklearn的模型选择工具（包括GridSearchCV）的scoring参数支持传入签名为(estimator, X, y_true)的可调用对象，这里传入的estimator就是当前折叠下已经拟合完成的评估器，不需要手动拟合。如果你的时间区间可以用全局训练集提前计算，直接用这个方案即可。

修改后的完整可运行代码：

import numpy as np
from sksurv.ensemble import RandomSurvivalForest
from sklearn.model_selection import train_test_split
from sklearn.model_selection import GridSearchCV
from sksurv.metrics import integrated_brier_score
from sksurv.datasets import load_breast_cancer

# 修正后的时间区间计算函数
def IB_time_interval(y_train, y_test):
    y_times_tr = [i[1] for i in y_train]
    y_times_te = [i[1] for i in y_test]

    T1 = np.percentile(y_times_tr, 5, interpolation='higher')
    T2 = np.percentile(y_times_tr, 95, interpolation='lower')
    T3 = np.percentile(y_times_te, 5, interpolation='higher')
    T4 = np.percentile(y_times_te, 95, interpolation='lower')
    
    return np.linspace(np.maximum(T1,T3), np.minimum(T2, T4))

# 自定义IBS评分器，符合sklearn要求的签名
def ibs_scorer(estimator, X_val, y_val):
    surv_funcs = estimator.predict_survival_function(X_val, return_array=False)
    # 过滤有效时间区间
    min_t = surv_funcs[0].x.min()
    max_t = surv_funcs[0].x.max()
    valid_times = times_linspace[(times_linspace >= min_t) & (times_linspace < max_t)]
    # 生成生存概率矩阵
    pred_surv = np.array([[fn(t) for t in valid_times] for fn in surv_funcs])
    # IBS越小越好，返回负值让GridSearchCV默认最大化评分
    return -integrated_brier_score(y_train, y_val, pred_surv, valid_times)

# 数据加载拆分
X, y = load_breast_cancer()
X = X.drop(["er", "grade"], axis=1)
y_cens = np.array([i[0] for i in y]) 

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size= 0.2,
                                                    shuffle=True,
                                                    random_state=0,
                                                    stratify = y_cens)

param_grid = {
            'max_depth': [4, 20, None],
            'max_features': ["sqrt", None]
            }

rsf = RandomSurvivalForest(n_jobs=1, random_state=0)
times_linspace = IB_time_interval(y_train, y_test)

# 直接传入自定义评分器，不需要用make_scorer
clf = GridSearchCV(rsf, param_grid, refit=True, n_jobs=1, 
                    scoring=ibs_scorer)

clf.fit(X_train, y_train)
print("final score clf", clf.score(X_train, y_train))
print(clf.best_params_)

方案2：手写网格搜索+交叉验证循环（灵活度最高）

如果需要每个交叉验证折叠都单独计算时间区间，或者有其他自定义逻辑，推荐直接手写循环，避免GridSearch的黑盒限制，适配所有复杂评分场景：

from sklearn.model_selection import StratifiedKFold

# 分层交叉验证保证每折删失比例均匀
y_cens_train = np.array([e[0] for e in y_train])
kf = StratifiedKFold(n_splits=5, shuffle=True, random_state=0)

best_score = float('inf')
best_params = None

# 遍历所有参数组合
for max_depth in param_grid['max_depth']:
    for max_features in param_grid['max_features']:
        params = {'max_depth': max_depth, 'max_features': max_features}
        fold_scores = []
        # 遍历每个交叉验证折
        for train_idx, val_idx in kf.split(X_train, y_cens_train):
            X_tr_fold, X_val_fold = X_train.iloc[train_idx], X_train.iloc[val_idx]
            y_tr_fold, y_val_fold = y_train[train_idx], y_train[val_idx]
            # 拟合当前模型
            model = RandomSurvivalForest(**params, n_jobs=1, random_state=0)
            model.fit(X_tr_fold, y_tr_fold)
            # 单独计算当前折叠的时间区间
            times_fold = IB_time_interval(y_tr_fold, y_val_fold)
            # 预测生存函数计算IBS
            surv_funcs = model.predict_survival_function(X_val_fold, return_array=False)
            min_t = surv_funcs[0].x.min()
            max_t = surv_funcs[0].x.max()
            valid_times = times_fold[(times_fold >= min_t) & (times_fold < max_t)]
            pred_surv = np.array([[fn(t) for t in valid_times] for fn in surv_funcs])
            ibs = integrated_brier_score(y_tr_fold, y_val_fold, pred_surv, valid_times)
            fold_scores.append(ibs)
        # 记录最优参数
        avg_ibs = np.mean(fold_scores)
        print(f"参数{params} 平均IBS:{avg_ibs:.4f}")
        if avg_ibs < best_score:
            best_score = avg_ibs
            best_params = params

print(f"最优参数：{best_params}，最优IBS：{best_score:.4f}")
# 用最优参数在全量训练集拟合最终模型
final_model = RandomSurvivalForest(**best_params, n_jobs=1, random_state=0)
final_model.fit(X_train, y_train)

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