Anomaly Detection using EMA (Exponential Moving Average)

In my recent work on anomaly detection using Exponential Moving Average (EMA), I've found it incredibly effective in identifying unusual patterns in stock price movements. By smoothing out short-term volatility and highlighting longer trends, EMA helps me spot deviations that could indicate significant market events. Moving forward, I plan to integrate more sophisticated statistical techniques to enhance the accuracy of my anomaly detection model and explore real-time data analysis for more timely insights.

Link to my code

Install and import librariespip install gradio==4.5.0 -q

!pip install gradio==4.5.0 -q
import gradio as gr
import yfinance as yf
import pandas as pd
import plotly.graph_objects as go
import warnings
warnings.filterwarnings('ignore')        

Download stock price data

In this code, I've written a function named download_stock_data that takes a stock ticker symbol as input. It uses the yfinance library to download the last six months of closing prices for that particular stock and returns this data as a DataFrame.

# Function to download stock data
def download_stock_data(ticker):
    stock_data = yf.Ticker(ticker)
    df = stock_data.history(period="6mo")[['Close']]
    return df        

Calculate EMA and Control Limits (UCL & LCL)

In this code, I'm defining a function named calculate_ema_and_limits to process financial data. It calculates the Exponential Moving Average (EMA) of the 'Close' prices over a specified window and then determines the Upper Control Limit (UCL) and Lower Control Limit (LCL) based on a rolling standard deviation multiplied by a factor (no_of_stds). The results are added as new columns to the DataFrame df.

# Function to calculate EMA and Control Limits
def calculate_ema_and_limits(df, window, no_of_stds):
    # Calculate EMA
    df['EMA'] = df['Close'].ewm(span=window, adjust=False).mean()
    # Calculate rolling standard deviation
    rolling_std = df['Close'].rolling(window=window).std()
    # Calculate UCL and LCL
    df['UCL'] = df['EMA'] + (rolling_std * no_of_stds)
    df['LCL'] = df['EMA'] - (rolling_std * no_of_stds)
    return df        

Anomaly detection using EMA

In this code, I've written a function named detect_anomalies that identifies anomalies in a DataFrame df. It flags as anomalies all instances where the 'Close' values are either above the upper control limit ('UCL') or below the lower control limit ('LCL').

# Function for anomaly detection using EMA
def detect_anomalies(df):
    anomalies = (df['Close'] > df['UCL']) | (df['Close'] < df['LCL'])
    return anomalies        

Visualize the plots

# Function to create the plot
def plot_stock(ticker, window, no_of_stds):
    df = download_stock_data(ticker)
    df_with_ema = calculate_ema_and_limits(df, window, no_of_stds)
    anomalies = detect_anomalies(df_with_ema)
    df_with_ema['Anomaly'] = anomalies

    fig = go.Figure()
    fig.add_trace(go.Scatter(x=df_with_ema.index, y=df_with_ema['Close'], name='Close Price', mode='lines'))
    fig.add_trace(go.Scatter(x=df_with_ema.index, y=df_with_ema['EMA'], name='EMA', line=dict(width=1, dash='dash')))
    fig.add_trace(go.Scatter(x=df_with_ema.index, y=df_with_ema['UCL'], name='UCL', line=dict(width=1, dash='dot')))
    fig.add_trace(go.Scatter(x=df_with_ema.index, y=df_with_ema['LCL'], name='LCL', line=dict(width=1, dash='dot')))
    fig.add_trace(go.Scatter(x=df_with_ema.index, y=df_with_ema['Close'].where(df_with_ema['Anomaly']), mode='markers', name='Anomaly', marker=dict(color='red', size=10)))
    fig.update_layout(title=f'EMA and Anomalies for {ticker}', xaxis_title='Date', yaxis_title='Price', height=600)
    return fig        

Create UI using Gradio

In this code, I'm setting up a Gradio interface for my plot_stock function. It allows users to select a stock ticker from a dropdown, specify the window size for Exponential Moving Average (EMA), and choose the number of standard deviations through sliders. Once the parameters are set, it displays a plot showing stock anomalies, EMA, Upper Control Limit (UCL), and Lower Control Limit (LCL).

iface = gr.Interface(
    fn=plot_stock,
    inputs=[
        gr.Dropdown(choices=['RELIANCE.NS', 'TCS.NS', 'HDFCBANK.NS', 'INFY.NS', 'MARUTI.NS'], label="Select Stock Ticker"),
        gr.Slider(minimum=5, maximum=100, value=20, label="Window Size for EMA"),
        gr.Slider(minimum=1, maximum=3, value=2, label="Number of Standard Deviations")
    ],
    outputs="plot",
    title="Stock Anomaly Detection with EMA",
    description="Select a stock ticker and adjust the window size and standard deviations to view EMA, UCL, LCL, and anomalies."
)

if __name__ == "__main__":
    iface.launch()        

Output#1:

In the example below, we are able to find the Anomalies detected in RELIANCE.NS stock based on last 6 months data.

Output#2:

In the example below, we are able to find the Anomalies detected in INFY.NS stock based on last 6 months data.

Output#3:

In the example below, we are able to find the Anomalies detected in MARUTI.NS stock based on last 6 months data.