[PYTHON] Einführung in die Erkennung von Anomalien und Zusammenfassung der Methoden

――Annormales Verhalten hängt von individuellen Problemen ab, z. B. einer drastischen Zunahme oder Abnahme der beobachteten Werte im selben Abschnitt oder einer deutlich geringeren Änderung.

• Der Hauptunterschied zu statischen Anomalien besteht darin, dass dynamische Anomalien nicht immer anomale Werte aufweisen. ――Zum Beispiel sieht der zentrale Teil des folgenden Diagramms abnormal aus, aber der Wert selbst ist ein möglicher Wert. In diesem Fall finden Sie Anomalien im Verhalten von "Frequenzänderungen".

import numpy as np
import matplotlib.pyplot as plt

x1 = np.concatenate([np.linspace(-10, -3, 50), np.zeros(50), np.linspace(3, 9, 50)])
x2 = np.concatenate([np.zeros(50), np.linspace(-3, 3, 50), np.zeros(50)])
y = np.sin(x1) + np.sin(5*x2)
plt.plot(np.linspace(-10, 9, 150), y)

Als Formel ausgedrückt ist jeder Punkt $ x_j $ $ p (x) = \ int p (x, t) {\ rm d} t $ in den Seriendaten $ \ {x_1, ..., x_t \} $ Selbst wenn es im Sinne von normal ist, ist die bedingte Wahrscheinlichkeit der Zeit $ p (x_j \ mid t) = \ int \ cdots \ int p (x_1, ... x_t \ mid t) {\ rm d} t_1 \ cdots { \ rm d} t_ {j-1} {\ rm d} t_ {j + 1} \ cdots {\ rm d} t_t $ kann abnormal sein.

Selbst in diesem Fall darf der zentrale Teil jedoch nicht abnormal sein. Zum Beispiel ist es möglich, dass der obere Graph das Maschinenbetriebssignal der Fabrik ist und der Betriebsmodus nur im zentralen Teil umgeschaltet wird, so dass er nicht abnormal ist.

Wie oben erwähnt, gibt es verschiedene Aspekte bei Anomalien, aber Referenz [1] fasst die Arten von Anomalien auf folgende Weise zusammen:

出典:[1]

Die Punktanomalie in der oberen linken Abbildung sieht aus wie Single-Shot-Rauschdaten (Ausreißer) aufgrund von Beobachtungsfehlern usw., aber die Gruppenanomalie ähnelt eher Datengruppen, die aus einer anderen Verteilung generiert wurden als Ausreißer. Es wird angenommen, dass beide unterschiedliche Ansätze zur Erkennung von Anomalien erfordern. Wenn Sie die semantische Anomalie unten rechts erreichen, ist es schwierig, mit der klassischen Anomalieerkennungsmethode umzugehen, und es ist ein Deep-Learning-Ansatz erforderlich.

Wie Sie sehen, gibt es keine einheitliche Lösung für "Anomalie", und "Anomalie" muss von Fall zu Fall definiert werden.

Der Algorithmus zur Erkennung von Anomalien ist grundsätzlich unbeaufsichtigtes Lernen

Der Anomalieerkennungsalgorithmus verwendet grundsätzlich unbeaufsichtigtes Lernen anstelle von überwachtem Lernen. Es gibt zwei Gründe:

1. Meistens, weil es nur wenige abnormale Daten gibt
2. Abnormale Daten sind vielfältig und schwer umfassend zu modellieren

ist. Hunderttausende von Anomaliedaten sollten nicht erfasst werden, insbesondere an Orten, an denen Anomalien kritisch sind (medizinisch oder in der Fabrik). Außerdem ist Grund 2 wie zu Beginn erläutert. Es ist einfacher und einfacher, "normal / nicht normal" zu erraten als "normal / abnormal". Letzteres erwähnt keine abnormalen Daten und kann nur durch das Verständnis der Struktur normaler Daten erreicht werden. Unüberwachtes Lernen ist ein Rahmen für das Verständnis der Datenstruktur durch Schätzen von $ p (x) $ aus der Stichprobe $ x_i (i = 1, ..., n) $, sodass die Erkennung von Anomalien grundsätzlich unbeaufsichtigt ist. Sie werden das Lernen nutzen. Es muss jedoch davon ausgegangen werden können, dass die Stichprobe für unbeaufsichtigtes Lernen keine anomalen Daten enthält oder überwiegend klein ist.

Abgesehen davon gibt Referenz [2] Folgendes an und ist mein Favorit:

Eine Passage in Tolstois Roman "Anna Karenina", die besagt: "Glückliche Familien sind alle gleich; jede unglückliche Familie ist auf ihre Weise unglücklich." Ist suggestiv.

Algorithmen zur Erkennung von Anomalien werden grob in vier Typen unterteilt

Dann werde ich den Algorithmus zur Erkennung von Anomalien konkret einführen.

Algorithmen können grob in vier Ansätze eingeteilt werden. Klassifizierungsansatz, probabilistischer Ansatz, Rekonstruktionsansatz bzw. Distanzansatz. Zusätzlich zu diesen vier Klassifikationen finden Sie hier eine Tabelle, in der klassifiziert wird, ob Deep Learning verwendet werden soll oder nicht:

出典：[1]

Jedes Wort wie PCA oder OC-SVM ist der Name des Anomalieerkennungsalgorithmus, der in der Vergangenheit vorgeschlagen wurde. Die Details der einzelnen Algorithmen werden später vorgestellt, daher geben wir hier nur einen Überblick über jeden der vier Ansätze. (Ich selbst verstehe die folgende Tabelle nicht vollständig, daher würde ich mich freuen, wenn Sie mir Ihre Meinung mitteilen könnten.)

Die obige Tabelle wurde unter Bezugnahme auf die folgenden Materialien erstellt.

1. Hido: Einführung in Jubatus Casual Talks # 2 Abnormality Detection
2. PANG, et al. Deep Learning for Anomaly Detection: A Review

** §2 Einführung und Experiment des Algorithmus **

Zusammenfassung der einzuführenden Methoden

