Python中基于类别标签的连续变量监督式离散化：优化阈值以提升与标注类别一致性的方法

我手头有一组样本数据，包含两个核心字段：

• 连续变量cont_attribute：取值范围0-100，用于量化样本状态
• 有序类别变量class_label：人工标注的样本状态，分为1、2、3、4四个等级

由于cont_attribute的取值在不同类别间存在重叠，我需要对这个连续变量做离散化处理，目标是让离散后的结果和class_label的一致性达到最优。离散化时需要设置三个阈值x1、x2、x3，将连续变量划分为4个有序区间，并用以下代码评估和标注类别的一致性：

cohen_kappa_score(
    pd.cut(df['cont_attribute'], bins=[0, x1, x2, x3, 100], labels=['1','2','3','4']).astype('int'),
    df['class_label'].astype('int')
)

现有尝试遇到的问题 #

我已经试过Jenks自然断裂法、sklearn的Kmeans这些无监督离散化方法，但它们都没有考虑类别标签的信息，效果不尽如人意。后来我尝试用scipy.optimize的优化工具来最大化上述Cohen's Kappa值，给每个阈值限定了范围（高类别cont_attribute的最小值到低类别cont_attribute的最大值），但运行时出现了错误：ValueError: bins must increase monotonically。相关代码如下：

def objfunc(grid):
    x1, x2, x3 = grid
    return (-1) * cohen_kappa_score(
        pd.cut(df.cont_attribute, bins=[0, x1, x2, x3, 100], labels=['1','2','3','4'], duplicates='drop').astype('int'),
        df['class_label'].astype('int')
    )

grid = (
    slice(df[(df['class_label'] == 2)]['cont_attribute'].min(), df[(df['class_label'] == 1)]['cont_attribute'].max(), 0.5),
    slice(df[(df['class_label'] == 3)]['cont_attribute'].min(), df[(df['class_label'] == 2)]['cont_attribute'].max(), 0.5),
    slice(df[(df['class_label'] == 4)]['cont_attribute'].min(), df[(df['class_label'] == 3)]['cont_attribute'].max(), 0.5)
)

solution = brute(objfunc, grid, finish=None, full_output=True)
solution

我现在有两个疑问：

1. Python中有没有直接的监督式离散化方法，可以在优化阈值x1、x2、x3时直接考虑和类别标签的一致性？
2. 如果没有现成方法，应该怎么修改上面的代码，让它能通过scipy.optimize.minimize得到最大值？

解决方案 #

一、现成的监督式离散化工具 #

其实Python里有几个专门的库可以解决这个问题，不需要手动写优化逻辑：

1. Feature-engine的DecisionTreeDiscretiser

Feature-engine是专注于特征工程的库，它的DecisionTreeDiscretiser用决策树做监督式离散化——决策树的分裂点就是我们要找的阈值，而且是基于类别标签优化的，完美匹配你的需求。

用法示例：

from feature_engine.discretisation import DecisionTreeDiscretiser
import pandas as pd

# 初始化离散器：指定目标是分类任务，用5折交叉验证优化，分4个区间
dt_discretiser = DecisionTreeDiscretiser(
    variables=['cont_attribute'],
    cv=5,
    scoring='accuracy',
    regression=False,
    param_grid={'max_depth': [2,3,4], 'min_samples_leaf': [10,20]}
)

# 拟合数据，自动找到最优分裂点
dt_discretiser.fit(df, df['class_label'])

# 应用离散化
df['cont_discretised'] = dt_discretiser.transform(df)

# 查看得到的分箱边界
print(dt_discretiser.binner_dict_['cont_attribute'])

2. OptBinning库

OptBinning是专门为最优分箱设计的工具，支持分类/回归任务的监督式分箱，能直接以Cohen's Kappa为目标优化阈值，完全贴合你的场景。

用法示例：

from optbinning import OptimalBinning

# 初始化最优分箱器：指定数值型变量，分4个箱，用Cohen's Kappa作为优化指标
optb = OptimalBinning(
    name='cont_attribute',
    dtype='numerical',
    solver='cp',
    n_bins=4,
    metric='kappa'
)

