Conformal Learning 求解回归问题和多标签分类问题

Conformal Learning 是一种非参数统计方法，利用“样本属于某个标签时的离群程度”来进行回归和分类。本文分别使用“老忠实泉的爆发和等待时间数据”进行回归预测，使用“玻璃分类数据”进行多标签分类预测。

参考文献：A Tutorial on Conformal Prediction

回归问题

1. 对于训练集的某一个样本 \(i\)，找到离样本 \(i\) 最近的样本。

   • 若最近的样本只有一个，记为样本 \(j\)，则计算样本 \(i\) 和 样本 \(j\) 的标签之间的差值的绝对值；

   • 若最近的样本有多个，则先计算这多个样本的标签的中位数，再将样本 \(i\) 的标签值与该中位数做差后取绝对值。

2. 此“绝对值”就衡量了样本 \(i\) 的离群程度。

3. 对于一个新样本 \(n\)，同样找到离样本 \(n\) 最近的样本，用“离样本 \(n\) 最近的一个或多个样本的标签的中位数”作为新样本的标签预测值。

4. 根据信心水平 \(level\)（例如 \(90\%\)），选定一个离群程度，使得该离群程度在所有训练样本的离群程度中的大小排名分位数是 \(1-level\)（例如 \(10\%\)，即100个数中第10大的数）。

5. 在该预测值的基础上加减上一步选定的离群程度，就得到新样本标签值的预测区间。

分类问题

1. 对于一个新样本，为其赋予所有可能的标签后，将其纳入训练样本中，形成一个 Bag。
2. 对于 Bag 中的每一个样本：
   1. 对于与该样本的标签相同的其他样本，计算它们与该样本的距离，从中选择最小的，作为分子。
   2. 对于与该样本的标签不同的其他样本，计算它们与该样本的距离，从中选择最小的，作为分母。
   3. 将前两步的分子除以分母，即可衡量“为该样本赋予该标签时的离群程度”。该值越大，说明分子越大、分母越小。
      • 分子越大，说明虽然标签相同但距离很远，可以推测这个样本很可能并不属于这个标签。
      • 分母越大，说明虽然标签不同但距离很近，可以推测这个样本很可能属于其他标签。
3. 对于每一个可能的标签，根据信心水平 \(level\)（例如 \(90\%\)），判断：当新样本确实属于这个标签时，Bag 中有多少比例样本的离群程度比新样本的离群程度更高。如果这个比例超过了 \(1-level\)（例如 \(10\%\)），则将这个标签加入到预测标签集中。
4. 输出预测标签集，它可能有一个或多个预测值，也可能是空集。

Problem 1：使用 Conformal Learning 求解回归问题

导入包

Python

import pandas as pd
import numpy as np
from scipy.spatial.distance import cdist

导入数据

Python

# 读取 faithful.dat 数据集，两列数据分别为喷发时间和等待时间
df = pd.read_csv("./P1/faithful.dat", delimiter="\s+").reset_index(drop=True)
# 选取数据集的前 100 行作为训练集
train_df = df.iloc[:100, :]

基于 nearest neighbor 计算 nonconformity score

Python

def get_nonconformity_scores(train_df):
    x = train_df["eruptions"].values.reshape(-1, 1)
    y = train_df["waiting"].values.reshape(-1, 1)
    # 计算 x 之间的距离矩阵
    distance = cdist(x, x, "euclidean")
    # 将对角线元素设为无穷大
    np.fill_diagonal(distance, np.inf)
    # 找到每一行的最小值
    min_distance = np.min(distance, axis=1)
    # 找到每一行的最小值的索引
    min_index_row, min_index_column = np.where(distance == min_distance[:, None])
    min_dict = {
        key: [val for i, val in enumerate(min_index_column) if min_index_row[i] == key]
        for key in set(min_index_row)
    }
    # 计算离每个样本最近的样本的 y 的中位数，做差后计算绝对值，作为 nonconformity score
    nonconformity_scores = []
    for i in range(train_df.shape[0]):
        nonconformity_scores.append(abs(y[i] - np.median(y[min_dict[i]]))[0])
    return nonconformity_scores

计算每个新样本的 prediction interval

Python

def get_prediction_interval(train_df, new_x, level):
    # 计算训练样本的 nonconformity score
    nonconformity_scores = get_nonconformity_scores(train_df)
    # 计算新样本的与训练样本的距离的最小值
    min_distance = np.min(
        abs(train_df["eruptions"].values.reshape(-1, 1) - new_x), axis=0
    )
    # 找到最小值的索引
    min_index = np.where(
        abs(train_df["eruptions"].values.reshape(-1, 1) - new_x) == min_distance
    )[0]
    # 计算新样本的预测值
    prediction = np.median(train_df["waiting"].values[min_index])
    # 计算 nonconformity scores 的 level 分位数，即保证有 1 - level 的样本的 nonconformity score 大于等于该值
    nonconformity_score_at_level = nonconformity_scores[
        np.argsort(nonconformity_scores)[::-1][
            int((1 - level) * len(nonconformity_scores))
        ]
    ]
    # 计算新样本的 predition interval
    lower_bound = prediction - nonconformity_score_at_level
    upper_bound = prediction + nonconformity_score_at_level
    return lower_bound, upper_bound

