【代码复现】ELDB

1、算法

本文是论文《Multi-Instance Ensemble Learning With Discriminative Bags》中算法代码的复现。

2、代码

2.1、MIL.py

这一部分为多示例学习的原型文件，用于获取数据集名词、包空间、包大小等。

import warnings
import numpy as np
import os as os
from Function import load_file
warnings.filterwarnings("ignore")class MIL:def __init__(self,data_path,save_home="../Data/Distance/",bag_space=None):""":param data_path: 数据集的存储路径:param save_home: 距离矩阵的存储主目录:param bag_space: 格式与.mat文件一致"""self.data_path = data_pathself.save_home = save_homeself.bag_space = bag_spaceself.__init_mil()def __init_mil(self):"""初始化函数:return: nullN:包空间的大小bag_size:记录每个包大小的向量，长度为N，类型为intbag_lab:包标签向量bag_idx:包索引向量n:实例数量d:实例的维度C:数据里的类别数ins_space:实例空间ins_idx:实例空间中 包所对应的实例的范围ins_lab:实例标签ins_bag_idx:实例空间中 实例对应的包的序号data_name:数据集的名词zero_ratio:数据集含零比率"""if self.bag_space is None:self.bag_space = load_file(self.data_path)self.N = len(self.bag_space)self.bag_size = np.zeros(self.N,dtype=int)self.bag_lab = np.zeros_like(self.bag_size,dtype=int)self.bag_idx = np.arange(self.N)for i in range(self.N):self.bag_size[i] = len(self.bag_space[i][0])self.bag_lab[i] = self.bag_space[i][1]# 将所有包的标签调整到 [0, C - 1]的范围，C为数据集的类别数量self.__bag_lab_map()self.n = sum(self.bag_size)self.d = len(self.bag_space[0, 0][0]) - 1self.C = len(list(set(self.bag_lab)))self.ins_space = np.zeros((self.n, self.d))self.ins_idx = np.zeros(self.N + 1, dtype=int)self.ins_lab = np.zeros(self.n)self.ins_bag_idx = np.zeros(self.n, dtype=int)for i in range(self.N):self.ins_idx[i+1] = self.bag_size[i] + self.ins_idx[i]self.ins_space[self.ins_idx[i]: self.ins_idx[i + 1]] = self.bag_space[i, 0][:, :self.d]self.ins_lab[self.ins_idx[i]: self.ins_idx[i + 1]] = self.bag_space[i, 0][:, -1]self.ins_bag_idx[self.ins_idx[i]: self.ins_idx[i + 1]] = np.ones(self.bag_size[i]) * iself.data_name = self.data_path.strip().split("/")[-1].split(".")[0]self.zero_ratio = len(self.ins_space[self.ins_space == 0]) / (self.n * self.d)self.__generate_save_home()def __generate_save_hone(self):"""Generate the save home.如果不存在存储路径，则生成:return:"""if not os.path.exists(self.save_home):os.makedirs(self.save_home)def __bag_lab_map(self):"""将包标签映射为对应的类别 [0, 1, 2, ...]:return:"""lab_list = list(set(self.bag_lab))lab_dict = {}for i, lab in enumerate(lab_list):lab_dict[lab] = ifor i in range(self.N):self.bag_lab[i] = lab_dict[self.bag_lab[i]]def get_data_info(self):"""Print the data set information.打印数据集信息data_name：   数据集的名称bag_space：   包空间，详细格式请查看../Data/Benchmark/musk1+.matins_space：   实例空间bag_size：    记录每个包大小的向量，长度为Nbag_lab：     包标签向量ins_lab：     实例标签bag_idx：     包索引向量ins_idx：     实例空间中 包所对应的实例的范围ins_bag_idx： 实例空间中 实例对应的包的序号zero_ratio：  数据集含零比率N：           包空间的大小n：           实例数量d：           实例的维度C：           数据集的类别数"""temp_idx = 5 if self.N > 5 else self.Nprint("The {}'s information is:".format(self.data_name), "\n""Number bags:", self.N, "\n""Number classes:", self.C, "\n""Bag size:", self.bag_size[:temp_idx], "...\n""Bag label", self.bag_lab[:temp_idx], "...\n""Maximum bag's size:", np.max(self.bag_size), "\n""Minimum bag's size:", np.min(self.bag_size), "\n""Zero ratio:", self.zero_ratio, "\n""Number instances:", self.n, "\n""Instance dimensions:", self.d, "\n""Instance index:", self.ins_idx[: temp_idx], "...\n""Instance label:", self.ins_lab[: temp_idx], "...\n""Instance label corresponding bag'S index:", self.ins_bag_idx[:temp_idx], "...\n")def get_sub_ins_space(self, bag_idx):"""提高包索引数组，返回一个实例空间子集:param bag_idx: 包索引:return:"""n = sum(self.bag_size[bag_idx])ret_ins_space = np.zeros((n,self.d))ret_ins_label = np.zeros(n)ret_ins_bag_idx = np.zeros(n, dtype=int)count = 0for i in bag_idx:bag_size = self.bag_size[i]ret_ins_space[count: count + bag_size] = self.bag_space[i, 0][:, :-1]ret_ins_label[count: count + bag_size] = self.bag_lab[i]ret_ins_bag_idx[count: count + bag_size] = icount += bag_sizereturn ret_ins_space, ret_ins_label, ret_ins_bag_idxif __name__ == '__main__':temp_file_name = r"ucsb_breast.mat"mil = MIL(temp_file_name)mil.get_info()