• Classification Approach
  • One-Class SVM
• Probability Approach --T2 Hotelmethode
  • GMM
  • KDE
• Reconstruction Approach
  • AE
• Distance Approach
  • k-NN
  • LOF

Classification Approach One-Class SVM Angenommen, Sie erhalten die Daten $ D = \ {x_1, ..., x_N \} $. Es wird jedoch davon ausgegangen, dass $ D $ keine anomalen Daten enthält oder, falls dies der Fall ist, überwiegend klein ist.

Im vorherigen Kapitel haben wir eingeführt, dass der Klassifizierungsansatz die Grenze zwischen normal und abnormal schätzt. ** One-Class-SVM ist ein Algorithmus, der eine Kugel berechnet und begrenzt, die normale Daten vollständig umgibt. ** (Eine Kugel ist nicht unbedingt dreidimensional.)

Wenn Sie den Radius sehr groß machen, können Sie eine Kugel machen, die $ D $ vollständig enthält. Wenn die Kugel jedoch zu groß ist, ist False Negative 1 und ich bin nicht glücklich. Deshalb,

• Machen Sie das Volumen der Kugel so klein wie möglich
• Lassen Sie einige Daten in $ D $ über die Kugel hinausgehen

Berechnen Sie eine Allzweckkugel, indem Sie zwei Punkte zulassen. Insbesondere kann der optimale Kugelradius $ R $ und das Zentrum $ b $ durch Lösen des folgenden Optimierungsproblems erhalten werden:

python

\min_{R,b,u} \left\{R^2 + C\sum_{n=1}^{N}u_n \right\} \quad {\rm s.t.} \quad \|x_n - b\|^2 \le R^2 + u_n

$ U $ ist jedoch ein Strafbegriff, der sich über die Sphäre hinaus erstreckt, und $ C $ ist ein Parameter dafür, wie wichtig diese Strafe ist. Die nichtlineare Konvertierung von Daten durch Kernelfunktionen ist mit SVM möglich, aber natürlich auch mit One-Class-SVM. Durch die ordnungsgemäße Verwendung der Kernelfunktionen erhalten Sie die richtige Kugel.

Lassen Sie uns ein numerisches Experiment durchführen. Die Daten wurden wie folgt aufbereitet.

python

import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import seaborn as sns

sns.set()
np.random.seed(123)

#Mittelwert und Varianz
mean1 = np.array([2, 2])
mean2 = np.array([-2,-2])
cov = np.array([[1, 0], [0, 1]])

#Generierung normaler Daten(Generiert aus zwei Normalverteilungen)
norm1 = np.random.multivariate_normal(mean1, cov, size=50)
norm2 = np.random.multivariate_normal(mean2, cov, size=50)
norm = np.vstack([norm1, norm2])

#Erzeugung abnormaler Daten(Generiert aus gleichmäßiger Verteilung)
lower, upper = -10, 10
anom = (upper - lower)*np.random.rand(20,20) + lower

s=8
plt.scatter(norm[:,0], norm[:,1], s=s, color='green', label='normal')
plt.scatter(anom[:,0], anom[:,1], s=s, color='red', label='anomaly')
plt.legend()

python

from sklearn import svm
xx, yy = np.meshgrid(np.linspace(-10, 10, 500), np.linspace(-10, 10, 500))

# fit the model
clf = svm.OneClassSVM(nu=0.1, kernel="rbf", gamma=0.1)
clf.fit(norm)

def draw_boundary(norm, anom, clf):
    # plot the line, the points, and the nearest vectors to the plane
    Z = clf.decision_function(np.c_[xx.ravel(), yy.ravel()])
    Z = Z.reshape(xx.shape)

    plt.title("One-Class SVM")
    plt.contourf(xx, yy, Z, levels=np.linspace(Z.min(), Z.max(), 7), cmap=plt.cm.PuBu)
    bd = plt.contour(xx, yy, Z, levels=[0], linewidths=2, colors='darkred')
    plt.contourf(xx, yy, Z, levels=[0, Z.max()], colors='palevioletred')

    nm = plt.scatter(norm[:, 0], norm[:, 1], c='green', s=s)
    am = plt.scatter(anom[:, 0], anom[:, 1], c='red', s=s)

    plt.axis('tight')
    plt.xlim((-10, 10))
    plt.ylim((-10, 10))
    plt.legend([bd.collections[0], nm, am],
               ["learned frontier", "normal", "anomaly"],
               loc="upper left",
               prop=matplotlib.font_manager.FontProperties(size=11))
    plt.show()

draw_boundary(norm, anom, clf)

Übrigens, wenn Sie den Parameter $ C $ kleiner machen, dh die Strafe, die aus der Kugel herausragt, heller machen, ändert sich die Grenze wie folgt. (In sklearn wird es durch den Parameter $ \ nu $ dargestellt, aber es scheint, dass es im Nenner liegt, also scheint es, dass $ \ nu $ kleiner und $ = C $ größer gemacht wird.)

python

# fit the model
clf = svm.OneClassSVM(nu=0.5, kernel="rbf", gamma=0.1)
clf.fit(norm)
draw_boundary(norm, anom, clf)

Weitere Vorschläge sind One-Class NN. In One-Class SVM habe ich nach einer geeigneten nichtlinearen Transformation gesucht, indem ich die Kernelfunktion angepasst habe. Die Idee ist jedoch, sie durch NN zu ersetzen und zu automatisieren. Die Implementierung scheint nervig zu sein, also habe ich es nicht getan.

Probability Approach

Hoteling T2 Methode

Die T2-Methode von Hoteling ist eine klassische Methode und wurde theoretisch analysiert. Zusammenfassend wird ** unter der Annahme einer Normalverteilung ein Algorithmus ** verwendet, der die Eigenschaft verwendet, dass der Anomalie-Score der $ \ chi ^ 2 $ -Verteilung folgt.

Der Einfachheit halber werden wir hier eindimensionale Daten verwenden. Wie üblich, angesichts der Daten $ D $. Hier werden alle Daten $ x $ normal verteilt

python

