<jupyter_start><jupyter_text>Detecção de AnomaliasAntes de falarmos sobre __detecção de anomalias__ precisamos definir alguns __conceitos básicos em estatística:__ 1. Variáveis AleatóriasUma __variável aleatória__ é uma __variável quantitativa, cujo resultado (valor) depende de fatores aleatórios.__Uma variável aleatória pode ser uma medição de um parâmetro que pode gerar valores diferentes. O conceito de variável aleatória é essencial em estatística e em outros métodos quantitativos para a representação de fenômenos incertos.Um __exemplo__ de uma variável aleatória é o __resultado do lançamento de um dado__ que pode dar qualquer número entre 1 e 6. Embora possamos conhecer os seus possíveis resultados, o resultado em si depende de fatores de sorte (álea).  1.1 Variáveis aleatórias discretasConsidere que jogamos um dado com 6 faces. Qual a probabilidade de sair o número 1? E de sair outro valor qualquer?Intuitivamente, a resposta seria 1/6.Vamos representar por X a variável aleatória: resultado do lançamento de um dado. Dessa forma, X pode assumir os valores 1,2,3,4,5,6. Em notação matemática:X $\in$ {1,2,3,4,5,6} - leia-se X pertence ao conjunto {1,2,3,4,5,6}$Prob$(X = 1) = 1/6$Prob$(X = 2) = 1/6$Prob$(X = 3) = 1/6$Prob$(X = 4) = 1/6$Prob$(X = 5) = 1/6$Prob$(X = 6) = 1/6Desta forma, podemos definir a __distribuição__ de X como uma função que descreve a __probabilidade relativa de uma variável aleatória tomar um valor dado.__ Neste caso, a distribuição estaria representada pelas probabilidades acima.Veja que podemos alterar a lei de probabilidade: por exemplo, podemos __supor__ que tenhamos um dado irregular, cujas probabilidades são dadas por:$Prob$(X = 1) = 3/10$Prob$(X = 2) = 2/10$Prob$(X = 3) = 2/10$Prob$(X = 4) = 1/10$Prob$(X = 5) = 1/10$Prob$(X = 6) = 1/10Desta forma, é 3 vezes mais provável que, ao lançarmos esse dado irregular, saia o valor 1 do que o valor 6, por exemplo.__Obs.:__ É fundamental que a soma das probabilidades dê 1, quando considerados todos os cenários possíveis!Vamos implementar no python esses exemplos:<jupyter_code>import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
def lanca_dado_regular():
    return np.random.choice([1,2,3,4,5,6], size = 1, p = [1/6, 1/6, 1/6, 1/6, 1/6, 1/6])[0]

def lanca_dado_irregular(probs):
    return np.random.choice([1,2,3,4,5,6], size = 1, p = probs)[0]<jupyter_output><empty_output><jupyter_text>Vamos fazer alguns experimentos: __lançar 6000 vezes um dado regular e 6000 vezes um dado irregular. Na sequência, comparar os histogramas dos resultados__<jupyter_code>probs = [3/10, 2/10, 2/10, 1/10, 1/10, 1/10]

lista_dado_reg = []
lista_dado_irreg = []

for i in range(6000):
    resultado1 = lanca_dado_regular()
    resultado2 = lanca_dado_irregular(probs)
    
    lista_dado_reg.append(resultado1)
    lista_dado_irreg.append(resultado2)
    
serie_reg = pd.Series(lista_dado_reg)
serie_irreg = pd.Series(lista_dado_irreg)
serie_reg
serie_irreg
plt.figure(figsize=[20,6])


plt.subplot(1,2,1)
plt.bar(serie_reg.value_counts().index, serie_reg.value_counts().values)
plt.xticks(fontsize = 14, fontweight = 'bold')
plt.yticks(fontsize = 14, fontweight = 'bold')
plt.xlabel("Resultado",fontsize = 14, fontweight = 'bold')
plt.ylabel("Qtd. repetições",fontsize = 14, fontweight = 'bold')
plt.title("Dado regular | probs: [1/6, 1/6, 1/6, 1/6, 1/6, 1/6]", fontsize = 16, fontweight = 'bold')