2.2 Distance.py

距离度量文件，提供了一些距离度量函数，包括：欧氏距离、平均豪斯多夫距离。用来测量包与包之间的相似度。

import os
import numpy as np
from Function import print_progress_bar
# 由于需要进行文件读取，所有这里进行了存储精度的控制
np.set_printoptions(precision=6)def i2i_euclidean(ins1, ins2):"""欧式距离:paramins1：  向量1ins2：  向量2@return两个向量的欧式距离值"""return np.sqrt(np.sum((ins1 - ins2)**2))def ave_hausdorff(bag1, bag2):"""平均豪斯多夫距离，用于度量两个包的相似度:param bag1: 包1:param bag2: 包2:return: 距离度量"""sum_dis = 0for ins1 in bag1:#计算包1中实例与最近实例的距离temp_min = np.inffor ins2 in bag2:temp_min = min(i2i_euclidean(ins1, ins2), temp_min)sum_dis += temp_mindef simple_dis(bag1, bag2):return i2i_euclidean(np.average(bag1, 0), np.average(bag2, 0))class B2B:"""用于初始化数据集相关的包距离矩阵"""def __init__(self, data_name, bags, b2b_type="ave", b2b_save_home="Data/Distance/"):""":param data_name: 数据集名称:param bags: 包空间:param b2b_type: 包间距离度量方式:param b2b_save_home: 默认距离矩阵存储目录"""self._data_name = data_nameself._bags = bagsself._b2b_type = b2b_typeself._b2b_save_home = b2b_save_homeself.__initialize__b2b()def __initialize__b2b(self):"""初始化函数:return:null"""# 存储计算的距离矩阵self._dis = []# 获取距离矩阵的存储路径self._save_b2b_path = self._b2b_save_home + "b2b_" + self._data_name + '_' + self._b2b_type + ".npz"self._b2b_name = {"ave": "ave_hausdorff","sim": "simple_dis"}self.__compute_dis()def __compute_dis(self):"""计算距离:return:"""if not os.path.exists(self._save_b2b_path):#包大小N = len(self._bags)dis = np.zeros((N,N))print("使用%s距离计算距离矩阵..." % self._b2b_name[self._b2b_type])if self._b2b_type == "ave":dis_func = ave_hausdorffelse:dis_func = simple_disfor i in range(N):#打印进度条print_progress_bar(i, N)for j in range(i,N):dis[i, j] = dis[j, i] = dis_func(self._bags[i][0][:, : -1], self._bags[j][0][:, : -1])print()np.savez(self._save_b2b_path, dis=dis)self._dis = np.load(self._save_b2b_path)['dis']def get_dis(self):"""获取距离矩阵"""return self._dis

