Spaces:
Sleeping
Sleeping
File size: 1,765 Bytes
964e116 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 | #!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on 2024-07-24 13:04:10 Wednesday
@author: Nikhil Kapila
"""
from pyod.models.knn import KNN
import plotly.graph_objects as go
import numpy as np
import pandas as pd
def sliding_windows(data:pd.DataFrame, lookback:int=30)->tuple[np.array, np.array, pd.Index]:
X, y, timestamps = [], [], []
for i in range(len(data) - lookback):
X.append(data.iloc[i:i + lookback].values)
y.append(data.iloc[i + lookback])
timestamps.append(data.index[i + lookback])
return np.array(X), np.array(y), pd.Index(timestamps)
def detect_anomalies_with_sliding_windows(data: pd.DataFrame, lookback: int = 30):
X, y, timestamps = sliding_windows(data, lookback)
X_flat = X.reshape(X.shape[0], -1)
clf = KNN()
clf.fit(X_flat)
is_anomaly = clf.predict(X_flat) # 0 for normal, 1 for anomaly
anomaly_scores = clf.decision_scores_
anomalies = pd.DataFrame({
'Timestamp': timestamps,
'Is_Anomaly': is_anomaly,
'Anomaly_Score': anomaly_scores,
'electricity': y.flatten()
}).set_index('Timestamp')
return anomalies
def plotly_anomaly(anomalies):
anomaly_filter = anomalies['Is_Anomaly'] == 1 # This should be a boolean series
fig = go.Figure()
fig.add_trace(go.Scatter(x=anomalies.index, y=anomalies['electricity'], mode='lines', name='Normal'))
fig.add_trace(go.Scatter(x=anomalies[anomaly_filter].index, y=anomalies[anomaly_filter]['electricity'], mode='markers', name='Anomaly', marker=dict(color='red')))
fig.update_layout(
title='Time Series of Energy Usage with Anomalies Highlighted',
xaxis_title='Time',
yaxis_title='Energy Usage',
legend_title='Legend'
)
return fig
|