plt.subplot(1,2,2)
plt.bar(serie_irreg.value_counts().index, serie_irreg.value_counts().values)
plt.xticks(fontsize = 14, fontweight = 'bold')
plt.yticks(fontsize = 14, fontweight = 'bold')
plt.xlabel("Resultado",fontsize = 14, fontweight = 'bold')
plt.ylabel("Qtd. repetições",fontsize = 14, fontweight = 'bold')
plt.title("Dado irregular | probs: [3/10, 2/10, 2/10, 1/10, 1/10, 1/10]", fontsize = 16, fontweight = 'bold')

plt.show()<jupyter_output><empty_output><jupyter_text>No caso do exemplo anterior, a variável X era __discreta__, pois os valores que X podia assumir eram apenas 1,2,3,4,5,6. 2. Variáveis aleatórias contínuasAgora considere que X seja o resultado da medição da __altura de uma pessoa.__ Neste caso, X pode assumir qualquer valor real, desta forma X é uma variável aleatória __contínua.__Vamos trabalhar um pouco mais nesse exemplo: X = altura de pessoas.Neste caso, não podemos enunciar todos os valores de X e atribuir a cada um deles uma probabilidade. No entanto, logicamente é muito mais provável que a altura de uma pessoa esteja entre [1.5m, 1.8m] do que [1.9m, 2.1m].Além disso, é muito improvável que uma pessoa tenha altura maior do que 2m e quase impossível alguém ter mais do que 2.5m...Como podemos definir a __distribuição__ de X neste caso?!No caso em que X era discreto - lançamento de um dado - as distribuições eram:<jupyter_code>def distr_X_lanc_dado_regular(x):
    return 1/6

def distr_X_lanc_dado_irregular(x):
    if x == 1:
        return 3/10
    if x == 2:
        return 2/10
    if x == 3:
        return 2/10
    if x == 4:
        return 1/10
    if x == 5:
        return 1/10
    if x == 6:
        return 1/10
xvals = [1,2,3,4,5,6]

plt.figure(figsize = [20,6])
plt.subplot(1,2,1)
plt.scatter(xvals, [distr_X_lanc_dado_regular(x) for x in xvals],  s = 100, c = 'blue')
plt.title("Distribuição de X: dado regular", fontsize = 16, fontweight = 'bold')
plt.xlabel("Valores de X", fontsize = 14, fontweight = 'bold')
plt.ylabel("Densidade", fontsize = 14, fontweight = 'bold')
plt.ylim([0,0.4])
plt.grid()
plt.subplot(1,2,2)
plt.scatter(xvals, [distr_X_lanc_dado_irregular(x) for x in xvals],  s = 100, c = 'blue')
plt.title("Distribuição de X: dado irregular", fontsize = 16, fontweight = 'bold')
plt.xlabel("Valores de X", fontsize = 14, fontweight = 'bold')
plt.ylabel("Densidade", fontsize = 14, fontweight = 'bold')
plt.ylim([0,0.4])
plt.grid()
plt.show()<jupyter_output><empty_output><jupyter_text>Podemos fazer o mesmo no caso em que X é contínua - por ex, altura das pessoas!Desta forma, podemos __modelar__ X de modo que para qualquer valor que X assuma, tenhamos uma densidade atribuida.Essa função densidade caracteriza a __distribuição de X!__Finalmente, podemos definir a __distribuição gaussiana:__Uma variável aleatória contínua __X segue a distribuição gaussiana__ com __média__ $\mu$ e desvio padrão $\sigma$ se a densidade de X for modelada de acordo: ![Captura%20de%20Tela%202020-10-25%20a%CC%80s%2018.17.33.png](attachment:Captura%20de%20Tela%202020-10-25%20a%CC%80s%2018.17.33.png) Nesta fórmula, os valores de $\mu$ e $\sigma$ estão fixados. Para cada valor de $x$, podemos calcular a densidade $f(x)$. Quanto maior for essa densidade no ponto $x$, mais provável é de retirarmos um valor próximo do ponto.A distribuição gaussiana também é conhecida por __distribuição normal.__Vamos implementar essa função no python para analisarmos melhor:<jupyter_code>def distr_gaussiana(x, media, desvio_padrao):
    return (1 / (desvio_padrao * np.sqrt(2 * np.pi))) * np.exp( (-0.5) * ((x - media)/desvio_padrao)**2 )<jupyter_output><empty_output><jupyter_text>Abaixo, vamos __modelar__ X com distribuição gaussiana com média 0 e desvio padrão 1:<jupyter_code>media = 0