2.3 Function.py

本文件提供一些工具函数，诸如：交叉验证、打印进度条、获取分类性能度量等。

import numpy as np
from scipy.io import loadmatdef load_file(data_path):"""载入.mat类型的多示例数据集:param data_path::return:"""return loadmat(data_path)['data']def get_iter(tr, tr_lab, te, te_lab):""":param tr: 训练集:param tr_lab: 训练集标签:param te::param te_lab::return: 相应的迭代器"""yield tr, tr_lab, te, te_labdef get_k_cv_idx(num_x, k=10):"""获取k次交叉验证的索引，本函数k取10:param num_x:数据集的大小:param k:交叉验证次数:return:训练集索引，测试集索引"""# 随机初始化索引，从数据集索引打乱rand_idx = np.random.permutation(num_x)# 每一折的大小fold = int(np.floor(num_x/ k))ret_tr_idx = []ret_te_idx = []for i in range(k):# 获取当前折的训练集索引tr_idx = rand_idx[0: i*fold].tolist()tr_idx.extend(rand_idx[(i + 1) * fold:])ret_tr_idx.append(tr_idx)# 添加当前折的测试集索引ret_te_idx.append(rand_idx[i * fold: (i + 1) * fold].tolist())return ret_tr_idx, ret_te_idxdef get_performance(type_performance):"""获取分类性能度量:param type_performance:分类性能度量指标:return: 分类性能度量函数"""ret_per = {}for type_per in type_performance:if type_per == "acc":# 如果是准确度from sklearn.metrics import accuracy_scoremetric = accuracy_scoreelse:# 否则为f1_scorefrom sklearn.metrics import f1_scoremetric = f1_scoreret_per[type_per] = metricreturn ret_perdef print_progress_bar(idx, size):"""打印进度条:param idx: 当前位置:param size: 总进度:return:"""print('\r' + '▇' * int(idx // (size / 50)) + str(np.ceil((idx + 1) * 100 / size)) + '%', end='')

2.4 ClassifyTool.py

本文件调用分类器，对多示例向量进行分类。

import warnings
warnings.filterwarnings("ignore")class Classify:"""调用sklearn库中的分类器，实现多示例学习的映射向量分类"""def __init__(self, classifier_type=None, performance_type=None):""":param classifier_type: 所使用分类器的列表，已有分类器包括“knn”、“svm":param performance_type: 性能度量指标列表"""self.__classifier_type = classifier_typeself.__performance_type = performance_typeself.tr_true_label_arr = {}self.tr_predict_arr = {}self.te_true_label_arr = {}self.te_predict_arr = {}self.tr_per = {}self.te_per = {}self.__init_classify()def __init_classify(self):"""分类器初始化:return:"""self.__classifier = []self.__performance_er = []if self.__classifier_type is None:#如果没有默认分类器self.__classifier_type = ["knn"]#默认knnfor classifier_type in self.__classifier_type:if classifier_type == "knn":from sklearn.neighbors import KNeighborsClassifierself.__classifier.append(KNeighborsClassifier)elif classifier_type == "svm":from sklearn.svm import SVCself.__classifier.append(SVC(max_iter=10000))elif classifier_type == "j48":from sklearn.tree import DecisionTreeClassifierself.__classifier.append(DecisionTreeClassifier())if self.__performance_type is None:self.__performance_type = ["f1_score"]for performance_type in self.__performance_type:if performance_type == "f1_score":from sklearn.metrics import f1_scoreself.__performance_er.append(f1_score())elif performance_type == "acc":from sklearn.metrics import accuracy_scoreself.__performance_er.append(accuracy_score())elif performance_type == "roc":from sklearn.metrics import roc_auc_scoreself.__performance_er.append(roc_auc_score())def __reset_record(self):"""重设记录向量:return:"""for classifier_type in self.__classifier_type:self.tr_predict_arr[classifier_type], self.tr_true_label_arr[classifier_type] = [], []self.tr_per[classifier_type] = []self.te_predict_arr[classifier_type], self.te_true_label_arr[classifier_type] = [], []self.te_per[classifier_type] = []