p(x\mid \mu, \sigma) = \frac{1}{\sqrt{2\pi\sigma^2}}\exp\left\{-\frac{1}{2\sigma^2}(x-\mu)^2\right\}

Es wird angenommen, dass es folgt **. Durch Anpassen dieser Normalverteilung an die Daten nach der wahrscheinlichsten Schätzmethode

python

\begin{align*}
    \hat{\mu} &= \frac{1}{N}\sum_{n=1}^{N}x_n \\
    \hat{\sigma}^2 &= \frac{1}{N}\sum_{n=1}^{N}(x_n - \hat{\mu})^2
\end{align*}

Und jeder Parameter wird geschätzt. Mit anderen Worten, wir haben geschätzt, dass die Daten $ D $ der Normalverteilung $ p (x \ mid \ hat {\ mu}, \ hat {\ sigma} ^ 2) $ folgen.

Lassen Sie uns nun den Anomaly Score einführen. Abnormalität ist eine Statistik, die buchstäblich den Grad der Anomalie angibt, und häufig wird eine negative logarithmische Wahrscheinlichkeit verwendet.

python

prob = np.linspace(0.01, 1, 100)
neg_log_likh = [-np.log(p) for p in prob]

plt.plot(prob, neg_log_likh)
plt.title('Negative Log Likelihood')
plt.ylabel('Anomaly Score')
plt.xlabel('Prob.')

Wie in der Grafik gezeigt, hat die negative Log-Wahrscheinlichkeit eine Form, die einen großen Wert annimmt, wenn die Wahrscheinlichkeit klein ist, und sie scheint als Anomalie zu funktionieren. Wenn Sie die Anomalie der Normalverteilung berechnen, die Sie gerade geschätzt haben

python

-\log p(x\mid \hat{\mu}, \hat{\sigma}^2) = \frac{1}{2\hat{\sigma}^2}(x-\hat{\mu})^2 + \frac{1}{2}\log(2\pi\hat{\sigma}^2)

Es wird sein. Da die Anomalie eine für die Daten $ x $ definierte Größe ist, ist die Anomalie $ a (x) $ ohne die Begriffe, die nicht mit $ x $ zusammenhängen.

python

a(x) = \left( \frac{x - \hat{\mu}}{\hat{\sigma}} \right)^2

Ist berechnet. Das Anomaliemolekül bedeutet: "Je weiter die Daten vom Durchschnitt entfernt sind, desto höher ist die Anomalie." Der Nenner drückt aus, "wie viel Daten gestreut werden?", Und das Gefühl, dass gestreute Daten toleriert werden, auch wenn sie weit vom Durchschnitt entfernt sind, und umgekehrt lassen dichte Daten keine Abweichung zu. Es gibt.

Tatsächlich ist bekannt, dass diese Anomalie $ a (x) $ einer Chi-Quadrat-Verteilung $ \ chi ^ 2_1 $ mit einem Freiheitsgrad von $ 1 $ folgt, wenn die Stichprobengröße $ N $ groß genug ist.

python

from scipy.stats import chi2
x = np.linspace(0, 8, 1000)
for df in [1,2,3,4,5]:
    plt.plot(x, chi2.pdf(x ,df = df), label='deg. of freedom:'+str(df))
plt.legend()
plt.ylim(0, 0.5)
plt.title('Chi-Square Dist.')

Wenn Sie danach den Schwellenwert $ a_ {th} $ auf dem Signifikanzniveau von $ 5 $% festlegen, können Sie Anomalien erkennen. Zusammenfassung,

1. Angenommen, die Daten folgen einer Normalverteilung.
2. Schätzen Sie die Parameter der Normalverteilung $ \ mu $, $ \ sigma $ mit der wahrscheinlichsten Schätzmethode
3. Bestimmen Sie aus der Chi-Quadrat-Verteilung den Punkt $ a_ {th} $, der normal und abnormal trennt.
4. Wenn die neuen Daten $ x_ {new} $ $ a (x_ {new})> a_ {th} $ sind, wird dies als abnormal angesehen.

Ist die T2-Methode des Hotelierens. Dieses Mal habe ich es in einer Dimension erklärt, aber der Grad der Anomalie folgt auch der Chi-Quadrat-Verteilung in mehreren Dimensionen. Der Freiheitsgrad ist jedoch die Anzahl der Dimensionen.

Lass uns experimentieren. Die Daten wurden wie folgt erstellt.

python

norm = np.random.normal(0, 1, 20) #Normale Daten
anom = np.array([-3, 3]) #Abnormale Daten
plt.scatter(norm, [0]*len(norm), color='green', label='normal')
plt.scatter(anom, [0]*len(anom), color='red', label='anomaly')

python

#Höchstwahrscheinlich Schätzung
N = len(norm)
mu_hat = sum(norm)/N
s_hat = sum([(x-mu_hat)**2 for x in norm])/N

#Verwenden Sie den Teilungspunkt der Chi-Quadrat-Verteilung als Schwellenwert
a_th_5 = chi2.ppf(q=0.95, df=1)
a_th_30 = chi2.ppf(q=0.7, df=1)

#Berechnen Sie den Grad der Abnormalität der einzelnen Daten
data = np.hstack([norm, anom])
anom_score = []
for d in data:
    score = ((d - mu_hat)/s_hat)**2
    anom_score.append(score)

#Zeichnung
norm_pred_5 = [d for d,a in zip(data, anom_score) if a<=a_th_5]
anom_pred_5 = [d for d,a in zip(data, anom_score) if a>a_th_5]
norm_pred_30 = [d for d,a in zip(data, anom_score) if a<=a_th_30]
anom_pred_30 = [d for d,a in zip(data, anom_score) if a>a_th_30]

fig, axes = plt.subplots(1,3, figsize=(15,5))

