人工智能 大数据 计算机科学

异常值检测-滑动均值实现智能告警

2022-04-13  本文已影响0人  万州客

疯狂学习起来吧，metis里用了EWMA的。这篇讲得比较细致。

URL：

https://blog.csdn.net/xsdxs/article/details/71608157

前言

当前的分析对象是一段 timestamp-value 的时间序列，该时间序列可能是cpu使用率、磁盘使用率等数据。对于这些序列实现智能告警功能（也就是检测出一段序列中的异常值），因此需要我们在计算前，首先要定义什么样的值是异常值。
  基于移动平均的方法，其朴素思想是在直观上来看图形，认为相近一段时间内的数据值，有相似的走向趋势。因此判断一个值是否是异常值，可通过判断该值是否对数据趋势造成了破坏，以此来得出结论。
  如下图所示，其中某一点明显波动异常。所以直观上来看，就可以判断出该点是异常点。

那么如何通过某些方法，对异常点进行更加科学地判断以及代码实现。即本文主要工作内容。

移动平均方法介绍

简单移动平均

简单移动平均线(Simple Moving Average，SMA)，是某变量之前 n 个数值的未作加权的算术平均。若计算收盘价的10日 SMA，是指“之前10日”收市价的平均值。假设，收市价为 p=(p1, p2, p3, p4, p5, p6, p7, p8, p9, p10)，则 SMA 公式为：

推广到一般情况为：

加权移动平均

加权移动平均(Weighted Moving Average，WMA)，是指计算平均值时将各个数据分别乘以不同权重，其基本思想是赋予近期数据更高的行为权重。WMA是线性的。
  若计算n日WMA，假设收市价为 p= (p1, p2, … , pn, pn+1, … , pm, …)，则为若干天的收市价及依倒数顺序递减一的不同比例，来加权计算的移动平均线，公式如下：

指数加权移动平均

指数加权移动平均(EWMA:exponentially weighted moving average)，公式如下：

x是实际值，s是ewma计算出来的平均值。也就是下一点的平均值是由上一点的平均值，加上当前点的实际值修正而来。这个修正的比例，就取决月这个alpha的decay factor的大小。视觉上来说就是ewma曲线是否紧跟实际曲线，也就是平滑程度。
  要避免由于简单移动平均导致的缺陷，最简单的方法是对近期的数据赋予更高的权重。这也是指数加权移动平均法（EWMA）背后的基本思想

Python 实现

python pandas包中，对于ewma和sma有现成的实现方法。但是对于wma没有找到实现，在此，自己写了个简单地wma实现函数。

# coding:utf-8
from __future__ import division
import pandas as pd
import matplotlib.pyplot as plt


class WMA(object):
    """
    加权移动平均实现
    """
    @staticmethod
    def get_wma_weights(span, flag=True):
        """
        计算每个数值点的wma权重值
        """
        paras = range(1, span + 1)
        count = sum(paras)
        if flag:
            return [float(para) / count for para in paras]
        else:
            return [float(para) / count for para in paras][::-1]

    def get_wma_values(self, datas):
        """
        计算wma数值
        """
        wma_values = []
        wma_keys = datas.index
        for length in range(1, len(datas) + 1):
            wma_value = 0
            weights = self.get_wma_weights(length)
            for index, weight in zip(datas.index, weights):
                wma_value += datas[index] * weight
            wma_values.append(wma_value)
        return pd.Series(wma_values, wma_keys)

'''
1. 首先构造测试数据
'''
# TODO 以后改成读取数据,直接生成 pandas Series 对象
data_dict_series = {
    '1490707920': 19.8219660272,
    '1490707980': 20.0681534509,
    '1490708040': 20.1842385903,
    '1490708100': 19.7650368611,
    '1490708160': 19.9269200861,
    '1490708220': 18.470530654,
    '1490708280': 18.2211077462,
    '1490708340': 19.3140179366,
    '1490708400': 20.5632611228,
    '1490708460': 20.6341463476,
    '1490708520 ': 8.9789650937,
    '1490708580': 19.3494873891,
    '1490708640': 20.0160623292,
    '1490708700': 19.6702364186,
    '1490708760': 19.624609674,
    '1490708820': 20.456564799,
    '1490708880': 17.6035744548,
    '1490708940': 17.1818237007,
    '1490709000': 20.600406309,
    '1490709060': 19.9717502842,
    '1490709120': 20.0490164061,
    '1490709180': 20.1280070096,
    '1490709240': 20.579920421,
    '1490709300': 19.0682847972,
    '1490709360': 20.6150588592,
    '1490709420': 19.8059894244,
    '1490709480': 19.7459333887,
    '1490709540': 20.1206678947,
    '1490709600': 21.0094336718,
    '1490709660': 19.8375952393
}
dps = pd.Series(data_dict_series)


'''
2. ewma进行拟合
'''
ewma_line = pd.ewma(dps, span=4)


'''
3. 简单移动平均
'''
sma_line = pd.rolling_mean(dps, window=4)


'''
4. wma加权移动平均
'''
wma_line = WMA().get_wma_values(dps)


'''
5. 对原始数据和拟合结果数据进行画图
'''
# plt.figure(figsize=(10, 4))