2.5 ELDB.py

ELDB算法主类，主要包括ELDB算法初始化、映射、记录得分等。

import numpy as np
import warnings
from ClassifyTool import Classify
from Distance import B2B
from Function import get_k_cv_idx, get_iter, get_performance
from MIL import MIL
from sklearn.metrics import euclidean_distances as eucl
warnings.filterwarnings('ignore')def compute_discer(vectors, labels):""":param vectors:实例向量:param labels:实例对应标签:return:"""positive_vectors, negative_vectors = [], []for i in range(len(vectors)):if labels[i] == 1:positive_vectors.append(vectors[i])#若为正包elif labels[i] == 0:negative_vectors.append(vectors[i])#若为负包positive_vectors = np.array(positive_vectors)#生成对应向量negative_vectors = np.array(negative_vectors)#均值向量positive_mean = np.mean(positive_vectors, axis=0)negative_mean = np.mean(negative_vectors, axis=0)#平均距离positive_dis = np.mean(eucl(positive_vectors), axis=None)negative_dis = np.mean(eucl(negative_vectors), axis=None)fenmu = positive_dis + negative_dis#分母return eucl([positive_mean], [negative_mean])[0][0] / fenmu# if fenmu > 1e-3 else 1e-3class ELDB(MIL):"""ELDB算法主类"""def __init__(self, data_path, psi=0.9, alpha=0.75, batch=None, psi_max=200,type_b2b="ave", mode_bag_init="g", mode_action="a", k=10,type_classifier=None, type_performance=None, print_loop=False,save_home="../Data/Distance/", bag_space=None):""":param data_path: 数据存储路径:param psi: 基础dBagSet的大小:param alpha: 学习率，即基础dBagSet的大小与训练集的比值:param batch: 批次大小:param psi_max: 基础dBagSet的最大容量:param type_b2b: 距离函数的类型:param mode_bag_init: 基础dBagSet的初始化模式:param mode_action: 行为模式:param k:k折交叉验证:param type_classifier:转为单实例后的分类器，默认None时使用["knn", "svm, "j48]:param type_performance:性能度量类型，默认None时使用["acc", "f1_score"]:param print_loop:是否输出每一折的轮次:param save_home:保存地址:param bag_space:"""super(ELDB, self).__init__(data_path, save_home=save_home, bag_space=bag_space)self._psi = psiself._alpha = alphaself._batch = batchself._psi_max = psi_maxself._type_b2b = type_b2bself._mode_bag_init = mode_bag_initself._mode_action = mode_actionself._k = kself._type_classifier = type_classifierself._type_performance = type_performanceself._print_loop = print_loopself.__init_eldb()def __init_eldb(self):"""ELDB的初始化函数:return:"""self._type_classifier = ["knn", "svm", "j48"] if self._type_classifier is None else self._type_classifierself._type_performance = ["accuracy", "f1_score"] if self._type_performance is None else self._type_performance# 距离矩阵self.dis = B2B(self.data_name, self.bag_space, self._type_b2b, self.save_home).get_dis()# 记录不同分类器、不同分类性能的分类结果self.lab_predict = {}# 记录按照交叉验证顺利的真实标签self.lab_true = []# 记录分类性能self.val_performance = {}def __reset_record(self):"""重设记录相关的变量:return:"""self.lab_predict = {}self.lab_true = []self.val_performance = {}def __get_classifier(self):"""获取分类器:return:"""return Classify(self._type_classifier, self._type_performance)def get_state(self):"""获取使用的分类器以及度量性能:return:"""return self._type_classifier, self._type_performancedef get_mapping(self):"""获取映射结果:return:"""def __dBagSet_update_r(para_idx_dBagSet, para_score_dBagSet, idx_cur, score_cur):""":param para_idx_dBagSet:包集合的索引:param para_score_dBagSet: 包集合所得分数拷贝:param idx_cur: 第i个包:param score_cur: 包的得分:return:"""for idx_find in np.arange(len(para_idx_dBagSet)):if score_cur > para_score_dBagSet[idx_find]:#若有包的分数大于包集合中某个包的分数continueelse:idx_find += 1#向前推进idx_find = len(para_idx_dBagSet) - 1 if idx_find == len(para_idx_dBagSet) else idx_find#如果超出界限para_idx_dBagSet[idx_find + 1:] = para_idx_dBagSet[idx_find: -1]para_score_dBagSet[idx_find + 1:] = para_score_dBagSet[idx_find: -1]para_idx_dBagSet[idx_find], para_score_dBagSet[idx_find] = idx_cur, score_curbreakreturn para_idx_dBagSet, para_score_dBagSet# 获取训练集和测试集的索引idxes_tr, idxes_te = get_k_cv_idx(self.N, self._k)# 正负包标签lab_positive, lab_negative = np.max(self.bag_lab), np.min(self.bag_lab)# 获取单实例分类器classifier = self.__get_classifier(self._type_classifier)# 性能度量器performance = get_performance(self._type_performance)# 记录参数重设self.__reset_record(self)#主循环discer_list = []for loop, (idx_tr, idx_te) in enumerate(zip(idxes_tr, idxes_te)):"""步骤0:初始化操作"""# 进度条if self._print_loop:print("第{}折交叉验证".format(loop))# 计算训练集、基准数据集和更新数据集的大小N_T = len(idx_tr)N_Ts = int(N_T * (1 - self._alpha))# 计算批次大小batch = N_Ts // 2 if self._batch is None else self._batch# 计算最大更新次数n_l = N_Ts // batchN_Td = N_T - (n_l * batch)# 获取T_d和T_s的索引idx_td, idx_ts = np.array(idx_tr[:N_Td]), np.array(idx_tr[N_Td:])"""步骤1：模型和参数初始化"""# 计算Delta矩阵matrix_Delta = np.zeros((N_Td, N_Td), dtype=int)for i in range(N_Td):for j in range(N_Td):# 若标签相同，Delta设置为1，否则为0if self.bag_lab[idx_td[i]] == self.bag_lab[idx_td[j]]:# 对角矩阵， 对应位置设置为-1matrix_Delta[i, j] = -1else:matrix_Delta[i, j] = 1# 计算Gamma矩阵matrix_Gamma = np.diag(np.sum(matrix_Delta, 1))# 计算L矩阵matrix_L = matrix_Gamma - matrix_Delta# 只保留L矩阵del matrix_Delta, matrix_Gamma# 基于整个T_d进行映射，先获取所有训练包的映射，后面再分配mapping_bag = self.dis[idx_tr, :][:, idx_td]# 使用矩阵乘法score_t = np.dot(np.dot(mapping_bag, matrix_L))# 对角元素是包的得分score_t = np.diag(score_t)# 获取T_d和T_s中每一包的得分score_td, score_ts = score_t[: N_Td], score_t[N_Td:]# 获取初始dBagSet的大小psi = int(min(self._psi_max, N_Td) * self._psi)# 由大到小的得分索引排序arg_score_td = np.argsort(score_td)[::-1]# 获取dBagSet在训练集中的真实索引和相应得分if self._mode_bag_init == 'g':idx_dBagSet = arg_score_td[:psi].tolist()else:idx_dBagSet = []count = 0for i in arg_score_td:if count >= psi:breakif (self._mode_bag_init == 'p' and self.bag_lab[idx_td[i]] == lab_positive) or \(self._mode_bag_init == 'n' and self.bag_lab[idx_td[i]] != lab_positive):idx_dBagSet.append(i)count += 1score_dBagSet, idx_dBagSet = score_td[idx_dBagSet], [idx_td[idx_dBagSet].tolist()]del score_t, arg_score_td# 记录最小得分的索引和得分tau, p = len(idx_dBagSet[-1])