desvio_padrao = 1

xlims = np.arange(-10,10,0.01)
distr_gaus = [distr_gaussiana(x, media, desvio_padrao) for x in xlims]

plt.figure(figsize=[8,6])
plt.plot(xlims, distr_gaus, lw = 2, c = 'blue')
plt.xlabel("valores de X",fontsize = 14, fontweight = 'bold')
plt.ylabel("densidade",fontsize = 14, fontweight = 'bold')
plt.title("X ~ Normal(media = 0, desv_padrao = 1)",fontsize = 14, fontweight = 'bold')
plt.grid()
plt.show()<jupyter_output><empty_output><jupyter_text>__Faz sentido dizer que X, altura de pessoas, segue essa distribuição?!__De acordo com o distribuição acima, os valores mais prováveis de X é em torno de X = 0. Além disso, X pode assumir valores negativos...então me parece que essa não é a melhor modelagem para X = altura de pessoas.Mas.. e se mudarmos os parâmetros média e desvio padrão?Considere o exemplo:<jupyter_code>media = 1.7
desvio_padrao = 0.15

xlims = np.arange(1,2.5,0.01)
distr_gaus = [distr_gaussiana(x, media, desvio_padrao) for x in xlims]

plt.figure(figsize=[8,6])
plt.plot(xlims, distr_gaus, lw = 2, c = 'blue')
plt.xlabel("valores de X",fontsize = 14, fontweight = 'bold')
plt.ylabel("densidade",fontsize = 14, fontweight = 'bold')
plt.title("X ~ Normal(media = 0, desv_padrao = 1)",fontsize = 14, fontweight = 'bold')
plt.grid()
plt.show()<jupyter_output><empty_output><jupyter_text>__Agora sim parece que essa ditribuição descreve muito melhor X = altura de pessoas!__Existem várias outras distribuições, mas a __distribuição normal__ é a distribuição mais conhecida e usada dentro de modelagem estatística. Vários fenômenos da natureza podem ser modelados a partir dessa distribuição.__Pergunta:__ como escolher o melhor conjunto de parâmetros média e desvio padrão para a modelagem?Considere que temos um dataset com N medições:<jupyter_code>df = pd.read_csv("amostra_peso.csv")
df
plt.hist(df.peso, bins = 50)
plt.title("Histograma do peso de 1000 pessoas", fontsize = 16)
plt.show()<jupyter_output><empty_output><jupyter_text>Parece que segue uma distribuição normal! Como podemos modelar X = peso das pessoas de acordo com essa amostra?Podemos __usar a própria amostra__ para aproximar os valores da média $\mu$ e desvio padrão $\sigma$:<jupyter_code>#módulo para trabalhar com distribuições estatísticas em python
import scipy.stats as st
media_amostral = df.peso.mean()
desvio_padrao_amostral = df.peso.std()

