Python异常值检测与处理函数库

def detect_st_threshold(SLdata_list, sensor_id):
    '''
    根据给定的索力数据列表和传感器ID，检测超过阈值的异常数据索引。

    Args:
        SLdata_list: 索力数据列表。
        sensor_id: 传感器ID。

    Returns:
        异常数据索引列表。
    '''
    # 读取程序外部数据文件获得索力经验值字典
    SL_LIM_file_path = r'suoli_lim.csv'
    SL_LIM = pd.read_csv(SL_LIM_file_path)
    column_name = 'SNAME'
    SL_limit_dict = SL_LIM.set_index(column_name).to_dict(orient='index')
    error_index_list = []

    # 读取外部文档可进行实时修改模式
    upper_limit = SL_limit_dict[sensor_id]['MAXDATA']
    lower_limit = SL_limit_dict[sensor_id]['MINDATA']
    
    for i in range(len(SLdata_list)):
        if SLdata_list[i] > upper_limit or SLdata_list[i] < lower_limit:
            error_index_list.append(i)
    return error_index_list


def fill_point_B_spline(SLdata_list, error_index_list):
    '''
    使用B样条曲线插值的方法填充异常数据点。

    Args:
        SLdata_list: 索力数据列表。
        error_index_list: 异常数据索引列表。

    Returns:
        填充后的新索力数据列表。
    '''
    new_list = []
    num_list, SLdata_list_new = throw_outlier_ori(SLdata_list, error_index_list)
    fx = interp1d(num_list, SLdata_list_new, kind='cubic', fill_value='extrapolate') 

    for i in range(len(SLdata_list)):
        if i in error_index_list:
            if i >= num_list[0] and i <= num_list[-1]:
                new_list.append(fx(i))
            else:
                new_list.append(3186)
        else:
            new_list.append(SLdata_list[i])
    return new_list


def throw_outlier(data_list, error_index_list):
    '''
    去除异常数据点。

    Args:
        data_list: 数据列表。
        error_index_list: 异常数据索引列表。

    Returns:
        去除异常后的新数据列表。
    '''
    new_data_list = [data for i, data in enumerate(data_list) if i not in error_index_list]
    return new_data_list


def throw_outlier_ori(SLdata_list, error_index_list):
    '''
    去除异常数据点，并返回去除异常后的索引列表和新索力数据列表。

    Args:
        SLdata_list: 索力数据列表。
        error_index_list: 异常数据索引列表。

    Returns:
        去除异常后的索引列表和新索力数据列表。
    '''
    new_SLdata_list = []
    num_LIST = []
    for i in range(len(SLdata_list)):
        if i in error_index_list:
            pass
        else:
            num_LIST.append(i)
            new_SLdata_list.append(SLdata_list[i])
    return num_LIST, new_SLdata_list


def detect_sb_2order(SLdata_list, num):
    '''
    检测二阶差分异常点的索引。

    Args:
        SLdata_list: 索力数据列表。
        num: 数据间隔数。

    Returns:
        异常索引列表。
    '''
    error_index_list = []
    for i in range(num):
        diff_l = peak_find(SLdata_list, i * 100 - 1, 0)
        diff_r = peak_find(SLdata_list, i * 100 - 1, 1)
        if diff_l != None and diff_r != None:
            if diff_l * diff_r < 0 and abs(diff_r * diff_r) > 400:
                error_index_list.append(i * 100 - 1)
    return error_index_list


def peak_find(SLdata_list, point, direction):
    '''
    查找峰值点。

    Args:
        SLdata_list: 索力数据列表。
        point: 数据点。
        direction: 方向。

    Returns:
        峰值差分值。
    '''
    close_point = -1 if direction == 0 else 1
    cnt = -2 if direction == 0 else 2
    if point + close_point <= len(SLdata_list) - 1 and point + close_point >= 0:
        diff = SLdata_list[point + close_point] - SLdata_list[point]
        while(1):
            if point + cnt <= len(SLdata_list) - 1 and point + cnt >= 0:
                temp = SLdata_list[point + cnt] - SLdata_list[point + cnt - close_point]
                cnt += close_point
                if diff * temp > 0:
                    diff += temp
                else:
                    break
            else:
                break
        return diff
    return None