【代码复现】ELDB相关推荐

【神经网络】(17) EfficientNet 代码复现，网络解析，附Tensorflow完整代码
各位同学好,今天和大家分享一下如何使用 Tensorflow 复现 EfficientNet 卷积神经网络模型. EfficientNet 的网络结构和 MobileNetV3 比较相似,建议大家在学 ...
【神经网络】(16) MobileNetV3 代码复现，网络解析，附Tensorflow完整代码
各位同学好,今天和大家分享一下如何使用 Tensorflow 构建 MobileNetV3 轻量化网络模型. MobileNetV3 做了如下改动(1)更新了V2中的逆转残差结构:(2)使用NAS搜索 ...
【神经网络】(15) Xception 代码复现，网络解析，附Tensorflow完整代码
各位同学好,今天和大家分享一下如何使用 Tensorflow 构建 Xception 神经网络模型. 在前面章节中,我已经介绍了很多种轻量化卷积神经网络模型,感兴趣的可以看一下:https://blo ...
【神经网络】(14) MnasNet 代码复现，网络解析，附Tensorflow完整代码
各位同学好,今天和大家分享一下如何使用 Tensorflow 复现谷歌轻量化神经网络 MnasNet 通常而言,移动端(手机)和终端(安防监控.无人驾驶)上的设备计算能力有限,无法搭载庞大的神经网络 ...
【神经网络】(13) ShuffleNetV2 代码复现，网络解析，附Tensorflow完整代码
各位同学好,今天和大家分享一下如何使用 Tensorflow 复现轻量化神经网络 ShuffleNetV2. 为了能将神经网络模型用于移动端(手机)和终端(安防监控.无人驾驶)的实时计算,通常这些设备 ...
【神经网络】(11) 轻量化网络MobileNetV1代码复现、解析，附Tensorflow完整代码
各位同学好,今天和大家分享一下如何使用 Tensorflow 复现轻量化神经网络模型 MobileNetV1.为了能将神经网络模型用于移动端(手机)和终端(安防监控.无人驾驶)的实时计算,通常这些设备 ...
进阶必备：CNN经典论文代码复现 | 附下载链接
经常会看到类似的广告<面试算法岗,你被要求复现论文了吗?>不好意思,我真的被问过这个问题.当然也不是所有面试官都会问,究其原因,其实也很好理解.企业肯定是希望自己的产品是有竞争力,有卖点的 ...
DL之YoloV3：Yolo V3算法的简介(论文介绍)、各种DL框架代码复现、架构详解、案例应用等配图集合之详细攻略
DL之YoloV3:Yolo V3算法的简介(论文介绍).各种DL框架代码复现.架构详解.案例应用等配图集合之详细攻略目录 Yolo V3算法的简介(论文介绍) 0.YoloV3实验结果 1.Yol ...
时空上下文视觉跟踪（STC）算法的解读与代码复现
时空上下文视觉跟踪(STC)算法的解读与代码复现 zouxy09@qq.com http://blog.csdn.net/zouxy09 本博文主要是关注一篇视觉跟踪的论文.这篇论文是Kaihua Z ...
DSB2017第一名代码复现
DSB2017第一名代码复现训练准备数据集功能快捷键合理的创建标题,有助于目录的生成如何改变文本的样式插入链接与图片如何插入一段漂亮的代码片生成一个适合你的列表创建一个表格设定内容 ...