dps.plot(label='sample data')
ewma_line.plot(label='sample data EMA result')
sma_line.plot(label='sample data SMA result')
wma_line.plot(label='sample data EWMA result')
plt.legend()
plt.show()

结果展示

总体平滑效果对比


20170511115002922.png

异常点检测

上面通过滑动平均算法，得到了滑动均值，通过均值可计算方差，方差乘以一定的倍数可以得出对于振幅的容忍范围。比较实际的值是否超出了这个范围就可以知道是否可以告警了。
  在本文的示例中，告警的正常的浮动范围设为一倍的方差。可以正常浮动范围可以根据实际业务进行调整。对于异常点监测通过如下代码实现。

# coding:utf-8
from __future__ import division
import pandas as pd


class WMA(object):
    """
    加权移动平均实现
    """

    @staticmethod
    def get_wma_weights(span, flag=True):
        """
        计算每个数值点的wma权重值
        """
        paras = range(1, span + 1)
        count = sum(paras)
        if flag:
            return [float(para) / count for para in paras]
        else:
            return [float(para) / count for para in paras][::-1]

    def get_wma_values(self, datas):
        """
        计算wma数值
        """
        wma_values = []
        wma_keys = datas.index
        for length in range(1, len(datas) + 1):
            wma_value = 0
            weights = self.get_wma_weights(length)
            for index, weight in zip(datas.index, weights):
                wma_value += datas[index] * weight
            wma_values.append(wma_value)
        return pd.Series(wma_values, wma_keys)


def calculate_variance(dps, moving_average):
    variance = 0
    flag_list = moving_average.isnull()
    count = 0
    for index in range(len(dps)):
        if flag_list[index]:
            count += 1
            continue
        variance += (dps[index] - moving_average[index]) ** 2
    variance /= (len(dps) - count)
    return variance


'''
1. 首先构造测试数据
'''
# TODO 以后改成读取数据,直接生成 pandas Series 对象
data_dict_series = {
    '1490707920': 19.8219660272,
    '1490707980': 20.0681534509,
    '1490708040': 20.1842385903,
    '1490708100': 19.7650368611,
    '1490708160': 19.9269200861,
    '1490708220': 18.470530654,
    '1490708280': 18.2211077462,
    '1490708340': 19.3140179366,
    '1490708400': 20.5632611228,
    '1490708460': 20.6341463476,
    '1490708520 ': 8.9789650937,
    '1490708580': 19.3494873891,
    '1490708640': 20.0160623292,
    '1490708700': 19.6702364186,
    '1490708760': 19.624609674,
    '1490708820': 20.456564799,
    '1490708880': 17.6035744548,
    '1490708940': 17.1818237007,
    '1490709000': 20.600406309,
    '1490709060': 19.9717502842,
    '1490709120': 20.0490164061,
    '1490709180': 20.1280070096,
    '1490709240': 20.579920421,
    '1490709300': 19.0682847972,
    '1490709360': 20.6150588592,
    '1490709420': 19.8059894244,
    '1490709480': 19.7459333887,
    '1490709540': 20.1206678947,
    '1490709600': 21.0094336718,
    '1490709660': 19.8375952393
}
dps = pd.Series(data_dict_series)

'''
2. ewma进行拟合
'''
ewma_line = pd.ewma(dps, span=4)

'''
3. 简单移动平均
'''
sma_line = pd.rolling_mean(dps, window=4)

'''
4. wma加权移动平均
'''
wma_line = WMA().get_wma_values(dps)

'''
5. 计算告警
'''
# 计算每种平滑下原始数据的标准差

sma_var = calculate_variance(dps, sma_line)
wma_var = calculate_variance(dps, wma_line)
ewma_var = calculate_variance(dps, ewma_line)

flag_list = sma_line.isnull()

for index in sma_line.index:
    if not (sma_line[index] - sma_var <= dps[index] <= sma_line[index] + sma_var):
        if not flag_list[index]:
            print "异常点", dps[index]
print "=================================="
for index in wma_line.index:
    if not (wma_line[index] - wma_var <= dps[index] <= wma_line[index] + wma_var):
        print "异常点", dps[index]
print "=================================="
for index in ewma_line.index:
    if not (ewma_line[index] - ewma_var <= dps[index] <= ewma_line[index] + ewma_var):
        print "异常点", dps[index]

输出结果如下图所示：


20170511175808099.png

小结和参考

小结

本文使用了各种平滑算法对数据进行处理，但是算法见性能和效果，并没有进行科学地度量。这部分工作准备日后进行的。不过目前就从图上来看EWMA效果感觉还是不错的~~over

参考

告警方式介绍：
https://segmentfault.com/a/1190000003021919
滑动平均介绍：
http://www.cnblogs.com/liuning8023/p/3548770.html
https://en.wikipedia.org/wiki/Exponential_smoothing
相关代码实现：
http://stackoverflow.com/questions/9621362/how-do-i-compute-a-weighted-moving-average-using-pandas
http://stackoverflow.com/questions/18517722/weighted-moving-average-in-python
pandas用户手册：
http://pandas.pydata.org/pandas-docs/version/0.17.0/api.html#standard-moving-window-functions
————————————————
版权声明：本文为CSDN博主「宇毅」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/xsdxs/article/details/71608157

上一篇下一篇

猜你喜欢

热点阅读