def detect_2n(SLdata_list):
    '''
    检测满足条件的2n次幂点的索引。

    Args:
        SLdata_list: 索力数据列表。

    Returns:
        异常索引列表。
    '''
    error_index = []
    for i in range(len(SLdata_list)):
        if i == 0 or i == len(SLdata_list) - 1: pass
        else:
            if SLdata_list[i + 1] < -7 and SLdata_list[i - 1] > 7: 
                error_index.append(i)
            if SLdata_list[i - 1] > 7 and SLdata_list[i - 1] < -7:
                error_index.append(i)
    return error_index


def one_order(SLdata_list):
    '''
    计算一阶差分值。

    Args:
        SLdata_list: 索力数据列表。

    Returns:
        差分值列表。
    '''
    new_list = []
    for i in range(len(SLdata_list) - 3):
        s1 = SLdata_list[i + 1] - SLdata_list[i]
        s2 = SLdata_list[i + 2] - SLdata_list[i]
        s3 = SLdata_list[i + 3] - SLdata_list[i]
        if s1 > 0 and s2 - s1 > 0 and s3 - s2 > 0:   new_list.append(max(s1, s2, s3))
        elif s1 < 0 and s2 - s1 < 0 and s3 - s2< 0: new_list.append(min(s1, s2, s3))
        else: new_list.append(s1)

    return new_list


def twice_order(SLdata_list):
    '''
    计算两次一阶差分值

    Args:
        SLdata_list: 索力数据列表

    Returns:
        差分值列表
    '''
    new_list = one_order(SLdata_list)
    newnew_list = []
    newnew_list = one_order(new_list)
    return newnew_list


class detect_error(object):
    '''
    包含一些检测异常的方法。
    '''
    def get_accuracy(self, SLdata_list):
        '''
        计算准确率。(未实现)
        '''
        pass

    def get_false_alarm(self, SLdata_list, SLMdate_list, error_index_list, month, senor_id):
        '''
        计算误报率。

        Args:
            SLdata_list: 索力数据列表。
            SLMdate_list: 索力数据对应的时间列表。
            error_index_list: 异常数据索引列表。
            month: 月份。
            senor_id: 传感器ID。

        Returns:
            真阳性数量、假阳性数量和假阴性数量。
        '''
        false_alarm_cnt = 0
        true_cnt = 0
        new_outlier_list = []
        # new_SLdata_list = []
        for data in outlier_dict[month][senor_id]['SLdata']:
            new_outlier_list.append(int(data))
        for error_index in error_index_list:
            if int(SLdata_list[error_index]) in new_outlier_list:
               dict_index = new_outlier_list.index(int(SLdata_list[error_index]))
               
               if SLMdate_list[error_index].month == outlier_dict[month][senor_id]['Mdate'][dict_index].month and\
                  SLMdate_list[error_index].day == outlier_dict[month][senor_id]['Mdate'][dict_index].day:
                    true_cnt += 1
                    # print(SLdata_list[error_index], SLMdate_list[error_index])
            else:
                false_alarm_cnt += 1
        print(outlier_dict[month][senor_id]['SLdata'], outlier_dict[month][senor_id]['Mdate'])
        # return false_alarm_cnt/len(outlier_dict[month][senor_id]['SLdata'])
        return true_cnt, false_alarm_cnt, len(outlier_dict[month][senor_id]['SLdata']) - true_cnt



def fill_point(data_list, error_index_list):
    '''
    使用插值的方法填充异常数据点。

    Args:
        data_list: 数据列表。
        error_index_list: 异常数据索引列表

    Returns:
        填充后的新数据列表
    '''
    new_list = []
    for data in data_list:
        if data_list.index(data) in error_index_list:
            if data_list.index(data) == 0:
                new_list.append(data_list[1])
            elif data_list.index(data) == len(data_list) - 1:
                new_list.append(data_list[len(data_list) - 2])
            elif data_list.index(data) - 1 not in error_index_list and data_list.index(data) + 1 not in error_index_list:
                new_list.append((data_list[data_list.index(data) - 1] + data_list[data_list.index(data) + 1])/2)
            else:
                cnt = 1
                while(1):
                    if data_list.index(data) - 1 - cnt not in error_index_list:
                        new_list.append((data_list[data_list.index(data) - 1 - cnt]))
                        break
                    else:
                        cnt += 1
        else:
            new_list.append(data)
    return new_list


