Python & LLM for market Analysis — Part I (Technical Indicators)

Many traders incorporate technical strategies alongside their fundamental approaches in an attempt to perfect their market entry and exit timings. Achieving impeccable timing in the market might seem elusive, but there are valuable indicators that can significantly improve your buying and selling decisions.

In this post, we will explore how to compute essential technical indicators using Python and Pandas, leveraging TA (Technical Analysis) libraries. While there’s a multitude of indicators to choose from, I’ll focus on a few of my personal favorites:

1. RSI
2. Exponential Moving Average (EMA)
3. Keltner Channels

To get started, you’ll need historical data. While I prefer using the Zerodha Kite API for real-time trading and historical data, it does come with a subscription fee. Alternatively, you can access free data through the GOOGLEFINANCE function in Google Sheets or use the jugaad_data library in Python.

Here are the essential data points you’ll require:

• DATE
• Open
• High
• Low
• Close

For this guide, we’ll demonstrate using the jugaad_data library. Please note that while this library provides free data, it may not be actively maintained or completely reliable. For more dependable data sources, consider Google Finance or Zerodha Kite API.

Now, let’s delve into the process of calculating these crucial technical indicators to enhance your trading strategy.

1. Install the library

pip install jugaad-data

2. import necessary libraries

from datetime import datetime, timedelta
import pandas as pd
from jugaad_data.nse import stock_df

3. Get the data in a Pandas DataFrame

@staticmethod
def extract_jugaad_data(tick,start,end):
        df1 = stock_df(symbol=tick, from_date=start,
            to_date=end, series="EQ")
        return df1

Above method requires us to send the name of the stock and start and end date. This method will always return daily historical data. for more options like 5m, 10m, 1h, 1W etc, use Zerodha kite api.

4. Let’s say we want to extract 5 years historical data. we have to iteratively extract data for every year. running a huge date range can cause performance issues and latencies and can cause failures. hence we need this iterative approach.

class HistoricalData:
    @staticmethod
    def calculate_timelines(years):
        curr = datetime.now().date()
        ans = []
        while years>0:
            start = curr- timedelta(days=365)
            ans.insert(0,(start,curr))
            curr=start
            years-=1
        return ans

The above method will return a list of start and end date.

5. Now to get the historical data, lets do below

   def get_history_df(tick):
        df = pd.DataFrame({})
        for start,end in HistoricalData.calculate_timelines(5):
            df = pd.concat([df,HistoricalData.extract_jugaad_data(tick,start,end)],ignore_index=True)
        df = df[["DATE", "OPEN", "HIGH","LOW","CLOSE","VOLUME","SYMBOL"]]
        df["DATE"] = pd.to_datetime(df["DATE"], format='%Y-%m-%d')
        df.set_index('DATE',inplace=True)
        df = df[~df.index.duplicated(keep='first')]
        df = df.sort_values(by='DATE')

here we iterate through the historical data and start, end date and concat the data to the data frame. then use selected columns and convert the DATE column to pandas datetime field. Perform a data cleanup by removing duplicates and sort the data by DATE field. ensure to set the index as date as there should be only one row per date.

6. lets touch the most important part of this post, calculating the technical indicators. here we are using 2 libraries

pip install ta
pip install TA-Lib

7. Both the libraries support wide variety of technical indicators. lets try some of them. lets create a file tech_analysis.py

import pandas as pd
import numpy as np
from talib import RSI, EMA, stream
from ta.volatility import KeltnerChannel
class TechAnalysis:
    def __init__(self):
        self.ema_period = 13
        self.rsi_period = 7
        self.stochastic_period = 7
        self.rsi_divergence_period = 14

    def calculate_technicals(self, df):
        try:
            df['EMA'] = EMA(df['CLOSE'],timeperiod=self.ema_period)
            df['RSI7'] = RSI(df['EMA'], timeperiod=self.rsi_period)
            df['RSI14'] = RSI(df['CLOSE'], timeperiod=self.rsi_divergence_period)
            indicator_kc = KeltnerChannel(high=df["HIGH"],low=df["LOW"],close=df["CLOSE"], window=50, window_atr=50,fillna=False,multiplier=5, original_version=False)
            df["KC_UPPER"] = indicator_kc.keltner_channel_hband()
            df["KC_MIDDLE"] = indicator_kc.keltner_channel_mband()
            df["KC_LOWER"] = indicator_kc.keltner_channel_lband()
            return df
        except Exception as e:
            print('error while calculating technicals')
            print(f'exception occured {e}')

TA = TechAnalysis()

Note how we are using EMA for RSI7 but CLOSE for RSI14. there are different ways to calculate RSI indicators and each has its pros and cons. EMA(),RSI() are part of talib while KeltnerChannel() is part of ta.volatility.

Keltner channel has 3 bands, upper middle and lower. to calculate it, initiate a keltner channel

