異常檢測機制

前言

傳統的異常檢測系統通過設置一個固定的閾值來保證監控項處於正常水平，一旦超過設定的閾值，就會觸發報警來提醒人們的注意。

靜態閾值法適用於在一定範圍內波動的監控項，比如磁碟使用率，CPU使用率等，但是如果遇到網路流量這種不具有明顯上限，波動比較劇烈的情況，單純利用靜態閾值法如果設置的閾值比較小，會出現很多誤報的情況，增加人工成本；而如果將閾值設置的比較大，又會出現漏報的情況。

所以我們提出了一個應對複雜場景的異常檢測演算法，它不僅考慮同期數據的情況，也會將周期性納入考慮範圍，通過設置動態閾值的方法來對異常數據進行檢測。

異常檢測演算法基於曲線擬合的檢測

對於時間序列（是指將同一統計指標的數值按其發生的時間先後順序排列而成的數列）來說，T時刻的數值對於T-1時刻有很強的依賴性。比如某個遊樂園人數在8:00很多，在8:01時刻的人數很多的概率是很大的，但是如果07:01時刻人數對於8:01時刻影響不是很大。

針對最近時間窗口內的數據遵循某種趨勢的現象，我們使用一條曲線對該趨勢進行擬合，如果新的數據打破了這種趨勢，使曲線變得不平滑，則該點就出現了異常。

曲線擬合的方法有很多，比如回歸、moving average……。在這篇文章中，我們使用EWMA，即指數權重移動平均方法來擬合曲線。EWMA的遞推公式是：

EWMA(1) = p(1) // 有時也會取前若干值的平均值。α越小時EWMA(1)的取值越重要。 EWMA(i) = α * p(i) + (1-α) * EWMA(i – 1) // α是一個0-1間的小數，稱為smoothing factor.

可以從上面公式得出，下一點的平均值是由上一點的平均值，加上當前點的實際值修正而來。對於每一個EWMA值，每個數據的權重是不一樣的，最近的數據將擁有越高的權重。

有了平均值之後，我們就可以使用3-sigma理論來判斷新的input是否超過了容忍範圍。比較實際的值是否超出了這個範圍就可以知道是否可以告警了。超出了上界，可能是流量突然增加了；低於下界，可能是流量突然降低了，這兩種情況都需要告警。

我們可以使用pandas庫中的ewma函數來實現我們上面的計算過程，代碼如下：

def EWMA(timeseries): flag = "" series = pandas.Series([x[1] for x in timeseries]) expAverage = pandas.stats.moments.ewma(series, com=50) stdDev = pandas.stats.moments.ewmstd(series, com=50) if abs(series.iget(-1) - expAverage.iget(-1)) > 3 * stdDev.iget(-1) and series.iget(-1) - expAverage.iget(-1) > 0: flag = "突增" if abs(series.iget(-1) - expAverage.iget(-1)) > 3 * stdDev.iget(-1) and series.iget(-1) - expAverage.iget(-1) > 0: flag = "突減" return flag

EWMA由於其時效性被廣泛應用在時間序列的預測，它的優勢在於:

可以檢測到一個異常較短時間後發生的另一個(不太高的突變型)異常；

因為它更多的是參考突變之前的點，所以能更快對異常作出反應；

非常敏感，歷史如果波動很小，方差就很小，容忍的波動範圍也會非常小；

而劣勢則是

對漸進型而非突髮型的異常檢測能力較弱；

異常持續一段時間後可能被判定為正常；

業務曲線可能自身有規律性的陡增和陡降；

過於敏感，容易誤報。因為方差會隨著異常點的引入而變大，所以很難使用連續三點才告警這樣的策略；

所以我們需要引入周期性的檢測方法，來針對性處理具有周期性趨勢的曲線。

基於同期數據的檢測

很多監控項都具有一定的周期性，其中以一天為周期的情況比較常見，比如VIP流量在早上4點最低，而在晚上11點最高。為了將監控項的周期性考慮進去，我們選取了某個監控項過去14天的數據。對於某個時刻，將得到14個點可以作為參考值，我們記為xi，其中i=1,...,14。

我們先考慮靜態閾值的方法來判斷input是否異常（突增和突減）。如果input比過去14天同一時刻的最小值乘以一個閾值還小，就會認為該輸入為異常點（突減）；而如果input 比過去14天同一時刻的最大值乘以一個閾值還大，就會認為該輸入為異常點（突增）。我們將上面的計算過程用代碼來表示：

