Introduction

Supertrend is a popular technical analysis indicator used by traders to identify potential trend reversals and market entry/exit points. It is particularly useful in volatile markets and can be implemented using Python programming. The SuperTrend indicator provides a dynamic support and resistance level, adjusting to price movements.

Platform Used

We have used FYERS API V3 platform to calculate the SuperTrend direction and it's value. FYERS API is FREE service offered by FYERS.

Components

• FYERS Object Creation
• HISTORY Data Fetch from Fyers
• CONVERT data to Pandas DataFrame
• DUPLICATE Records Removal
• EXTRA Row Removal
• SuperTrend Calculation
• Exact Timestamp Matching for Records

FYERS Object Creation

In order to fetch any data or communication with FYERS API, we need to make a fyers object. I have already explained the process in details. You can refer it here.

HISTORY Data Fetch from Fyers

Fyers has provided detailed documentation for different functions and data available through API. We are going to use HISTORY Data API to fetch the details. You can refer the official FYERS documentation here.

Below is the function created using the FYERS DATA API

def getLTP_his(symbol,intr,fyers,dd):
    print(f"getLTP_his ion being called with symbol {symbol}")
    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=dd)
    try:
        data = {"symbol":symbol,"resolution":intr,"date_format":"1","range_from":start_date,"range_to":end_date,"cont_flag":"1"}
        print(data)
        res = fyers.history(data)

        return res
    except Exception as e:
        print("Error occured =>", res)

Here,

• symbol is the stock/indice/script name for which you want to fetch the details
• intr is the interval for which data needs to be fetched like 5 minute, 15 minute, 60 minute, 1D etc. Each data candle will be of this time interval.
• fyers is the FYERS Object created.
• dd is the number of days for the date range. Like we need to fetch 15 minute interval data for last 10 days so dd value will be 10.

CONVERT data to Pandas DataFrame

Our next step is to convert the data received from getLTP_his function to pandas data frame so that required mathematical operations can be performed to calculate super trend.

def giveDF(tradeSymbol,duration,fyers,dd):

    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=dd)
    # start_date = end_date
    print(f"giveDF called and Now fetching records from {start_date} to {end_date}")
    response = getLTP_his(tradeSymbol,duration,logging,fyers,dd)
    #print(f"Response recived: {response}")
    volumes = []
    try:
        df = pd.DataFrame(response["candles"], columns=["timestamp", "Open", "High", "Low", "Close","Volume"])
    except Exception as e:
        print("An exception occurred:", e)
    #print(df.tail(10))
    df["date"] = df["timestamp"].apply(lambda x: datetime.fromtimestamp(x))
    df = df.drop("timestamp", axis=1)

    df = df.sort_values(by='date')
    print(f"Data Frame returned for {tradeSymbol}")
    return df

Here,

• symbol is the stock/indices/script name for which you want to fetch the details
• duration is the interval for which data needs to be fetched like 5 minute, 15 minute, 60 minute, 1D etc. Each data candle will be of this time interval.
• fyers is the FYERS Object created.
• dd is the number of days for the date range. Like we need to fetch 15 minute interval data for last 10 days so dd value will be 10.

Cleaning Data

Many times, we can duplicate records like two records of same timestamp or extra records of irrelevant timestamp and we need to have clean data to perform calculation for super trend.

DUPLICATE Records Removal

Below is the code to remove the duplicate records.

try:
        # First, make sure the "date" column is in datetime format
        df['date'] = pd.to_datetime(df['date'])
E
        # Check for duplicates based on the "date" column
        duplicates = df[df.duplicated(subset='date', keep='first')]
    
        if not duplicates.empty:
            print("************ Duplicate Records Exist ****************")
            print(f"Duplicate Records: {duplicates}")
        
            # Sort the DataFrame by the "date" column in descending order
            df = df.sort_values(by='date', ascending=True)

            # Remove duplicates based on the "date" column, keeping the first occurrence (which is the latest one)
            df = df.drop_duplicates(subset='date', keep='first')

            print(f"Deleted Duplicate Records: {duplicates}")
            print(duplicates)
    except Exception as e:
        print(f"Exception occured while checking duplicate record: {e}")

EXTRA Raw Removal