axes[0].scatter(norm, [0]*len(norm), color='green', label='normal')
axes[0].scatter(anom, [0]*len(anom), color='red', label='anomaly')
axes[0].set_title('Truth')
axes[1].scatter(norm_pred_5, [0]*len(norm_pred_5), color='green', label='normal pred')
axes[1].scatter(anom_pred_5, [0]*len(anom_pred_5), color='red', label='anomaly pred')
axes[1].set_title('Threshold:5%')
axes[2].scatter(norm_pred_30, [0]*len(norm_pred_30), color='green', label='normal pred')
axes[2].scatter(anom_pred_30, [0]*len(anom_pred_30), color='red', label='anomaly pred')
axes[2].set_title('Threshold:30%')
plt.legend()

Es ist ersichtlich, dass sich der als anomal angesehene Bereich ausdehnt, wenn die Anomalieschwelle verringert wird. Der Schwellenwert sollte entsprechend dem Problem bestimmt werden. Zum Beispiel sollte der Schwellenwert klein eingestellt werden, wenn das Übersehen von Anomalien kritisch ist, wie im medizinischen Bereich, und kann im Fall der Datenbereinigung etwas höher eingestellt werden.

GMM(Gaussian Mixture Model) Die T2-Methode von Hoteling ging von einer Normalverteilung der Daten aus, in einigen Fällen kann dies jedoch unangemessen sein. Wenn sich die Daten beispielsweise an mehreren Stellen konzentrieren (mehrere Peaks anstelle einzelner Peaks), wie unten gezeigt, entsprechend einer Normalverteilung

python

lower, upper = -3, -1
data1 = (upper - lower)*np.random.rand(20) + lower
lower, upper = 1, 3
data2 = (upper - lower)*np.random.rand(30) + lower
data = np.hstack([data1, data2])

#Höchstwahrscheinlich Schätzung
N = len(data)
mu_hat = sum(data)/N
s_hat = sum([(x-mu_hat)**2 for x in data])/N

#Handlung
from scipy.stats import norm
X = np.linspace(-7, 7, 1000)
Y = norm.pdf(X, mu_hat, s_hat)

plt.scatter(data, [0]*len(data))
plt.plot(X, Y)

Es wird geschätzt, dass der Peak der Normalverteilung an den Ort kommt, an dem keine Daten beobachtet werden. In solchen Fällen ist es besser, anstelle der Normalverteilung eine etwas kompliziertere Verteilung festzulegen. Dieses Mal werden wir als Beispiel die Erkennung von Anomalien ** durch GMM (Gaussian Mixture Model) einführen.

GMM ist eine Verteilung, die einen Multi-Peak-Typ durch Hinzufügen mehrerer Normalverteilungen ausdrückt. Die hinzuzufügende Zahl wird häufig in $ K $ ausgedrückt und als Anzahl der Komponenten bezeichnet. Die GMM-Formeln und Grafiken sind wie folgt (eine Dimension).

python

p(x\mid \pi, \mu, \sigma) = \sum_{k=1}^{K}\pi_k N(x\mid \mu_k, \sigma_k)

Jedoch,

python

0 \le\pi_k\le 1,\quad \sum_{k=1}^{K}\pi_k=1,\quad N(x\mid\mu,\sigma)={\Textnormalverteilung}

ist.

python

#Handlung
from scipy.stats import norm
X = np.linspace(-7, 7, 1000)

Y1 = norm.pdf(X, 1, 1)
Y2 = norm.pdf(X, -3, 2)
Y = 0.5*Y1 + 0.5*Y2 

plt.plot(X, Y1, label='Component1')
plt.plot(X, Y2, label='Component2')
plt.plot(X, Y,  label='GMM')
plt.legend()

Der Mittelwert ($ \ mu ) / Varianz ( \ sigma $) und das Mischungsverhältnis $ \ pi $ jeder Normalverteilung sind Parameter, die aus den Daten gelernt wurden. Die Anzahl der Komponenten $ K $ muss jedoch festgelegt werden. Die Parameter, über die der Analyst auf diese Weise entscheidet, werden als Hyperparameter bezeichnet.

Experimentieren wir mit der Erkennung von Anomalien durch GMM. Beobachten wir auch, wie sich die Form der geschätzten Verteilung in Abhängigkeit von der Anzahl der Komponenten $ K $ ändert. Die Daten verwenden dasselbe wie die Ein-Klassen-SVM:

python

def draw_GMM(norm, anom, clf, K):
    # plot the line, the points, and the nearest vectors to the plane
    xx, yy = np.meshgrid(np.linspace(-10, 10, 1000), np.linspace(-10, 10, 1000))
    Z = np.exp(clf.score_samples(np.c_[xx.ravel(), yy.ravel()]))
    Z = Z.reshape(xx.shape)

    plt.title("GMM:K="+str(K))
    plt.contourf(xx, yy, Z, levels=np.linspace(Z.min(), Z.max(), 7), cmap=plt.cm.PuBu)
    
    nm = plt.scatter(norm[:, 0], norm[:, 1], c='green', s=s)
    am = plt.scatter(anom[:, 0], anom[:, 1], c='red', s=s)

    plt.axis('tight')
    plt.xlim((-10, 10))
    plt.ylim((-10, 10))
    plt.legend([nm, am],
               ["normal", "anomaly"],
               loc="upper left",
               prop=matplotlib.font_manager.FontProperties(size=11))
    plt.show()

python

#Montage durch GMM
from sklearn.mixture import GaussianMixture

for K in [1,2,3,4]:
    clf = GaussianMixture(n_components=K, covariance_type='full')
    clf.fit(norm)
    draw_GMM(norm, anom, clf, K)

Wenn Sie die Anzahl der Komponenten auf diese Weise erhöhen, können Sie sehen, dass sogar die Trends in den Details der Daten erfasst werden. Es ist jedoch vorsichtig, da die Details nicht erfasst werden = die Generalisierungsleistung ist hoch. Die Anzahl der Komponenten kann anhand von Informationsmengenkriterien wie WAIC bestimmt werden (ich habe vorerst versucht, WAIC zu verwenden: GitHub)

KDE(Kernel Density Estimation) Bisher haben wir eine Normalverteilung und eine gemischte Normalverteilung angenommen und interne Parameter durch Anpassung an Daten geschätzt. Ein solcher Ansatz wird als parametrischer Ansatz bezeichnet. Aber was ist, wenn die parametrische Verteilung nicht funktioniert?