#criando a variável aleatória X no python
X = st.norm(loc = media_amostral, scale = desvio_padrao_amostral)<jupyter_output><empty_output><jupyter_text>__Simulando valores de X:__a função "rvs" gera valores random, ou seja, aleatórios, a partir da distribuição.<jupyter_code>X.rvs(1)
X.rvs(10)<jupyter_output><empty_output><jupyter_text>Vamos gerar 5000 valores aleatórios e comparar os resultados:<jupyter_code>plt.figure(figsize=[20,8])

plt.subplot(1,2,1)

plt.hist(df.peso, bins = 50)
plt.title("Histograma do peso de 1000 pessoas", fontsize = 16)

plt.subplot(1,2,2)

plt.hist(X.rvs(5000), bins = 50)
plt.title("valores simulados", fontsize = 16)
plt.show()<jupyter_output><empty_output><jupyter_text>__Veja quanta semelhança!!__Qual a função de densidade para esse X?Podemos usar a função "pdf" para isso. "pdf" significa probability density function!<jupyter_code>xvals = [30,40,50,60,70,80,90,100,110]

plt.plot(xvals, [X.pdf(x) for x in xvals], lw = 2, c = 'blue')
plt.grid()
plt.show()<jupyter_output><empty_output><jupyter_text>Agora que já entendemos bastante a respeito da distribuição normal, vamos começar a falar sobre detecção de anomalias e faremos forte uso da distribuição normal neste tema! 3. Detecção de anomalias: MotivaçãoConsidere o seguinte problema:Você é um cientista de dados que trabalha no ramo de aviação. Uma vez que um novo motor é construído, testes precisam ser executados para garantir que este motor esteja em bom estado.Considere que você tenha acesso a duas medidas de teste do motor: o calor gerado e a intensidade de vibração.Além disso, você possui uma base histórica com essas variáveis de outros motores que já foram testados e funcionam bem:<jupyter_code>df = pd.read_csv("dataset_deteccao_anomalias_motor_aviacao.csv")
print(df.shape)
df.head()
plt.figure(figsize=[10,6])
plt.title("Dispersão das variáveis - dados históricos",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(df.calor, df.vibracao, s = 50)
plt.xlim([5,15])
plt.ylim([15,25])
plt.grid()
plt.show()
calor_novoMotor = 10
vibracao_novoMotor = 20

plt.figure(figsize=[10,6])
plt.title("Esse novo motor poderia ser considerado uma anomalia!?",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(df.calor, df.vibracao, s = 50)
plt.scatter(calor_novoMotor, vibracao_novoMotor, s = 300, marker = 'x', lw = 5, c = 'red', label = 'novo motor')
plt.legend(fontsize = 14)
plt.xlim([5,15])
plt.ylim([15,25])
plt.grid()
plt.show()<jupyter_output><empty_output><jupyter_text><jupyter_code>calor_novoMotor = 12
vibracao_novoMotor = 20

plt.figure(figsize=[10,6])
plt.title("E esse!?",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(df.calor, df.vibracao, s = 50)
plt.scatter(calor_novoMotor, vibracao_novoMotor, s = 300, marker = 'x', lw = 5, c = 'red', label = 'novo motor')
plt.legend(fontsize = 14)
plt.xlim([5,15])
plt.ylim([15,25])
plt.grid()
plt.show()<jupyter_output><empty_output><jupyter_text><jupyter_code>calor_novoMotor = 6
vibracao_novoMotor = 24

plt.figure(figsize=[10,6])
plt.title("E esse!?",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(df.calor, df.vibracao, s = 50)
plt.scatter(calor_novoMotor, vibracao_novoMotor, s = 300, marker = 'x', lw = 5, c = 'red', label = 'novo motor')
plt.legend(fontsize = 14)
plt.xlim([5,15])
plt.ylim([15,25])
plt.grid()
plt.show()<jupyter_output><empty_output><jupyter_text>Essencialmente, um método para respondermos a essas perguntas seria:__1: Modelar os dados de treino com uma distribuição específica.____2: A partir desta distribuição, calcular a probabilidade de encontrarmos esse novo motor.____3: Se a probabilidade for baixa (menor do que um limiar fixado), então dizemos que o novo motor é uma anomalia. Caso contrário, diremos que não é uma anomalia.__Pronto! Usando os conceitos de distribuição e probabilidade estudados anteriormente, podemos definir um algoritmo para detecção de anomalias! 4. Detecção de anomalias: AlgoritmoDenote por X1 a variável aletória que mede o calor nos motores e por X2 a variável aleatória que mede a vibração.Utilizando o dataset de treino, podemos modelar X1 e X2 com distribuições gaussianas, cada um com uma média e um desvio padrão calculados a partir dos dados.Agora, considerando um vetor 2-dimensional $x = [x1,x2]$, podemos definir:$ p(x) = p(x1) \cdot p(x2) $, em que $p(x1)$ é a densidade de $x1$, de acordo com a gaussiana definida e analogamente para $x2$.Desta forma, se $p(x)$ for menor do que um limiar $\epsilon$, dizemos que o vetor $x$ é uma anomalia. Caso contrário, não é uma anomalia.__Obs.:__ veja que o algoritmo acima pode ser generalizado!Considere que temos um dataset com p variáveis $X_1$, ..,  $X_p$. Podemos calcular:$\mu_1, \sigma_1$: média e desvio padrão de $X_1$ $\mu_2, \sigma_2$: média e desvio padrão de $X_2$ ..$\mu_p, \sigma_p$: média e desvio padrão de $X_p$ e modelar:$X_1$ ~  $Normal(\mu_1, \sigma_1)$, $X_2$ ~ $Normal(\mu_2, \sigma_2)$, .., $X_p$ ~ $Normal(\mu_p, \sigma_p)$Então, definir  $ p(x) = p(x1) \cdot p(x2) \cdot .. \cdot p(x_p)$ para um vetor $x = [x_1, x_2, .., x_p]$.Dessa forma, __$x$ será anomalia caso $p(x) \lt \epsilon$.__Implementando em python:<jupyter_code>class DetectorAnomalias():
    
    def __init__(self, epsilon):
        self.epsilon = epsilon
        
    def fit(self, X):
        medias = X.mean(axis = 0)
        desvios = X.std(axis = 0)
        gaussianas = [st.norm(loc = m, scale = d) for m, d in zip(medias, desvios)]  
        self.gaussianas = gaussianas
        self.X = X
        
    def prob(self, x):
        p = 1
        for i in range(self.X.shape[1]):
            gaussiana_i = self.gaussianas[i]
            x_i = x[i]
            p *= gaussiana_i.pdf(x_i)
        return p
    
    def isAnomaly(self, x):
        return int(np.where(self.prob(x) < self.epsilon, 1, 0))
df.head()
X = df.values
#instanciando o algoritmo
ann = DetectorAnomalias(epsilon = 0.001)
ann.fit(X)
#fazendo testes
x = X[0, ]
x
ann.prob(x)
ann.isAnomaly(x)
x_teste = np.array([6,20])
x_teste
print(ann.prob(x_teste))
print(ann.isAnomaly(x_teste))<jupyter_output>0.00012932693620458093
1<jupyter_text>Vamos, abaixo, plotar alguns gráficos que resumem o resultado do algoritmo.Plotaremos as curvas de probabilidade de cada ponto e a divisão binária entre ser ou não uma anomalia.Além disso, vamos comparar dois detectores de anomalia com limiares distintos e ver como isso afeta as predições.<jupyter_code>xx = np.arange(6,14, 0.01)
yy = np.arange(16.5,23.5, 0.01)

xx0, xx1 = np.meshgrid(xx, yy)
grid0 = xx0.ravel()
grid1 = xx1.ravel()
grid = np.c_[grid0, grid1]

lista_anomalia = []
lista_prob = []
L=[]
for x in grid: #selecionando os pontos cuja probabilidade está muito próxima do limiar epsilon pré-fixado.
    lista_anomalia.append( ann.isAnomaly(x) )
    lista_prob.append( ann.prob(x) )
    
    if (( (ann.epsilon - 0.0001) < ann.prob(x)) & ((ann.epsilon + 0.0001) > ann.prob(x))):
        L.append(x)
    
plt.figure(figsize=[12,15])
plt.subplot(2,1,1)
plt.scatter(grid0, grid1, c = lista_prob, cmap = plt.cm.rainbow)
plt.clim(0.01, 0.15)
plt.colorbar()
for l in L:
    plt.scatter(l[0], l[1], c = 'white', s = 10)

plt.title("Detector de anomalias (limiar 0.001): probabilidades",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(df.calor, df.vibracao, s = 75, c = 'black', label = 'dados de treino')
plt.legend(loc = 'lower left', fontsize = 14)

plt.xlim([6,14])
plt.ylim([16.5,23.5])
plt.grid()

plt.subplot(2,1,2)
plt.title("Detector de anomalias (limiar 0.001): predição",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(grid0, grid1, c = lista_anomalia, cmap = plt.cm.Blues_r)
plt.scatter(df.calor, df.vibracao, s = 75, c = 'black')

plt.xlim([6,14])
plt.ylim([16.5,23.5])
plt.grid()

plt.show()<jupyter_output><empty_output><jupyter_text>__Vamos alterar o limiar (aumentá-lo 10x) e comparar os resultados:__<jupyter_code>ann = DetectorAnomalias(epsilon = 0.01)
ann.fit(X)
xx = np.arange(6,14, 0.01)
yy = np.arange(16.5,23.5, 0.01)

xx0, xx1 = np.meshgrid(xx, yy)
grid0 = xx0.ravel()
grid1 = xx1.ravel()
grid = np.c_[grid0, grid1]

lista_anomalia = []
lista_prob = []
L=[]
for x in grid:
    lista_anomalia.append( ann.isAnomaly(x) )
    lista_prob.append( ann.prob(x) )
    
    if (( (ann.epsilon - 0.0001) < ann.prob(x)) & ((ann.epsilon + 0.0001) > ann.prob(x))):
        L.append(x)
    
plt.figure(figsize=[12,15])
plt.subplot(2,1,1)
plt.scatter(grid0, grid1, c = lista_prob, cmap = plt.cm.rainbow)
plt.clim(0.01, 0.15)
plt.colorbar()
for l in L:
    plt.scatter(l[0], l[1], c = 'white', s = 10)

plt.title("Detector de anomalias (limiar 0.01): probabilidades",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(df.calor, df.vibracao, s = 75, c = 'black', label = 'dados de treino')
plt.legend(loc = 'lower left', fontsize = 14)

plt.xlim([6,14])
plt.ylim([16.5,23.5])
plt.grid()

plt.subplot(2,1,2)
plt.title("Detector de anomalias (limiar 0.01): predição",fontsize = 16, fontweight = 'bold')
plt.xlabel("Calor",fontsize = 14, fontweight = 'bold')
plt.ylabel("Vibração",fontsize = 14, fontweight = 'bold')
plt.scatter(grid0, grid1, c = lista_anomalia, cmap = plt.cm.Blues_r)
plt.scatter(df.calor, df.vibracao, s = 75, c = 'black')

plt.xlim([6,14])
plt.ylim([16.5,23.5])
plt.grid()

plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.128674772972141, 'fraction_numerical': 0.04186547637371094, 'mean_word_length': 4.937252512945476, 'pattern_counts': {'":': 0, '<': 71, '<?xml version=': 0, '>': 70, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '470140', 'license': 'MIT', 'n_tokens_mistral': 8406, 'n_tokens_neox': 7663, 'n_words': 2096, 'path': 'Modulo 14 - Deteccao de Anomalias.ipynb', 'repo_name': 'dimagela29/Formacao_cientista_de_dados', 'size': '19491'}