时间序列异常点检测算法（Smoothed z-score algorithm）

目录

• 前言
• 一、原理
• 二、代码实现
• 总结

前言

时间序列异常点处理！

一、原理

• 算法原理：利用滑动窗口对之后的节点值做一个范围预测（利用平均值和方差），如果真实节点不在这个范围，则表明该节点是异常值。
• 平滑原理：对异常点做一个平滑，以用来评估之后的节点。（这里平滑的值=当前异常节点的百分之五十+上一个被平滑后节点的百分之五十，百分之五十这个值可以调节）
• 缺点：遇到连续的异常点，可能不会检测到第一个异常点之后的节点。
• 建议1：咱这边对于这个缺点，给出的建议是，对异常点的平滑可以做一个修改，可以使用之前几个值的一个平均值或者是中位数来做一个填充，如果咱这个技术过硬，可以采用机器学习预测异常点，嗨嗨嗨，也是极好的。
• 建议2：如果能大概测试到异常点有多少，可以使用之前平均值和跳过几个异常点之后的平均值做差，当作系数乘到这里threshold * stdFilter [i-1]。
• 图示解释：可以看到红色是序列数据，灰色区域是程序可接受的范围值，当红色点超出灰色区域，则在下面标记一个信号量，其取值为-1,0,1.（可以理解为信号量是对程序处理异常点的一个直观体现）
• 与代码的对应：信号量对应程序中的列表signals，方差列表对应stdFilter ，修正后的输入列表对应filteredY

二、代码实现

#!/usr/bin/env python
# Implementation of algorithm from https://stackoverflow.com/a/22640362/6029703
import numpy as np
import pylab

# 源代码地址
# https://stackoverflow.com/questions/22583391/peak-signal-detection-in-realtime-timeseries-data/43512887#43512887
def thresholding_algo(y, lag, threshold, influence):
    signals = np.zeros(len(y))
    filteredY = np.array(y)
    avgFilter = [0]*len(y)
    stdFilter = [0]*len(y)
    avgFilter[lag - 1] = np.mean(y[0:lag])
    stdFilter[lag - 1] = np.std(y[0:lag])
    for i in range(lag, len(y)):
        if abs(y[i] - avgFilter[i-1]) > threshold * stdFilter [i-1]:
            if y[i] > avgFilter[i-1]:
                signals[i] = 1
            else:
                signals[i] = -1

            filteredY[i] = influence * y[i] + (1 - influence) * filteredY[i-1]
            avgFilter[i] = np.mean(filteredY[(i-lag+1):i+1])
            stdFilter[i] = np.std(filteredY[(i-lag+1):i+1])
        else:
            signals[i] = 0
            filteredY[i] = y[i]
            avgFilter[i] = np.mean(filteredY[(i-lag+1):i+1])
            stdFilter[i] = np.std(filteredY[(i-lag+1):i+1])

    return dict(signals = np.asarray(signals),
                avgFilter = np.asarray(avgFilter),
                stdFilter = np.asarray(stdFilter))
                
           

参考文章：

（No.38）时间序列异常点检测算法.

总结

烦、装linux环境装的我心烦意乱！ 装了好久了还没装好，亿点点毛病。