indicator_kc = KeltnerChannel(high=df["HIGH"],low=df["LOW"],
close=df["CLOSE"], window=50, window_atr=50,fillna=False,multiplier=5,
original_version=False)

we use 50 days window with 5 as a multiplier. Note that Keltner Channel requires high and low values.

then use the indicator_kc to calculate the bands. For example to calculate Upper band indicator_kc.keltner_channel_hband(). I recommend exporing other methods of these 2 libraries.

Here is the link for TA-Lib and link fot ta.

Return the dataframe so that the columns are added to the data frame.

8. now lets update our get_history_df(tick) from our step 5 to calculate the technical indicators. add a import at top and store the data in csv.

from tech_analysis import TA

def get_history_df(tick):
        df = pd.DataFrame({})
        for start,end in HistoricalData.calculate_timelines(5):
            df = pd.concat([df,HistoricalData.extract_jugaad_data(tick,start,end)],ignore_index=True)
        df = df[["DATE", "OPEN", "HIGH","LOW","CLOSE","VOLUME","SYMBOL"]]
        df["DATE"] = pd.to_datetime(df["DATE"], format='%Y-%m-%d')
        df.set_index('DATE',inplace=True)
        df = df[~df.index.duplicated(keep='first')]
        df = df.sort_values(by='DATE')
        df = TA.calculate_technicals(df)
        df.to_csv(f'./{tick}.csv')

the above code will now store the csv file in the same folder where the code is.

9. overall what we have seen so far can be put together as below. Run the technical_data.py and observe that the csv file is generated with technical indicator values

technicals_data.py

from datetime import datetime, timedelta
import pandas as pd
from jugaad_data.nse import stock_df
from tech_analysis import TA
class HistoricalData:
    @staticmethod
    def calculate_timelines(years):
        curr = datetime.now().date()
        ans = []
        while years>0:
            start = curr- timedelta(days=365)
            ans.insert(0,(start,curr))
            curr=start
            years-=1
        return ans

    @staticmethod
    def extract_jugaad_data(tick,start,end):
        df1 = stock_df(symbol=tick, from_date=start,
            to_date=end, series="EQ")
        return df1

    @staticmethod
    def get_history_df(tick):
        df = pd.DataFrame({})
        for start,end in HistoricalData.calculate_timelines(5):
            df = pd.concat([df,HistoricalData.extract_jugaad_data(tick,start,end)],ignore_index=True)
        df = df[["DATE", "OPEN", "HIGH","LOW","CLOSE","VOLUME","SYMBOL"]]
        df["DATE"] = pd.to_datetime(df["DATE"], format='%Y-%m-%d')
        df.set_index('DATE',inplace=True)
        df = df[~df.index.duplicated(keep='first')]
        df = df.sort_values(by='DATE')
        df = TA.calculate_technicals(df)
        df.to_csv(f'./{tick}.csv')

HistoricalData.get_history_df("INFY")

tech_analysis.py

import pandas as pd
import numpy as np
from talib import RSI, EMA, stream
from ta.volatility import KeltnerChannel
class TechAnalysis:
    def __init__(self):
        self.ema_period = 13
        self.rsi_period = 7
        self.stochastic_period = 7
        self.rsi_divergence_period = 14

    def calculate_technicals(self, df):
        try:
            df['EMA'] = EMA(df['CLOSE'],timeperiod=self.ema_period)
            df['RSI'] = RSI(df['EMA'], timeperiod=self.rsi_period)
            df['RSI14'] = RSI(df['CLOSE'], timeperiod=self.rsi_divergence_period)
            indicator_kc = KeltnerChannel(high=df["HIGH"],low=df["LOW"],close=df["CLOSE"], window=50, window_atr=50,fillna=False,multiplier=5, original_version=False)
            df["KC_UPPER"] = indicator_kc.keltner_channel_hband()
            df["KC_MIDDLE"] = indicator_kc.keltner_channel_mband()
            df["KC_LOWER"] = indicator_kc.keltner_channel_lband()
            return df
        except Exception as e:
            print('error while calculating technicals')
            print(f'exception occured {e}')

TA = TechAnalysis()

Try to build chart with the indicators in excel and notice how similar it is when compared to the charting tools.Feel free to refer the documentation of both the libraries and extend and add as many technical indicators as needed.

In the next post we will dive deep into how to develop Trading Strategies based on the indicators and also how to back test each strategy to see which one is more effective. Further more, we will see how to automate the trading with Python based on the chosen strategy and develop a trading bot that performs buy, sell automatically in Zerodha Kite Platform. In the last few posts, we will also see how to leverage LLMs to scrape the news from several sources and predict impacts that it may generate on the market.

Incase of queries or details, Please feel free to connect with me on LinkedIn or X(formerly twitter).