class k_means():    
    '''
    包含一些使用K-means算法进行异常检测的方法。
    '''
    def k_choose(self ,data_list):
        '''
        选择最佳K值。(未使用)
        '''
        SSE = []
        for k in range(1,9):
            k_tmp = KMeans(n_clusters=k)
            k_tmp.fit(np.array(data_list).reshape(-1, 1))
            SSE.append(k_tmp.inertia_) 
        return SSE
    

    def k_means_train(self, k, data_list):
        '''
        使用K-means算法进行异常检测。

        Args:
            k: 聚类数量。
            data_list: 数据列表。

        Returns:
            异常数据索引列表。
        '''
        label_cnt = []
        error_index = []
        K_means_ = KMeans(n_clusters=k)
        K_means_.fit(np.array(data_list).reshape(-1, 1))
        data_labels = list(K_means_.labels_)
        for i in range(k):
            label_cnt.append(0)
        for i in range(len(data_labels)):
            label_cnt[data_labels[i]] = label_cnt[data_labels[i]] + 1
        error_label = label_cnt.index(min(label_cnt))
        if label_cnt[error_label] > label_cnt[1 - error_label] * 0.2:
            return []
        else:
            error_label = label_cnt.index(min(label_cnt))
            for i in range(len(data_labels)):
                if data_labels[i] == error_label: error_index.append(i)
            return error_index

limit_dict = {'SLS01': [5600, 6000], 'SLS02': [3450, 3700], 'SLS03': [3000, 3400], 'SLS04': [2850, 3300], 'SLS05': [2750, 3000], 'SLS06': [2800, 3200], 'SLS07': [3400, 3800], 'SLS08': [5600, 6000], 'SLS09': [6930, 6930], 'SLS10': [4100, 4400], 'SLS11': [3650, 3900], 'SLS12': [3350, 3650], 'SLS13': [3350, 3650], 'SLS14': [3500, 3900], 'SLS15': [4000, 4400], 'SLS16': [6500, 7000], 'SLS17': [6300, 6700], 'SLS18': [3650, 3950], 'SLS19': [3200, 3500], 'SLS20': [3100, 3500], 'SLS21': [3250, 3550], 'SLS22': [3300, 3600], 'SLS23': [3850, 4100], 'SLS24': [6050, 6500], 'SLX01': [5600, 6000], 'SLX02': [3400, 3700], 'SLX03': [3000, 3400], 'SLX04': [2900, 3300], 'SLX05': [2750, 3050], 'SLX06': [2850, 3150], 'SLX07': [3400, 3800], 'SLX08': [5600, 6000], 'SLX09': [6700, 7200], 'SLX10': [4100, 4400], 'SLX11': [3600, 3900], 'SLX12': [3300, 3700], 'SLX13': [3300, 3700], 'SLX14': [3500, 3900], 'SLX15': [4000, 4400], 'SLX16': [6500, 7000], 'SLX17': [6300, 6700], 'SLX18': [3650, 3900], 'SLX19': [3200, 3500], 'SLX20': [3200, 3450], 'SLX21': [3250, 3550], 'SLX22': [3300, 3600], 'SLX23': [3800, 4100], 'SLX24': [6100, 6600]}

def set_scale(func):
    '''
    装饰器函数，用于根据阈值范围对异常检测结果进行筛选。
    '''
    def inner(*args, **kwargs):
        upper_limit = limit_dict[args[4]][1]
        lower_limit = limit_dict[args[4]][0]
        error_index_list = func(*args, **kwargs)
        new_error = []
        for error_index in error_index_list:
            if args[1][error_index] >= lower_limit and \
               args[1][error_index] <= upper_limit:
                pass
            else: new_error.append(error_index)
        return new_error
    return inner


class three_sigma(object):
    '''
    包含一些使用3σ原则进行异常检测的方法。
    '''
    def __init__(self):
        pass

    # @set_scale
    def three_sigma_(self, SLdata_list, SLdata_mean, SLdata_std, sensor_id):
        '''
        使用3σ原则进行异常检测。

        Args:
            SLdata_list: 索力数据列表。
            SLdata_mean: 索力数据平均值。
            SLdata_std: 索力数据标准差。
            sensor_id: 传感器ID。

        Returns:
            异常数据索引列表。
        '''
        error_index = []
        if SLdata_std > SLdata_mean/10:
            for i in range(len(SLdata_list)):
                if abs(SLdata_list[i] - SLdata_mean) > 1.5 * SLdata_std:
                    error_index.append(i)
        else:
            for i in range(len(SLdata_list)):
                if abs(SLdata_list[i] - SLdata_mean) > 3 * SLdata_std:
                    error_index.append(i)

        return error_index


    def three_sigma_strict(self, list):
        '''
        使用严格的3σ原则进行异常检测。

        Args:
            list: 数据列表。

        Returns:
            异常数据索引列表。
        '''
        error_index = []
        list_mean = np.mean(list)
        list_std = np.std(list)
        threshold = 3 * list_std
        for i in range(len(list)):
            if list_mean - threshold < list[i] < list_mean + threshold:
                pass
            else:
                error_index.append(i)
        return error_index
    

    