# 拟合数据
optb.fit(df['cont_attribute'], df['class_label'])

# 查看最优阈值
print(optb.splits)

# 应用分箱
df['cont_discretised'] = optb.transform(df['cont_attribute'], metric='kappa')

这个库会直接输出你需要的x1、x2、x3，而且保证是最优解。

二、修改现有scipy优化代码 #

如果不想引入新库，核心问题是要保证x1 < x2 < x3，因为分箱边界必须单调递增。这里有两种修改思路：

1. 重新定义变量，强制阈值递增

我们可以把变量定义为增量形式，天生满足单调关系：

import pandas as pd
from sklearn.metrics import cohen_kappa_score
from scipy.optimize import brute

def objfunc(params):
    a, b, c = params
    x1 = a
    x2 = x1 + b
    x3 = x2 + c
    
    # 确保x3不超过100，否则返回极大值让优化器避开
    if x3 >= 100:
        return float('inf')
    
    try:
        kappa = cohen_kappa_score(
            pd.cut(df['cont_attribute'], bins=[0, x1, x2, x3, 100], labels=['1','2','3','4']).astype('int'),
            df['class_label'].astype('int')
        )
        return -kappa  # 最大化kappa等价于最小化负kappa
    except ValueError:
        return float('inf')

# 设置参数范围：a是x1的初始值，b和c是正增量
grid = (
    slice(0, 80, 0.5),    # x1的范围，可根据你的数据调整
    slice(0.1, 20, 0.5),  # x2 = x1 + b，保证x2 > x1
    slice(0.1, 20, 0.5)   # x3 = x2 + c，保证x3 > x2
)

# 运行 brute 搜索
solution = brute(objfunc, grid, finish=None, full_output=True)

# 解析结果
best_a, best_b, best_c = solution[0]
best_x1 = best_a
best_x2 = best_x1 + best_b
best_x3 = best_x2 + best_c
best_kappa = -solution[1]

print(f"最优阈值：x1={best_x1:.2f}, x2={best_x2:.2f}, x3={best_x3:.2f}")
print(f"对应的Cohen's Kappa值：{best_kappa:.4f}")

2. 带约束的minimize优化

用scipy.optimize.minimize的SLSQP方法，添加明确的约束条件保证阈值递增：

from scipy.optimize import minimize

def objfunc(params):
    x1, x2, x3 = params
    try:
        kappa = cohen_kappa_score(
            pd.cut(df['cont_attribute'], bins=[0, x1, x2, x3, 100], labels=['1','2','3','4']).astype('int'),
            df['class_label'].astype('int')
        )
        return -kappa
    except ValueError:
        return float('inf')

# 初始猜测值，可基于类别均值设置
x0 = [
    df[df.class_label==1]['cont_attribute'].mean(),
    df[df.class_label==2]['cont_attribute'].mean(),
    df[df.class_label==3]['cont_attribute'].mean()
]

# 约束条件：x1 < x2，x2 < x3，x1>0，x3<100
constraints = [
    {'type': 'ineq', 'fun': lambda x: x[1] - x[0]},  # x2 - x1 > 0
    {'type': 'ineq', 'fun': lambda x: x[2] - x[1]},  # x3 - x2 > 0
    {'type': 'ineq', 'fun': lambda x: x[0]},         # x1 > 0
    {'type': 'ineq', 'fun': lambda x: 100 - x[2]}    # 100 - x3 > 0
]

# 变量边界
bounds = [(0, 100), (0, 100), (0, 100)]

# 运行优化
result = minimize(objfunc, x0, method='SLSQP', constraints=constraints, bounds=bounds)

# 输出结果
if result.success:
    best_x1, best_x2, best_x3 = result.x
    best_kappa = -result.fun
    print(f"最优阈值：x1={best_x1:.2f}, x2={best_x2:.2f}, x3={best_x3:.2f}")
    print(f"对应的Cohen's Kappa值：{best_kappa:.4f}")
else:
    print("优化失败：", result.message)

内容的提问来源于stack exchange，提问作者Levi.Steinberg