Some time, we have extra records. Like for 15 minute interval, first candle would be at 09:15 and last would be at 15:15 but data may contain records like 15:30 or 09:00 which is extract and same needs to be removed.
Below is the code to remove extra records:

 try:
        # Convert the 'Date' column to datetime if it's not already in that format
        df['date'] = pd.to_datetime(df['date'])

        # Define the times you want to filter out
        times_to_delete = ['15:30:00','09:00:00','09:14:00']

        # Check if the specified times exist in the DataFrame
        #mask = df['date'].dt.strftime('%H:%M:%S').isin(times_to_delete)
        mask = df['date'].dt.strftime('%H:%M:%S').isin(times_to_delete)

        if any(mask):
            # Delete rows with the specified times
            print(f"Before del len {len(df)}")
            df = df[~mask].reset_index(drop=True)
            print(f"After del len {len(df)}.")
            
        else:
            print("No matching rows found for before market timestamp.")
    except Exception as e:
        print(f"Exception occured in deleting before market rows: {e}")

SuperTrend Calculation

Now, we have all the cleaned up data and need to perform the calculation using pandas_ta library.

  # Assing SuperTrend Values
    try:
        sti = ta.supertrend(df['High'], df['Low'], df['Close'], length=10, multiplier=3)
        df['SuperTrend'] = np.round(sti['SUPERT_10_3.0'] / 0.05) * 0.05
        df['Direction'] = np.where(df['Close'] > df['SuperTrend'], 'up', 'down')
        print("Now priting last 5 rows of data frame...")
        print(df.tail(5))

    except Exception as e:
        print(f"Exception in SuperTrend 15 Minute Assining: {e}")

Now, our data frame each row have superTrend value and direction assigned for length 10 and multiplier 3 on 15 minute dataframe.

Fetch Record of Exact Timestamp

For any day, any interval, timestamp is the time when candle started so at 09:30 AM, dataframe will contain 09:15 candle and at 09:45 AM, data frame will contain 09:30 candle and so on.

But sometime, for example - at 09:30 you are expecting 09:15 candle to be present in dataframe but it may have both 09:15 completed and 09:30 candle as well with live data which is not completed hence incorrect. So we need to calculate the exact timestamp of 15 minute before (for 15 minute interval) and fetch the data.

    try:
        now = datetime.now()

        given_date = (now - timedelta(minutes=15)).replace(second=0).strftime("%Y-%m-%d %H:%M:%S")

        print(f"calculated currentTime is => {given_date}")

        try:
            print(f"calculated super currentTime is => {given_date}")
            matching_values_super = df.loc[df["date"] == given_date, ['Open', 'High', 'Low', 'Close','SuperTrend','Direction']].values
            print(f"Super Trend Matching values => {matching_values_super}")
            open_p, high_p, low_p, close_p,superT,Direc = matching_values_super[0]
            
        except Exception as e:
            print(e)
            print(f"Exception occured while fetching latest values to return from data frame value: {e}")
    except Exception as e:
        print(f"Error Occured: {e}")

Summary

Once, we have all the components, we can execute the code every 15 minute to fetch the latest super trend values:

import time
import pandas as pd
from datetime import datetime,timedelta
import pandas_ta as ta
import numpy as np
import datetime as dt

#################################### Getting Symbol latest quote and History Data ########################################

def getLTP_his(symbol,intr,logging,fyers,dd):
    print(f"getLTP_his ion being called with symbol {symbol}")
    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=dd)
    try:
        data = {"symbol":symbol,"resolution":intr,"date_format":"1","range_from":start_date,"range_to":end_date,"cont_flag":"1"}
        print(data)
        res = fyers.history(data)

        return res
    except Exception as e:
        print("Error occured =>", res)



def giveDF(tradeSymbol,duration,logging,fyers,dd):

    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=dd)
    # start_date = end_date
    print(f"giveDF called and Now fetching records from {start_date} to {end_date}")
    response = getLTP_his(tradeSymbol,duration,logging,fyers,dd)
    #print(f"Response recived: {response}")
    volumes = []
    try:
        df = pd.DataFrame(response["candles"], columns=["timestamp", "Open", "High", "Low", "Close","Volume"])
    except Exception as e:
        print("An exception occurred:", e)
    #print(df.tail(10))
    df["date"] = df["timestamp"].apply(lambda x: datetime.fromtimestamp(x))
    df = df.drop("timestamp", axis=1)

    df = df.sort_values(by='date')
    print(f"Data Frame returned for {tradeSymbol}")
    return df