def return_index(func):
    '''
    装饰器函数，用于将异常检测结果转换为异常数据索引列表。
    '''
    def wrapper(*args, **kwargs):
        label_ = func(*args, **kwargs)
        error_index = []
        for i in range(len(label_)):
            if label_[i] == -1 : error_index.append(i)
        return error_index
    return wrapper


def return_lrweights(func):
    '''
    装饰器函数，用于计算左右权重。(未使用)
    '''
    def wrapper(self, number):
        close_num = func(self, number)
        rn = close_num if close_num > number else close_num + 1
        rw = 1 - rn + number
        lw = 1 - rw
        return lw, rw
    return wrapper


class iostation_forest():
    '''
    包含使用孤立森林算法进行异常检测的方法。
    '''
    @return_index
    def ioslation_forest_train(self, SLdata_list):
        '''
        使用孤立森林算法进行异常检测。

        Args:
            SLdata_list: 索力数据列表。

        Returns:
            异常数据索引列表。
        '''
        isof = IsolationForest(n_estimators=100, max_samples='auto', contamination=float(0.05),max_features=1.0)
        isof.fit(np.array(SLdata_list).reshape(-1, 1))
        SLdata_label = isof.predict(np.array(SLdata_list).reshape(-1, 1))
        return SLdata_label


class box_plot():
    '''
    包含使用箱线图进行异常检测的方法。
    '''
    def box_plot_train(self, data_list):
        '''
        使用箱线图进行异常检测。

        Args:
            data_list: 数据列表

        Returns:
            异常数据索引列表
        '''
        error_index_list = []
        sorted_list = sorted(data_list)
        num = len(sorted_list)
        q1_pos = (num + 1) / 4 - 1
        q2_pos = (num + 1) / 2 - 1
        q3_pos = 3 * (num + 1)/ 4 - 1
        q1 = self.get_q(q1_pos, sorted_list)
        q2 = self.get_q(q2_pos ,sorted_list)
        q3 = self.get_q(q3_pos ,sorted_list)
        IQR = q3 - q1
        bottom_line = q1 - 1.5 * IQR
        top_line = q3 + 1.5 * IQR
        for data in data_list:
            if data < bottom_line or data > top_line : error_index_list.append(data_list.index(data))
        return error_index_list
    

    def get_q(self, q_pos, data_list):
        '''
        获取分位数的值

        Args:
            q_pos: 分位数的位置
            data_list: 数据列表

        Returns:
            分位数的值
        '''
        q_lw, q_rw = self.left_or_right(q_pos)
        return q_lw * data_list[floor(q_pos)] + q_rw * data_list[ceil(q_pos)]


    @return_lrweights
    def left_or_right(self, number):
        '''
        判断一个数是更靠近左边还是右边。(未使用)
        '''
        left_diff = abs(number - ceil(number))
        right_diff = abs(floor(number) - number)
        return ceil(number) if left_diff <= right_diff else floor(number)
    
class svm_():
    '''
    包含使用支持向量机进行异常检测的方法。
    '''
    @return_index
    def svm_train_(self, SLdata_list):
        '''
        使用支持向量机进行异常检测。

        Args:
            SLdata_list: 索力数据列表

        Returns:
            异常数据索引列表
        '''
        clf = svm.OneClassSVM(nu=0.001, kernel='rbf', gamma=0.0001)
        clf.fit(np.array(SLdata_list).reshape(-1, 1))
        error_label = clf.predict(np.array(SLdata_list).reshape(-1, 1))
        return error_label

    def svm_train_test(self, SLdata_list):
        '''
        使用支持向量机进行异常检测(测试用)。

        Args:
            SLdata_list: 索力数据列表

        Returns:
            异常数据标签列表
        '''
        clf = svm.OneClassSVM(nu=0.001, kernel='rbf', gamma=0.0001)
        clf.fit(np.array(SLdata_list).reshape(-1, 1))
        error_label = clf.predict(np.array(SLdata_list).reshape(-1, 1))
        return error_label

if __name__ == '__main__':
    pass