Ein nicht parametrischer Ansatz ist in solchen Fällen effektiv, und ** KDE ist eine typische nicht parametrische Schätzmethode **. KDE ist eine Methode zum Schätzen der Wahrscheinlichkeitsverteilung durch Überlagern eines Kernels (normalerweise Gauß) auf jedem Datenpunkt **. Die Visualisierung ist wie folgt.

python

from scipy.stats import norm
np.random.seed(123)
lower, upper = -1, 5
data = (upper - lower)*np.random.rand(3) + lower
for i, d in enumerate(data):
    X = np.linspace(lower,upper,500)
    Y = norm.pdf(X, loc=d, scale=1)
    knl = plt.plot(X,Y,color='green', label='kernel'+str(i+1))

# KDE
n = 3
h = 0.24
Y_pdf = []
for x in X:
    Y = 0
    for d in data:
        Y += norm.pdf((x-d)/h, loc=0, scale=1)
    Y /= n*h
    Y_pdf.append(Y)
kde = plt.plot(X,Y_pdf, label='KDE')
plt.scatter(data, [0]*len(data), marker='x')
plt.ylim([-0.05, 1])
plt.legend()

Auf diese Weise wird die Wahrscheinlichkeitsverteilung durch Überlagerung der Kernel geschätzt. Wenn Sie eine Normalverteilung für den Kernel verwenden, müssen Sie die Verteilung festlegen (Bandbreite genannt). Beobachten wir in nachfolgenden Experimenten auch die Änderung der geschätzten Verteilung, wenn diese Bandbreite geändert wird.