def superTrend():
    print(f"SuperTrend function is being called.")
    df = giveDF(globalSymbol,15,logging,fyers,40)

    try:
        # First, make sure the "date" column is in datetime format
        df['date'] = pd.to_datetime(df['date'])

        # Check for duplicates based on the "date" column
        duplicates = df[df.duplicated(subset='date', keep='first')]
    
        if not duplicates.empty:
            print("************ Duplicate Records Exist ****************")
            print(f"Duplicate Records: {duplicates}")
        
            # Sort the DataFrame by the "date" column in descending order
            df = df.sort_values(by='date', ascending=True)

            # Remove duplicates based on the "date" column, keeping the first occurrence (which is the latest one)
            df = df.drop_duplicates(subset='date', keep='first')

            print(f"Deleted Duplicate Records: {duplicates}")
            print(duplicates)
    except Exception as e:
        print(f"Exception occured while checking duplicate record: {e}")

    try:
        # Convert the 'Date' column to datetime if it's not already in that format
        df['date'] = pd.to_datetime(df['date'])

        # Define the times you want to filter out
        times_to_delete = ['15:30:00','09:00:00','09:14:00']

        # Check if the specified times exist in the DataFrame
        #mask = df['date'].dt.strftime('%H:%M:%S').isin(times_to_delete)
        mask = df['date'].dt.strftime('%H:%M:%S').isin(times_to_delete)

        if any(mask):
            # Delete rows with the specified times
            print(f"Before del len {len(df)}")
            df = df[~mask].reset_index(drop=True)
            print(f"After del len {len(df)}.")
            
        else:
            print("No matching rows found for before market timestamp.")
    except Exception as e:
        print(f"Exception occured in deleting before market rows: {e}")

    # Assing SuperTrend Values
    try:
        sti = ta.supertrend(df['High'], df['Low'], df['Close'], length=10, multiplier=3)
        df['SuperTrend'] = np.round(sti['SUPERT_10_3.0'] / 0.05) * 0.05
        df['Direction'] = np.where(df['Close'] > df['SuperTrend'], 'up', 'down')
        print("Now priting last 5 rows of data frame...")
        print(df.tail(5))

    except Exception as e:
        print(f"Exception in SuperTrend 15 Minute Assing: {e}")

    try:
        now = datetime.now()

        given_date = (now - timedelta(minutes=15)).replace(second=0).strftime("%Y-%m-%d %H:%M:%S")
    
        print(f"calculated currentTime is => {given_date}")

        try:
            print(f"calculated super currentTime is => {given_date}")
            matching_values_super = df.loc[df["date"] == given_date, ['Open', 'High', 'Low', 'Close','SuperTrend','Direction']].values
            print(f"Super Trend Matching values => {matching_values_super}")
            open_p, high_p, low_p, close_p,superT,Direc = matching_values_super[0]
            
        except Exception as e:
            print(e)
            print(f"Exception occured while fetching latest values to return from data frame value: {e}")
    except Exception as e:
        print(f"Error Occured: {e}")

    return superT,Direc

if __name__ == "__main__":

    fyers, = <<YOUR CODE TO GENERATE THE FYERS OBJECT>>
    print(f"Session and Logging settings applied.")

    ############# GLOBAL VARIABLE ################

    globalSymbol = "NSE:NIFTY50-INDEX"
    start_time = datetime(datetime.now().year, datetime.now().month, datetime.now().day, 9, 15, 0)
    end_time = datetime(datetime.now().year, datetime.now().month, datetime.now().day, 19, 31, 0)

    while datetime.now() > start_time and datetime.now() < end_time:
        current_time = datetime.now()
        if (current_time.minute % 1 ==0 and current_time.second > 1 and current_time.second <= 3 and current_time.microsecond > 20000 ) :
            
            super,dire = superTrend()
            print(f"Super Trend value of {globalSymbol} is {super} and direction is {dire}.")
            time.sleep(3)

Output

Disclaimer

This blog is intended for educational purposes only. It provides insights into calculating the SuperTrend stock trading indicator using Python, motivated solely by my passion for Python scripting and the aim to assist people in scripting and developing their ideas in Python. It is important to note that the content presented in this blog does not endorse or encourage any reader to engage in stock trading or investment activities. I do not assume responsibility for any financial decisions made by readers. All individuals are strongly advised to consult with their licensed financial advisor before making any investment decisions. Keep in mind that trading involves risks, and past performance is not indicative of future results.

Not Satisfied, Follow us on YouTube for complete walkthrough