计算不同置信度下的预测准确率

Python

level_list = [0.9, 0.8, 0.6]
for level in level_list:
    count = 0
    for i in range(100, df.shape[0]):
        lower_bound, upper_bound = get_prediction_interval(
            train_df, df["eruptions"].iloc[i], level
        )
        if df["waiting"].iloc[i] > lower_bound and df["waiting"].iloc[i] < upper_bound:
            count += 1
    correct_rate = count / (i - 100 + 1)
    print("level: {:.2%}, correct rate: {:.2%}".format(level, correct_rate))

Text Only

level: 90.00%, correct rate: 86.05%
level: 80.00%, correct rate: 79.07%
level: 60.00%, correct rate: 51.16%

在不同置信度下，预测准确率大致略低于置信度。

Problem 2：使用 Conformal Learning 求解多标签分类问题

导入包

Python

from sklearn.preprocessing import StandardScaler

导入数据

Python

# 读取 glass.data 数据集
df = pd.read_csv("./P2/glass.data", delimiter=",", header=None, index_col=0)
# 创建标准化器对象
scaler = StandardScaler()
# 随机选择 150 个样本作为训练集
train_df = df.sample(n=150, random_state=0)
# 对自变量进行标准化
train_df.iloc[:, :-1] = scaler.fit_transform(train_df.iloc[:, :-1])
# 剩下的样本作为测试集
test_df = df.drop(train_df.index)
# 对自变量进行标准化
test_df.iloc[:, :-1] = scaler.transform(test_df.iloc[:, :-1])
# 重新索引
train_df = train_df.reset_index(drop=True)
test_df = test_df.reset_index(drop=True)

基于 nearest neighbor 计算 nonconformity score

Python

def get_nonconformity_scores(train_df, new_x, new_y):
    x = pd.concat([train_df.iloc[:, :-1], new_x.to_frame().T], axis=0).reset_index(
        drop=True
    )
    y = pd.concat([train_df.iloc[:, -1], pd.Series([new_y])]).reset_index(drop=True)
    # 计算 x 之间的距离矩阵
    distance = cdist(x, x, "euclidean")
    # 将对角线元素设为无穷大
    np.fill_diagonal(distance, np.inf)
    # 计算相同标签的最小距离和不同标签的最小距离的比值，作为 nonconformity score
    nonconformity_scores = []
    for i in range(x.shape[0]):
        # 找到第 i 行中，与样本 i 标签相同的所有样本，并求其距离的最小值
        min_distance_with_same_label = np.min(distance[i, y == y[i]])
        # 找到第 i 行中，与样本 i 标签不同的所有样本，并求其距离的最小值
        min_distance_with_different_label = np.min(distance[i, y != y[i]])
        # 计算第 i 行的 nonconformity score
        nonconformity_score = (
            min_distance_with_same_label / min_distance_with_different_label
        )
        nonconformity_scores.append(nonconformity_score)
    return nonconformity_scores

计算每个新样本的 prediction set

Python

def get_prediction_set(train_df, new_x, level):
    # 遍历所有可能的标签值，计算新样本在该标签下的 p value
    p_values = []
    labels = list(set(train_df.iloc[:, -1]))
    for label in labels:
        # 计算训练样本的 nonconformity scores
        nonconformity_scores = get_nonconformity_scores(train_df, new_x, label)
        # 计算新样本的 nonconformity score 在训练样本中的排名
        rank = sum(nonconformity_scores >= nonconformity_scores[-1])
        # 计算 p value
        p_value = rank / len(nonconformity_scores)
        p_values.append(p_value)
    # 找到 p value 大于 1 - level 的标签
    prediction_set = [
        labels[i] if p_values[i] > 1 - level else None for i in range(len(p_values))
    ]
    # 去掉 None
    prediction_set = [i for i in prediction_set if i is not None]
    return prediction_set

计算不同置信度下的预测准确率

Python

level_list = [0.9, 0.8, 0.6]
for level in level_list:
    count = 0
    for i in range(test_df.shape[0]):
        prediction_set = get_prediction_set(train_df, test_df.iloc[i, :-1], level)
        if test_df.iloc[i, -1] in prediction_set:
            count += 1
    correct_rate = count / test_df.shape[0]
    print("level: {:.2%}, correct rate: {:.2%}".format(level, correct_rate))

Text Only

level: 90.00%, correct rate: 90.62%
level: 80.00%, correct rate: 76.56%
level: 60.00%, correct rate: 57.81%

在不同置信度下，预测准确率与置信度十分相近。

评论