Details der KDE-Theorie und ihrer Stärken und Schwächen sind [Suzuki. Data Analysis 10. "Nichtparametrische Dichteschätzmethode"](http://ibis.tu-tokyo.ac.jp/suzuki/lecture/2015/dataanalysis/L9. pdf) war leicht zu verstehen:

** Geschätzte Verteilung **

python

p(x) = \frac{1}{Nh}\sum_{n=1}^{N}K\left(\frac{x-X_n}{h}\right)

$ H> 0 $ ist jedoch die Bandbreite, $ K () $ ist die Kernelfunktion und $ X_n $ ist jeder Datenpunkt. ** Vorteile Nachteile ** Quelle: Suzuki. Datenanalyse 10. "Nichtparametrische Dichteschätzmethode"

Experimentieren wir mit der Erkennung von Anomalien durch KDE. Daten sind wie gewohnt Verwenden Sie die. Die geschätzte Verteilung beim Ändern der Bandbreite ist wie folgt.

python

def draw_KDE(norm, anom, clf, K):
    # plot the line, the points, and the nearest vectors to the plane
    xx, yy = np.meshgrid(np.linspace(-10, 10, 1000), np.linspace(-10, 10, 1000))
    Z = np.exp(clf.score_samples(np.c_[xx.ravel(), yy.ravel()]))
    Z = Z.reshape(xx.shape)

    plt.title("KDE:BandWidth="+str(bw))
    plt.contourf(xx, yy, Z, levels=np.linspace(Z.min(), Z.max(), 7), cmap=plt.cm.PuBu)
    
    nm = plt.scatter(norm[:, 0], norm[:, 1], c='green', s=s)
    am = plt.scatter(anom[:, 0], anom[:, 1], c='red', s=s)

    plt.axis('tight')
    plt.xlim((-10, 10))
    plt.ylim((-10, 10))
    plt.legend([nm, am],
               ["normal", "anomaly"],
               loc="upper left",
               prop=matplotlib.font_manager.FontProperties(size=11))
    plt.show()

python

from sklearn.neighbors import KernelDensity
for bw in [0.2,0.5,1.0,2.0]:
    clf = KernelDensity(bandwidth=bw, kernel='gaussian')
    clf.fit(norm)
    draw_KDE(norm, anom, clf, bw)

Wie Sie sehen können, wird die geschätzte Verteilung mit zunehmender Bandbreite gleichmäßiger. Da es neben der Normalverteilung noch andere Kernelfunktionen gibt, ist es auch erforderlich, sowohl "Bandbreite als auch Kernelfunktion" entsprechend einzustellen.

Reconstruction Approach Da der Rekonstruktionsansatz häufig für Bilddaten verwendet wird, werden wir auch hier Bilddaten verwenden. Die normalen Daten seien MNIST und die abnormalen Daten Fashion-MNIST.

python

#MNIST laden
import torch
from torch.utils.data import DataLoader
from torchvision import datasets, transforms

#Trainingsdaten lesen
train_mnist = datasets.MNIST(
    root="./data", 
    train=True, 
    transform=transforms.Compose([
               transforms.ToTensor(),
               transforms.Normalize((0.5,), (0.5,)) #Sektion[-1,1]Innerhalb der Standardabweichung= 0.5, 0.Normalisiert auf 5
               ]), 
    download=True
)

train_mnist_loader = torch.utils.data.DataLoader(
    train_mnist, batch_size=64, shuffle=True
)

#Testdaten lesen
test_mnist = datasets.MNIST(
    root="./data", 
    train=False, 
    transform=transforms.Compose([
               transforms.ToTensor(),
               transforms.Normalize((0.5,), (0.5,)) #Sektion[-1,1]Innerhalb der Standardabweichung= 0.5, 0.Normalisiert auf 5
               ]), 
    download=True
)

test_mnist_loader = torch.utils.data.DataLoader(
    test_mnist, batch_size=64, shuffle=True
)

python

# Fashion-MNIST laden
#Trainingsdaten lesen
train_fashion = datasets.FashionMNIST(
    root="./data", 
    train=True, 
    transform=transforms.Compose([
               transforms.ToTensor(),
               transforms.Normalize((0.5,), (0.5,)) #Sektion[-1,1]Innerhalb der Standardabweichung= 0.5, 0.Normalisiert auf 5
               ]), 
    download=True
)

train_fashion_loader = torch.utils.data.DataLoader(
    train_fashion, batch_size=64, shuffle=True
)

#Testdaten lesen
test_fashion = datasets.FashionMNIST(
    root="./data", 
    train=False, 
    transform=transforms.Compose([
               transforms.ToTensor(),
               transforms.Normalize((0.5,), (0.5,)) #Sektion[-1,1]Innerhalb der Standardabweichung= 0.5, 0.Normalisiert auf 5
               ]), 
    download=True
)

test_fashion_loader = torch.utils.data.DataLoader(
    test_fashion, batch_size=64, shuffle=True
)

python

import matplotlib.pyplot as plt

#Eine Funktion, die Bilder gleichzeitig nebeneinander visualisiert
def imshow(image_set, nrows=2, ncols=10):
    plot_num = nrows * ncols
    fig, ax = plt.subplots(ncols=ncols, nrows=nrows, figsize=(15, 15*nrows/ncols))
    ax = ax.ravel()
    for i in range(plot_num):
        ax[i].imshow(image_set[i].reshape(28, 28), cmap='gray')
        ax[i].set_xticks([])
        ax[i].set_yticks([])

python

#MNIST-Beispielvisualisierung
imshow(train_mnist.data)

python

# Fashion-MNIST-Beispielvisualisierung
imshow(train_fashion.data)

AE(Auto Encoder) AE (Auto Encoder) ist eine Art von NN (Neural Network) und ein NN, der lernt, dass Ein- und Ausgang gleich sind. Da es nur mit einer konstanten Funktion langweilig ist, wird normalerweise eine Zwischenschicht mit einer Abmessung eingelegt, die kleiner als die Eingangsabmessung ist.

Dies "fasst" die Eingabedaten in der mittleren Schicht zusammen. Mathematisch bedeutet dies, dass die Eingabedaten auf einen niedrigdimensionalen Raum abgebildet werden, während genügend Informationen erhalten bleiben, um sie zu rekonstruieren.

** Die Erkennung von Anomalien mithilfe von AE basiert auf der Idee, dass "AEs, die für die Rekonstruktion normaler Daten trainiert wurden, abnormale Daten möglicherweise nicht gut rekonstruieren können". ** ** **

Lassen Sie uns nun mit der Erkennung von Anomalien durch AE experimentieren. Nach dem Training von AE mit MNIST werden wir versuchen, die Rekonstruktionsgenauigkeit von MNIST und Fashion-MNIST als Grad der Anomalie zu ermitteln.

python

import torch.nn as nn
import torch.nn.functional as F

#Definition von AE
class AE(nn.Module):
    def __init__(self):
        super(AE, self).__init__()
        self.fc1 = nn.Linear(28*28, 256)
        self.fc2 = nn.Linear(256, 28*28)
        
    def forward(self, x):
        x = torch.tanh(self.fc1(x))
        x = torch.tanh(self.fc2(x))
        return x
ae = AE()

#Definition der Verlustfunktion und Optimierungsmethode
import torch.optim as optim

criterion = nn.MSELoss()
optimizer = optim.Adam(ae.parameters(), lr=0.0001)

python

#Lernschleife
N_train = train_mnist.data.shape[0]
N_test = test_mnist.data.shape[0]

train_loss = []
test_loss = []
for epoch in range(10):  
    # Train step
    running_train_loss = 0.0
    for data in train_mnist_loader:
        inputs, _ = data
        inputs = inputs.view(-1, 28*28)

        # zero the parameter gradients
        optimizer.zero_grad()

        # forward + backward + optimize
        outputs = ae(inputs)
        loss = criterion(outputs, inputs)
        loss.backward()
        optimizer.step()

        running_train_loss += loss.item()
    running_train_loss /= N_train
    train_loss.append(running_train_loss)
    
    # Test step
    with torch.no_grad():
        running_test_loss = 0
        for data in test_mnist_loader:
            inputs, _ = data
            inputs = inputs.view(-1, 28*28)
            outputs = ae(inputs)
            running_test_loss += criterion(outputs, inputs).item()
        running_test_loss /= N_test
        test_loss.append(running_test_loss)
    if (epoch+1)%1==0:    # print every 1 epoch
        print('epoch:{}, train_loss:{}, test_loss:{}'.format(epoch + 1, running_train_loss, running_test_loss))
        
plt.plot(train_loss, label='Train loss')
plt.plot(test_loss, label='Test loss')
plt.ylabel('Loss')
plt.xlabel('epoch')
plt.legend()

python

#Lassen Sie uns den Zustand des Wiederaufbaus visualisieren
with torch.no_grad():
    for data in test_mnist_loader:
        inputs, _ = data
        inputs = inputs.view(-1, 28*28)
        outputs = ae(inputs)

n = 5 # num to viz
fig, axes = plt.subplots(ncols=n, nrows=2, figsize=(15, 15*2/n))
for i in range(n):
    axes[0, i].imshow(inputs[i].reshape(28, 28), cmap='gray')
    axes[1, i].imshow(outputs[i].reshape(28, 28), cmap='gray')
    axes[0, i].set_title('Input')
    axes[1, i].set_title('Reconstruct')
    axes[0, i].set_xticks([])
    axes[0, i].set_yticks([])
    axes[1, i].set_xticks([])
    axes[1, i].set_yticks([])

Es scheint, dass es ordentlich rekonstruiert wurde.

Zeichnen wir nun die Verteilung der Rekonstruktionsgenauigkeit (diesmal quadratischer Fehler) von MNIST und Fashion-MNIST. Wenn die Anomalieerkennung durch AE erfolgreich ist, sollte die Genauigkeit von Fashion-MNIST niedrig sein.

python

#Berechnung der Rekonstruktionsgenauigkeit
mnist_loss = []
with torch.no_grad():
    for data in test_mnist_loader:
        inputs, _ = data
        for data_i in inputs:
            data_i = data_i.view(-1, 28*28)
            outputs = ae(data_i)
            loss = criterion(outputs, data_i)
            mnist_loss.append(loss.item())

fashion_loss = []
with torch.no_grad():
    for data in test_fashion_loader:
        inputs, _ = data
        for data_i in inputs:
            data_i = data_i.view(-1, 28*28)
            outputs = ae(data_i)
            loss = criterion(outputs, data_i)
            fashion_loss.append(loss.item())

python

plt.hist([mnist_loss, fashion_loss], bins=25, label=['MNIST', 'Fashion-MNIST'])
plt.xlabel('Loss')
plt.ylabel('data count')
plt.legend()
plt.show()

Sicherlich scheint die Rekonstruktion von Fashion-MNIST nicht gut zu funktionieren, so dass es scheint, dass die Erkennung von Anomalien durch Rekonstruktionsgenauigkeit hergestellt wird.

Schließlich werde ich diejenigen mit geringem Verlust (= diejenigen, die schwer als abnormal zu beurteilen sind) und diejenigen mit hohem Verlust (= diejenigen, die leicht als abnormal beurteilt werden können) unter Fashion-MNIST aufgreifen und zeichnen.

python

lower_loss, upper_loss = 0.1, 0.8
data_low = [] #Speichert Daten mit geringem Verlust
data_up = []  #Speichert Daten mit hohem Verlust

with torch.no_grad():
    for data in test_fashion_loader:
        inputs, _ = data
        for data_i in inputs:
            data_i = data_i.view(-1, 28*28)
            outputs = ae(data_i)
            loss = criterion(outputs, data_i).item()
            if loss<lower_loss:
                data_low.append([data_i, outputs, loss])
            if loss>upper_loss:
                data_up.append([data_i, outputs, loss])

python

#Index zum zufälligen Zeichnen
np.random.seed(0)

n = 3 #Anzahl der Zeichnungsdaten
lower_indices = np.random.choice(np.arange(len(data_low)), n)
upper_indices = np.random.choice(np.arange(len(data_up)), n)
indices = np.hstack([lower_indices, upper_indices])

fig, axes = plt.subplots(ncols=n*2, nrows=2, figsize=(15, 15*2/(n*2)))

#Zeichnen von Daten mit geringem Verlust
for i in range(n):
    inputs = data_low[indices[i]][0]
    outputs = data_low[indices[i]][1]
    loss = data_low[indices[i]][2]
    axes[0, i].imshow(inputs.reshape(28, 28), cmap='gray')
    axes[1, i].imshow(outputs.reshape(28, 28), cmap='gray')
    axes[0, i].set_title('Input')
    axes[1, i].set_title('Small Loss={:.2f}'.format(loss))
    axes[0, i].set_xticks([])
    axes[0, i].set_yticks([])
    axes[1, i].set_xticks([])
    axes[1, i].set_yticks([])

#Zeichnen von Daten mit großem Verlust
for i in range(n, 2*n):
    inputs = data_up[indices[i]][0]
    outputs = data_up[indices[i]][1]
    loss = data_up[indices[i]][2]
    axes[0, i].imshow(inputs.reshape(28, 28), cmap='gray')
    axes[1, i].imshow(outputs.reshape(28, 28), cmap='gray')
    axes[0, i].set_title('Input')
    axes[1, i].set_title('Large Loss={:.2f}'.format(loss))
    axes[0, i].set_xticks([])
    axes[0, i].set_yticks([])
    axes[1, i].set_xticks([])
    axes[1, i].set_yticks([])
plt.tight_layout()

Bilder mit geringem Verlust weisen tendenziell große schwarze Bereiche auf, während Bilder mit hohem Verlust große weiße Bereiche aufweisen. Da MNIST im Grunde ein Bild mit einem großen schwarzen Bereich ist, ist der weiße Bereich für AE "fremd" und es scheint, dass die Rekonstruktion nicht gut funktioniert.

Distance Approach k-NN(k-Nearest Neighbor) k-NN ist eine Methode zum Messen des Anomaliegrades basierend auf dem Abstand zwischen Daten. Betrachten Sie in k-NN einen Kreis, der k Datenpunkte in der Nähe enthält. Die folgende Abbildung gilt für $ k = 5 $.

Wie in der Abbildung gezeigt, definiert k-NN den Radius des Kreises als den Grad der Anomalie, basierend auf der Idee, dass ** der von den anomalen Daten gezeichnete Kreis größer sein sollte als die normalen Daten. Der Schwellenwert des Radius, der normal / abnormal von der Anzahl benachbarter Punkte $ k $ trennt, muss jedoch entsprechend eingestellt werden.

Lassen Sie uns ein numerisches Experiment durchführen. Der Schwellenwert ist der 90% -Teilungspunkt in der Radiuswertverteilung normaler Daten.

python

#Daten erstellen
np.random.seed(123)

#Mittelwert und Varianz
mean = np.array([2, 2])
cov = np.array([[3, 0], [0, 3]])

#Generierung normaler Daten
norm = np.random.multivariate_normal(mean1, cov, size=50)

#Erzeugung abnormaler Daten(Generiert aus gleichmäßiger Verteilung)
lower, upper = -10, 10
anom = (upper - lower)*np.random.rand(20,20) + lower

s=8
plt.scatter(norm[:,0], norm[:,1], s=s, color='green', label='normal')
plt.scatter(anom[:,0], anom[:,1], s=s, color='red', label='anomaly')
plt.legend()

Ich wusste nicht, wie ich die Anomalieerkennung durch k-NN mit der Bibliothek abschließen sollte, also habe ich es selbst gemacht. .. ..

python

#Eine Funktion, die den Radius des Kreises in der Nähe von k aller Daten berechnet
def kNN(data, k=5):
    """
    input: data:list, k:int
    output:Radius der Kreise in der Nähe von k in allen Daten:list
    
Die Entfernung nimmt die euklidische Entfernung an.
    """
    output = []
    for d in data:
        distances = []
        for p in data:
            distances.append(np.linalg.norm(d - p))
        output.append(sorted(distances)[k])
    return output

#1, wenn größer als der Schwellenwert,Eine Funktion, die 0 zurückgibt, wenn sie klein ist
def kNN_scores(point, data, thres, K):
    distances = []
    for p in data:
        distances.append(np.linalg.norm(point - p))
    dist = sorted(distances)[K]
    if dist > thres:
        return 1
    else:
        return 0

python

def draw_KNN(norm, anom, thres, K, only_norm=False):
    # plot the line, the points, and the nearest vectors to the plane
    xx, yy = np.meshgrid(np.linspace(-10, 10, 100), np.linspace(-10, 10, 100))
    Z = [kNN_scores(point, norm, thres, K) for point in np.c_[xx.ravel(), yy.ravel()]]
    Z = np.array(Z).reshape(xx.shape)

    plt.title("KNN:K="+str(K))
    plt.contourf(xx, yy, Z)
    
    if not only_norm:
        nm = plt.scatter(norm[:, 0], norm[:, 1], c='green', s=s)
        am = plt.scatter(anom[:, 0], anom[:, 1], c='red', s=s)

        plt.axis('tight')
        plt.xlim((-10, 10))
        plt.ylim((-10, 10))
        plt.legend([nm, am],
                   ["normal", "anomaly"],
                   loc="upper left",
                   prop=matplotlib.font_manager.FontProperties(size=11))

    if only_norm:
        nm = plt.scatter(norm[:, 0], norm[:, 1], c='green', s=s)

        plt.axis('tight')
        plt.xlim((-10, 10))
        plt.ylim((-10, 10))
        plt.legend([nm],
                   ["normal"],
                   loc="upper left",
                   prop=matplotlib.font_manager.FontProperties(size=11))
        
    plt.show()

python

for k in [1,5,30]:
    anomaly_scores = kNN(norm, k=k)
    thres = pd.Series(anomaly_scores).quantile(0.90)
    draw_KNN(norm, anom, thres, k)

Sie können sehen, dass mit zunehmendem k die normale / abnormale Grenze glatter wird.

Lassen Sie uns nun mit einem etwas chaotischen Beispiel experimentieren. Angenommen, ** Normaldaten werden aus zwei Normalverteilungen mit unterschiedlichen Dichten ** generiert, wie unten gezeigt.

python

np.random.seed(123)

#Mittelwert und Varianz
mean1 = np.array([3, 3])
mean2 = np.array([-5,-5])
cov1 = np.array([[0.5, 0], [0, 0.5]])
cov2 = np.array([[3, 0], [0, 3]])

#Generierung normaler Daten(Generiert aus zwei Normalverteilungen)
norm1 = np.random.multivariate_normal(mean1, cov1, size=50)
norm2 = np.random.multivariate_normal(mean2, cov2, size=10)
norm = np.vstack([norm1, norm2])

s=8
plt.scatter(norm[:,0], norm[:,1], s=s, color='green', label='normal')
plt.legend()

python

for k in [1,5,30]:
    anomaly_scores = kNN(norm, k=k)
    thres = pd.Series(anomaly_scores).quantile(0.90)
    draw_KNN(norm, [], thres, k, only_norm=True)

Insbesondere wenn man sich auf den Fall von k = 1 konzentriert, kann man sehen, dass kNN nicht gut funktioniert, wenn normale Daten aus verschiedenen Clustern mit unterschiedlichen Dichten erzeugt werden. ** Es ist vorstellbar, dass dies durch Hinzufügen des Prozesses "Wichtig ist die Abweichung vom dichten Cluster und toleriere die Abweichung vom spärlichen Cluster" behandelt werden kann **.

Tatsächlich ist dies die Methode namens LOF, die im nächsten Abschnitt vorgestellt wird.

LOF (lokaler Ausreißerfaktor)

Der Fall, in dem im vorherigen Abschnitt eingeführtes k-NN nicht gut funktioniert, ist in der Abbildung dargestellt.

Es wird sein. Natürlich möchte ich den Grad der Anomalie auf der linken Seite erhöhen, aber LOF führt das Konzept der ** Datendichte ** ein und realisiert es. Ich werde die mathematischen Details weglassen, aber ich mache Folgendes.

Der Grad der Anomalie wird definiert durch ** das Verhältnis des Squishyness-Grades von sich selbst zum Squishyness-Grad benachbarter Punkte **.

Experimentieren wir jetzt mit LOF.

python

def draw_LOF(norm, anom, K, only_norm=False):
    # plot the line, the points, and the nearest vectors to the plane
    xx, yy = np.meshgrid(np.linspace(-10, 10, 100), np.linspace(-10, 10, 100))
    Z = - model.predict(np.c_[xx.ravel(), yy.ravel()])
    Z = np.array(Z).reshape(xx.shape)

    plt.title("LOF:K="+str(K))
    plt.contourf(xx, yy, Z)
    
    if not only_norm:
        nm = plt.scatter(norm[:, 0], norm[:, 1], c='green', s=s)
        am = plt.scatter(anom[:, 0], anom[:, 1], c='red', s=s)

        plt.axis('tight')
        plt.xlim((-10, 10))
        plt.ylim((-10, 10))
        plt.legend([nm, am],
                   ["normal", "anomaly"],
                   loc="upper left",
                   prop=matplotlib.font_manager.FontProperties(size=11))

    if only_norm:
        nm = plt.scatter(norm[:, 0], norm[:, 1], c='green', s=s)

        plt.axis('tight')
        plt.xlim((-10, 10))
        plt.ylim((-10, 10))
        plt.legend([nm],
                   ["normal"],
                   loc="upper left",
                   prop=matplotlib.font_manager.FontProperties(size=11))
        
    plt.show()

python

from sklearn.neighbors import LocalOutlierFactor

for k in [1, 5, 30]:
    model = LocalOutlierFactor(n_neighbors=k, novelty=True)
    model.fit(norm)
    draw_LOF(norm, [], k, only_norm=True)

Auf diese Weise scheint es, dass eine Anomalieerkennung durchgeführt werden kann, nachdem die Dichte des Clusters berücksichtigt wurde.

Problem bei der Erkennung von Anomalien

Da ich mich in Zukunft mit der Erkennung von Anomalien befasse, möchte ich folgende Themen untersuchen:

--Abnormalitätserkennung von Zeitreihendaten --Abnormalitätserkennung von Grafikdaten

• Kann zur Erkennung von Geldwäsche verwendet werden