[PYTHON] Ich habe versucht, die Erkennung von Anomalien durch spärliches Strukturlernen zu implementieren
- Artikel von Datenwissenschaftlern aus der Fertigungsindustrie
- Dieses Mal haben wir Anomalieerkennungsmethoden implementiert und organisiert, die in der Fertigungsindustrie eingesetzt werden können.
Einführung
In Anlehnung an die vorherige Anomalieerkennungsmethode (MSPC) haben wir diesmal die Anomalieerkennungsmethode durch spärliches Strukturlernen organisiert.
Was ist Anomalieerkennung durch spärliches Strukturlernen?
Die $ T ^ 2 $ -Methode und $ MSPC $ von Hotelling, bekannte Anomalieerkennungsmethoden, werden zur Überwachung von Daten verwendet, deren Durchschnittswert sich nicht ändert. Andererseits werden am Fertigungsstandort nicht nur solche Daten, sondern auch Geräte- und Betriebsdaten mit einer Datenstruktur, deren Werte sich ändern, während eine bestimmte Beziehung zwischen Variablen beibehalten wird. Zu diesem Zeitpunkt ist eine der Anomalieerkennungsmethoden, die sich auf die Beziehungen zwischen Variablen (Korrelation usw.) konzentriert, die Anomalieerkennungsmethode durch Lernen mit geringer Struktur.
Überblick über die Methode
- Um die Korrelation zwischen Variablen zu ermitteln, nehmen Sie eine multivariate Normalverteilung hinter den Daten an und ermitteln Sie die Präzisionsmatrix (die Umkehrung der Kovarianzmatrix).
- Da es schwierig ist, nur aussagekräftige Korrelationen aus tatsächlichen Daten zu extrahieren (Rauscheffekt), verwenden wir dieses Mal die Sparse Inverse Covariance Matrix Optimization.
- Abnormalität ist definiert als die Differenz (KL-Entfernung) in der Wahrscheinlichkeitsverteilung zwischen Normal und Bewertung (Online-Überwachung).
Einzelheiten zur Methode finden Sie unter (Erkennung von Anomalien und Erkennung von Änderungspunkten. ] E7% BF% 92% E3% 83% 97% E3% 83% AD% E3% 83% 95% E3% 82% A7% E3% 83% 83% E3% 82% B7% E3% 83% A7% E3% 83% 8A% E3% 83% AB% E3% 82% B7% E3% 83% AA% E3% 83% BC% E3% 82% BA-% E4% BA% 95% E6% 89% 8B-% E5% Wenn Sie 89% 9B / dp / 4061529080)) lesen, werden Sie Ihr Verständnis vertiefen. Ich habe dieses Buch auch gelesen, studiert und in meiner eigentlichen Arbeit umgesetzt.
Implementierung einer Anomalieerkennungsmethode durch spärliches Strukturlernen
Dieses Mal habe ich für die Probe einen Datensatz namens Water Treatment Plant aus dem UCI Machine Learning Repository verwendet. Wir haben 100 normale Daten und die restlichen 279 als Bewertungsdaten verwendet, um festzustellen, wie stark sie vom normalen Zustand abweichen.
Der Python-Code ist unten.
#Importieren Sie die erforderlichen Bibliotheken
import pandas as pd
import numpy as np
from matplotlib import pyplot as plt
%matplotlib inline
from scipy import stats
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors.kde import KernelDensity
from sklearn.covariance import GraphicalLassoCV
df = pd.read_csv('watertreatment_mod.csv', encoding='shift_jis', header=0, index_col=0)
df.head()
window_size = 10
cr = 0.99
Die Eingabedaten sehen wie oben aus.
Als nächstes werden wir die Trainingsdaten und Bewertungsdaten aufteilen und standardisieren.
#Splitpunkt
split_point = 100
#Unterteilt in Trainingsdaten (Zug) und Bewertungsdaten (Test)
train_df = df.iloc[:(split_point-1),]
test_df = df.iloc[split_point:,]
#Daten standardisieren
sc = StandardScaler()
sc.fit(train_df)
train_df_std = sc.transform(test_df)
test_df_std = sc.transform(test_df)
Als nächstes schätzen wir die Kovarianz.
#Schätzung der Kovarianz
n_samples, n_features = train_df_std.shape
emp_cov = np.dot(train_df_std.T, train_df_std) / n_samples
model = GraphicalLassoCV()
model.fit(train_df_std)
cov_ = model.covariance_
prec_ = model.precision_
Definieren Sie als Nächstes eine Funktion, um den KL-Abstand zu ermitteln.
def Calc_KL(cov_, prec_, xtest):
"""Berechnen Sie den KL-Abstand
Parameters
----------
cov_ : np.array
Kovarianzmatrix der Trainingsdaten
prec_ : np.array
Genauigkeitsmatrix der Trainingsdaten
df : pd.DataFrame
Datensatz
Returns
-------
d_ab : pd.DataFrame
KL Entfernung
"""
n_samples, n_features = xtest.shape
d_abp=np.zeros(n_features)
d_abm=np.zeros(n_features)
d_ab=np.zeros(n_features)
model_test = GraphicalLassoCV()
try:
model_test.fit(xtest)
except FloatingPointError:
print("floating error")
return d_ab
cov__test = model_test.covariance_
prec__test = model_test.precision_
#Berechnen Sie die Größe des Korrelationskollapses für jede Variable
for i in range(n_features):
temp_prec_a = np.r_[prec_[i:n_features,:], prec_[0:i,:]]
temp_prec_a = np.c_[temp_prec_a[:,i:n_features], temp_prec_a[:,0:i]]
temp_prec_b = np.r_[prec__test[i:n_features,:], prec__test[0:i,:]]
temp_prec_b = np.c_[temp_prec_b[:,i:n_features], temp_prec_b[:,0:i]]
temp_cov_a = np.r_[cov_[i:n_features,:], cov_[0:i,:]]
temp_cov_a = np.c_[temp_cov_a[:,i:n_features], temp_cov_a[:,0:i]]
temp_cov_b = np.r_[cov__test[i:n_features,:], cov__test[0:i,:]]
temp_cov_b = np.c_[temp_cov_b[:,i:n_features], temp_cov_b[:,0:i]]
La = temp_prec_a[:-1, :-1]
la = temp_prec_a[:-1, -1]
lama = temp_prec_a[-1, -1]
Wa = temp_cov_a[:-1, :-1]
wa = temp_cov_a[:-1, -1]
sigmaa = temp_cov_a[-1, -1]
Lb = temp_prec_b[:-1, :-1]
lb = temp_prec_b[:-1, -1]
lamb = temp_prec_b[-1, -1]
Wb = temp_cov_b[:-1, :-1]
wb = temp_cov_b[:-1, -1]
sigmab = temp_cov_b[-1, -1]
d_abp[i] = np.dot(wa, la)+0.5*(np.dot(np.dot(lb, Wb), lb)-np.dot(np.dot(la, Wa), la))+0.5*(np.log(lama/lamb)+sigmaa-sigmab)
d_abm[i] = np.dot(wb, lb)+0.5*(np.dot(np.dot(la, Wa), la)-np.dot(np.dot(lb, Wb), lb))+0.5*(np.log(lamb/lama)+sigmab-sigmaa)
d_ab[i] = max(-d_abp[i], -d_abm[i])
return d_ab
Bereiten Sie als Nächstes eine Funktion vor, die die Verwaltungsgrenze durch Schätzung der Kerneldichte berechnet.
def cl_limit(x, cr=0.99):
"""Kontrollgrenzen berechnen
Parameters
----------
x : np.array
KL Entfernung
cr : float
Kontrollgrenzengrenzen
Returns
-------
cl : float
Management-Limit-Grenze
"""
X = x.reshape(np.shape(x)[0],1)
bw= (np.max(X)-np.min(X))/100
kde = KernelDensity(kernel='gaussian', bandwidth=bw).fit(X)
X_plot = np.linspace(np.min(X), np.max(X), 1000)[:, np.newaxis]
log_dens = kde.score_samples(X_plot)
prob = np.exp(log_dens) / np.sum(np.exp(log_dens))
calprob = np.zeros(np.shape(prob)[0])
calprob[0] = prob[0]
for i in range(1,np.shape(prob)[0]):
calprob[i]=calprob[i-1]+prob[i]
cl = X_plot[np.min(np.where(calprob>cr))]
return cl
Überprüfen Sie die Trainingsdaten gegenseitig, um das Managementlimit zu berechnen.
K = 5
cv_data_size = np.int(np.shape(train_df_std)[0]/5)
n_train_samples = np.shape(train_df_std)[0]
counter = 0
for i in range(K):
cv_train_data=np.r_[train_df_std[0:i*cv_data_size,], train_df_std[(i+1)*cv_data_size:,]]
if i < K-1:
cv_test_data=train_df_std[i*cv_data_size:(i+1)*cv_data_size,]
else:
cv_test_data=train_df_std[i*cv_data_size:,]
model_cv = GraphicalLassoCV()
model_cv.fit(cv_train_data)
cov__cv = model.covariance_
prec__cv = model.precision_
for n in range(window_size, np.shape(cv_test_data)[0]):
count = i*cv_data_size + n
tempX = cv_test_data[n-window_size:n,:]
d_ab_temp = Calc_KL(cov__cv, prec__cv, tempX)
if 0 == counter:
d_ab = d_ab_temp.reshape(1,n_features)
TimeIndices2 = TimeIndices[count]
else:
d_ab = np.r_[d_ab,d_ab_temp.reshape(1,n_features)]
#Hier Fehler
TimeIndices2 = np.vstack((TimeIndices2,TimeIndices[count]))
counter = counter + 1
split_point = np.shape(d_ab)[0]
d_ab_cv = d_ab[np.sum(d_ab,axis=1)!=0,:]
cl = np.zeros([n_features])
for i in range(n_features):
cl[i] = cl_limit(d_ab_cv[:,i],cr)
Schließlich wird die Kovarianz der Bewertungsdaten geschätzt und visualisiert.
#Schätzen Sie die Kovarianz für Bewertungsdaten
n_test_samples = np.shape(test_df_std)[0]
for n in range(window_size, n_test_samples):
tempX = test_df_std[n-window_size:n,:]
d_ab_temp = Calc_KL(cov_, prec_, tempX)
d_ab = np.r_[d_ab,d_ab_temp.reshape(1,n_features)]
TimeIndices2 = np.vstack((TimeIndices2,TimeIndices[n+n_train_samples]))
x2 = [0, np.shape(d_ab)[0]]
x3 = [split_point, split_point]
x = range(0, np.shape(TimeIndices2)[0],20)
NewTimeIndices = np.array(TimeIndices2[x])
for i in range(38):
plt.figure(figsize=(200, 3))
plt.subplot(1, 38, i+1)
plt.title('%s Contribution' % (i))
plt.plot(d_ab[:, i], marker="o")
plt.xlabel("Time")
plt.ylabel("Contribution")
plt.xticks(x,NewTimeIndices,rotation='vertical')
y2 = [cl[i],cl[i]]
plt.plot(x2,y2,ls="-", color = "r")
y3 = [0, np.nanmax(d_ab[:,i])]
plt.plot(x3,y3,ls="--", color = "b")
plt.show()
schließlich
Vielen Dank für das Lesen bis zum Ende. Dieses Mal haben wir die Erkennung von Anomalien durch spärliches Strukturlernen implementiert.
Wenn Sie eine Korrekturanfrage haben, würden wir uns freuen, wenn Sie uns kontaktieren könnten.
Recommended Posts