時間序列異常點檢測算法（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環境裝的我心煩意亂！ 裝了好久了還沒裝好，億點點毛病。