一種簡單的方法，可通過Python實現在數據流中查找異常值

在上一篇文章中，我解釋了流算法的概念，并給出了許多如何應用流算法的示例。 其中之一是在不保存數據流元素的情況下計算數據流的滾動平均值。 現在，我想擴展這個示例，并在異常值檢測的背景下向您展示另一種流算法的用例。

當我們監視機器的功耗以檢測任何異常行為時，可能會出現類似的問題。 如果我們發現異常值有所增加（異常觀察），則可能表明這臺機器的默認值，可能值得檢查。

定義和示例

離群值可以通過多種方式定義。 在本文中，我們將使用以下定義：

如果數字數據流中的元素與到目前為止所見元素的平均值不在3個標準偏差之內，則該元素被視為離群值。

這需要一個小例子。 假設我們按順序獲得數據3、2、4、3、5、3、2、10、2、3、1。 讓我們進一步假設，我們從零的均值和方差（以及因此的標準差）開始，即，如果不等于零，則始終將第一個元素視為離群值。

因此，將3視為離群值，因為3> 0 3 * 0。 現在，我們根據到目前為止看到的元素（僅是數字3）更新均值和方差。因此，新均值是3，方差是0。

然后我們看到2。我們有2> 3 3 * 0，所以2也被認為是離群值。 這是有道理的，因為到目前為止我們只看到了3，所以其他任何數字都不適合該模式。 平均值更新為（3 2）/2=2.5，方差更新為（（3-2.5）2（2-2.5）2）/2=0.25，這意味著標準偏差為0.5。

現在我們看到4。由于2.5–3 *0.5≤4≤2.53 * 0.5，因此該數字不是異常值（即正常值）。 平均值更新為（3 2 4）/ 3 = 3，方差更新為（（3–3）2（2–3）2（4–3）2）/ 3 = 2/3，因此標準偏差為 約0.81。

以下數字3、5、3、2被認為是正常的。 憑直覺，我們將下面的數字10視為離群值。 讓我們看看該算法的作用。 此時的平均值約為3.1，標準偏差約為1。由于10> 3.1 3 * 1，因此我們希望將10視為離群值。

如果繼續最后三個元素，您將看到它們都是正常的。

問題：要計算平均值和標準偏差，我們必須記住到目前為止看到的所有元素。 如果我們有一個每天輸出成千上萬個元素的系統，那么這不是一個選擇。

救援的流式算法

解決此問題的一種方法是使用流算法，該算法在從數據流中每個被掃描元素之后更新其內部狀態。 內部狀態由到目前為止在任何點看到的所有元素的均值和方差組成，從看到任何元素之前的均值和方差為零開始。 確切地說，在看到數據流的第n個元素之后，令m?為平均值，v?為方差，并附加定義m?=v?= 0。

計算均值

在我有關流算法的文章中，我們看到了如何僅使用舊的均值，正在掃描的最新元素以及到目前為止看到的元素數量來更新均值。 這意味著我們只需要隨時使用這種方法存儲兩個數字，而不是像幼稚的方法那樣存儲n。 讓我再次顯示它，將數據流的第i個傳入元素表示為a?：

這個公式不難開發，對吧？ 有了它，我們就有了我們期望的元素大小的基線。 現在，我們只需要可以用均值圍繞的標準偏差即可將輸入的示例分類為離群值和正常數據點。 我們通過計算方差來做到這一點，然后取其平方根即可達到標準偏差。

計算方差

在這種情況下，我們也可以輕松找到遞歸公式。 首先，看到n個元素后的方差為

讓我們嘗試再次根據n，v?和最新元素來寫。 由于方差取決于均值，因此我們也要包含m?。 在開始之前，讓我們重新安排這個公式，以使事情變得更容易：

現在，目標是使v?進入那里。 一種進行方式是從以下簡單的重新排列開始，以隔離平方和直到索引n，它也以v?中的一項出現：

這相當于

反過來導致

現在，我們有了公式，讓我們看看它在Python中是如何工作的！

用Python實現

我們可以通過以下方式實現上述解釋：

class StreamingMeanAndVariance:

def __init__(self):

self.mean = 0

self.variance = 0

self.n_elements = 0

def update(self, element):

self.variance = ((self.variance + self.mean ** 2) * self.n_elements + element ** 2) / (self.n_elements + 1)

self.mean = ((self.mean * self.n_elements) + element) / (self.n_elements + 1)

self.variance = self.variance - self.mean ** 2

self.n_elements += 1

關于此的注釋：update方法的第一行計算方差，但不減去當前均方根。 在第二行中，計算當前平均值。 在第三行中，然后將其從方差中減去，因為在第一行中仍然缺少此值。

要使用它，我們會

import numpy as np

m = StreamingMeanAndVariance()

n = 10000

for i, s in enumerate(np.random.randn(n)):

if not - 3 <= (s - m.mean) / np.sqrt(m.variance) <= 3:

print(i, s)

m.update(s)

這將掃描數據流，該數據流在此示例中由10000個正態分布的數字組成（我們將其表示為N（0,1）），并在出現異常時打印異常值。

如果將法線的間隔和平均值（以黃色表示）作圖，則會得到以下圖片：

以藍色顯示，您可以看到測量值。綠色區域包含法線點，其外部的測量值（紅色表示）被認為是異常值。以黃色顯示您的期望值（平均值）。

討論區

該算法可以達到我們的期望！ 但是，到目前為止，我們還沒有看到它如何處理分配的變化，而是始終只有標準的正態分布數。

讓我們測試執行以下操作時算法的行為：

結果看起來像這樣：

Everything adapts slowly to the new data.

優點

這看起來很有希望！ 一切都會自動適應新數據。 當數據的平均值從0變為2時，我們可以看到很多離群值，這是有道理的。 新平均值2越多，觀測到的異常值就越少，因為2左右是新的常態。

當將平均值從2更改為-2時，我們可以看到更多的離群值，因為這種變化要嚴重得多。 到目前為止，一切都很好。

缺點

如果查看圖的右半部分，可以看到對新數據的適應非常慢。 如您所見，平均值和標準偏差將在一段時間后再次達到正確的水平，因為黃線（均值）下降并且綠色區域再次變窄。 但是直到穩定為止，沒有發現異常值。

為了解決這個問題，我們只能使用最后k個樣本來計算均值和標準差，因為

這會破壞第一次測量的影響。 如果將k設置為無窮大，則可以從之前獲得算法。

我們將k設置得越低，算法就越快地適應新數據。 但是，將k設置得太小可能會導致丟棄異常值，因為該算法認為新數據就是這樣。 在設置k = 1的極端情況下，沒有元素被視為離群值，因為僅考慮了最新元素。

根據用例，可能幾百或幾千個就可以了。

結論

在本文中，我們已經看到了如何為數據流建立一個非常簡單的異常檢測機制。 我們的算法不需要存儲所有測量值，因此非常容易應用，也可以在極其受限制的硬件上使用，并且只需固定存儲即可。 該算法甚至可以適應數據更改，因此無需手動更新。

唯一需要調整的是自適應率，我們在本文中沒有介紹，但是很容易做到。