def simultaneous(data, min_threshold, max_threshold): last_time = data.index[-1] last_time = datetime.datetime.strptime(last_time, "%Y-%m-%d %H:%M:%S") simultaneous_data =[] for i in range(1, days_num+1): before_time = last_time + datetime.timedelta(days=int("-%s" %i)) before_time_index = before_time.strftime("%Y-%m-%d %H:%M:%S") if before_time_index in data.keys(): simultaneous_data.append(int(data[before_time_index])) if int(data[-1])

max(simultaneous_data) * max_threshold: return "突增"

靜態閾值的方法是根據歷史經驗得出的值，實際中如何給max_threshold和min_threshold是一個需要討論的話題。根據目前動態閾值的經驗規則來說，取平均值是一個比較好的思路。

上面我們把基於同期數據的檢測方法進行了說明，下面我們討論一下它的優缺點：

優點：

反映出周期性；

可以確保發現大的故障，出了告警一定是大問題；

缺點：

依賴周期性的歷史數據，計算量大，而且無法對新接入的曲線告警；

非常不敏感，小波動無法發現；

基於振幅的時間周期性

檢測方法二遇到下圖的現象就不能檢測出異常。比如今天是11.18日，過去14天的歷史曲線必然會比今天的曲線低很多。那麼今天出了一個小故障，曲線下跌了，相對於過去14天的曲線仍然是高很多的。這樣的故障使用方法二就檢測不出來，那麼我們將如何改進我們的方法呢？一個直覺的說法是，兩個曲線雖然不一樣高，但是「長得差不多」。那麼怎麼利用這種「長得差不多」呢？那就是振幅了。

怎麼計算t時刻的振幅呢？ 我們使用x(t) – x(t-1) 再除以 x(t-1)來表示振幅。舉個例子，例如t時刻的在線900人，t-1時刻的在線是1000人，那麼可以計算出掉線人數是10%。如果參考過去14天的數據，我們會得到14個振幅值。使用14個振幅的絕對值作為標準，如果m時刻的振幅（{m(t) – m(t-1)]/m(t-1)）大於amplitudethreshold並且m時刻的振幅大於0，則我們認為該時刻發生突增，而如果m時刻的振幅大於amplitudethreshold並且m時刻的振幅小於0，則認為該時刻發生突減。

我們將上面的計算過程用一段代碼表示：

def amplitude_max(data, threshold): #最後一個值和倒數第二個值的振幅 last_amplitude = 0.0 last_time = data.index[-1] last_time = datetime.datetime.strptime(last_time, "%Y-%m-%d %H:%M:%S") last_amplitude_time = last_time + datetime.timedelta(minutes=int(-1)) last_amplitude_time_index = last_amplitude_time.strftime("%Y-%m-%d %H:%M:%S") if last_amplitude_time_index in data.keys(): last_amplitude = float((float(data.values[-1])-float(data[last_amplitude_time_index]))/float(data.values[-1])) last_time = last_time + datetime.timedelta(days=int(-1)) #同期的振幅 amplitude_data = [] for i in range(0, days_num): now_time = last_time prior_time = last_time + datetime.timedelta(minutes=int(-1)) prior_time_index = prior_time.strftime("%Y-%m-%d %H:%M:%S") now_time_index = now_time.strftime("%Y-%m-%d %H:%M:%S") if now_time_index in data.keys() and prior_time_index in data.keys(): tmp = float((float(data[now_time_index])-float(data[prior_time_index]))/float(data[now_time_index])) amplitude_data.append(abs(round(tmp, 2))) last_time = last_time + datetime.timedelta(days=int(-1)) if abs(last_amplitude) > max(amplitude_data) * threshold and last_amplitude > 0: return "突增" if abs(last_amplitude) > max(amplitude_data) * threshold and last_amplitude

同樣，我們也討論一下該方法的優缺點，先說優點：

比絕對值要敏感；

利用了時間周期性，規避了業務曲線自身的周期性陡降；

要求原曲線是光滑的；

周期性陡降的時間點必須重合，否則誤警；

按百分比計算容易在低峰時期誤警；

陡降不一定代表故障，由上層服務波動引起的沖高再回落的情況時有發生；

總結

上面介紹了三種檢測的方法，每個方法都有每個方法的優缺點，也有每個方法能檢測和不能檢測的範圍。因此單純依靠一種方法並不能達到我們預期的效果，而且具有很高的誤報率。我們借鑒了投票制度中「少數服從多數」的原則，即三個方法中有兩個或者兩個以上符合，我們才認為符合。

如果你覺得上面的三種方法還不夠精確，你甚至可以使用開源項目skyline來豐富你的演算法庫，進一步提高準確率。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！