Merge pull request #50 from iamkrish-coder/develop

Added RSI, ATR, MA indicator #47 #48 #49

IntelliTrader/IntelliTrader.py

@@ -73,9 +73,9 @@
 
 # Input values
 exchange = 'NSE'
-symbol = 'SBIN'
-interval = '5minute'
-duration = 1 # in days
+symbol = 'SBICARD'
+interval = '15minute'
+duration = 3
 
 # Module usage
 
@@ -107,3 +107,7 @@
 
 ##### Indicators #####
 indicator.execute_handler('macd', datasource)
+indicator.execute_handler('rsi', datasource)
+indicator.execute_handler('atr', datasource)
+indicator.execute_handler('sma', datasource)
+indicator.execute_handler('ema', datasource)

IntelliTrader/IntelliTrader.pyproj

@@ -26,7 +26,10 @@
     <Compile Include="src\fetch.py" />
     <Compile Include="src\helper.py" />
     <Compile Include="IntelliTrader.py" />
+    <Compile Include="src\indicators\atr.py" />
+    <Compile Include="src\indicators\ma.py" />
     <Compile Include="src\indicators\macd.py" />
+    <Compile Include="src\indicators\rsi.py" />
     <Compile Include="src\modify.py" />
     <Compile Include="src\orders.py" />
     <Compile Include="src\ticker.py" />

IntelliTrader/Notes.txt

@@ -111,4 +111,7 @@ elif isinstance(data, (int, float)):
     # If data is an integer or float, convert it to a pandas Series
     data = pd.Series([data])
 else:
-    print("Calculating MACD...")
+    print("Calculating MACD...")
+
+
+    last_value = ema_line.iloc[-1]

IntelliTrader/src/indicator.py

@@ -5,6 +5,9 @@
 import pandas as pd
 from src.helper import Helper
 from src.indicators.macd import macd
+from src.indicators.rsi import rsi
+from src.indicators.atr import atr
+import src.indicators.ma as ma
 
 class Indicator:
     def __init__(self, params):
@@ -14,23 +17,26 @@ def execute_handler(self, indicator_option, dataset):
         match indicator_option:
             case 'macd':
                 self.option_macd(dataset)
-            #case 'rsi':
-            #    self.option_rsi(dataset)
-            #case 'atr':
-            #    self.option_atr(dataset)
-            #case 'sma':
-            #    self.option_sma(dataset)
-            #case 'ema':
-            #    self.option_ema(dataset)
+            case 'rsi':
+                self.option_rsi(dataset)
+            case 'atr':
+                self.option_atr(dataset)
+            case 'sma':
+                self.option_sma(dataset)
+            case 'ema':
+                self.option_ema(dataset)
             case _:
                 self.invalid_option(dataset)
 
     def option_macd(self, dataset):
         try:
             if dataset is not None and not dataset.empty:
                 # Calculate MACD
-                pdf = pd.DataFrame(dataset['close'])
+                pdf = pd.DataFrame(dataset)
                 macd_line, signal_line, macd_histogram = macd(pdf)
+                last_macd_value = macd_line[-1]
+                last_signal_value = signal_line[-1]
+                last_histogram_value = macd_histogram[-1]
 
                 # Print the calculated MACD values
                 print("\nMACD Line:")
@@ -47,6 +53,73 @@ def option_macd(self, dataset):
         except:
              self.prop['log'].error("The received object is not a valid DataFrame") 
 
+    def option_rsi(self, dataset):
+            try:
+                if dataset is not None and not dataset.empty:
+                    # Calculate RSI
+                    pdf = pd.DataFrame(dataset)
+                    rsi_line = rsi(pdf)
+                    last_rsi_value = rsi_line[-1]
+
+                    # Print the calculated RSI values
+                    print("\nRSI Line:")
+                    print(rsi_line)
+                else:
+                    self.prop['log'].error("Failed to calculate RSI") 
+                    return False
+            except:
+                 self.prop['log'].error("The received object is not a valid DataFrame") 
+
+    def option_atr(self, dataset):
+            try:
+                if dataset is not None and not dataset.empty:
+                    # Calculate ATR
+                    pdf = pd.DataFrame(dataset)
+                    atr_line = atr(pdf)
+                    last_atr_value = atr_line[-1]
+
+                    # Print the calculated ATR values
+                    print("\nATR Line:")
+                    print(atr_line)
+                else:
+                    self.prop['log'].error("Failed to calculate ATR") 
+                    return False
+            except:
+                 self.prop['log'].error("The received object is not a valid DataFrame") 
+
+    def option_sma(self, dataset):
+            try:
+                if dataset is not None and not dataset.empty:
+                    # Calculate SMA
+                    pdf = pd.DataFrame(dataset)
+                    sma_line = ma.sma(pdf)
+                    last_sma_value = sma_line.iloc[-1]
+
+                    # Print the calculated SMA values
+                    print("\nSMA Line:")
+                    print(sma_line)
+                else:
+                    self.prop['log'].error("Failed to calculate SMA") 
+                    return False
+            except:
+                 self.prop['log'].error("The received object is not a valid DataFrame") 
+
+    def option_ema(self, dataset):
+            try:
+                if dataset is not None and not dataset.empty:
+                    # Calculate EMA
+                    pdf = pd.DataFrame(dataset)
+                    ema_line = ma.ema(pdf)
+                    last_ema_value = ema_line.iloc[-1]
+
+                    # Print the calculated EMA values
+                    print("\nEMA Line:")
+                    print(ema_line)
+                else:
+                    self.prop['log'].error("Failed to calculate EMA") 
+                    return False
+            except:
+                 self.prop['log'].error("The received object is not a valid DataFrame") 
 
     def invalid_option(self, dataset):
         # Invalid indicator option provided

IntelliTrader/src/indicators/atr.py

@@ -0,0 +1,26 @@
+def atr(dataset, period=14):
+    # Check if dataset contains the required columns
+    data = dataset if 'close' in dataset and 'high' in dataset and 'low' in dataset else None
+    if data is not None and not data.empty:
+        # Extract the necessary columns from the data
+        close_prices = data['close'].astype(float)
+        high_prices = data['high'].astype(float)
+        low_prices = data['low'].astype(float)
+
+        # Check if the data has sufficient length for ATR calculations
+        if len(data) < period:
+            return None
+
+        true_ranges = []
+        for i in range(1, len(data)):
+            high_low = high_prices[i] - low_prices[i]
+            high_close = abs(high_prices[i] - close_prices[i - 1])
+            low_close = abs(low_prices[i] - close_prices[i - 1])
+            true_range = max(high_low, high_close, low_close)
+            true_ranges.append(true_range)
+
+        atr = [sum(true_ranges[:period]) / period]
+        for i in range(period, len(true_ranges)):
+            atr.append((atr[-1] * (period - 1) + true_ranges[i]) / period)
+
+        return atr

IntelliTrader/src/indicators/ma.py

@@ -0,0 +1,19 @@
+import pandas as pd
+
+def sma(dataset, period=5):
+    # Check if dataset contains the required columns
+    data = dataset if 'close' in dataset else None
+    if data is not None and not data.empty:
+        # Check if the data has sufficient length for SMA calculations
+        if len(data) < period:
+            return None
+        return data['close'].ewm(span=period, adjust=False).mean()
+
+def ema(dataset, period=5):
+    # Check if dataset contains the required columns
+    data = dataset if 'close' in dataset else None
+    if data is not None and not data.empty:
+        # Check if the data has sufficient length for EMA calculations
+        if len(data) < period:
+            return None
+        return data['close'].rolling(window=period).mean()

IntelliTrader/src/indicators/macd.py

@@ -1,7 +1,8 @@
 import pandas as pd
 
 def macd(dataset, short_period=12, long_period=26, signal_period=9):
-    data = dataset['close'] if 'close' in dataset else dataset
+    # Check if dataset contains the required columns
+    data = dataset['close'] if 'close' in dataset else None
     if data is not None and not data.empty:
         # Check if data has sufficient length for MACD calculations
         if len(data) < long_period:

IntelliTrader/src/indicators/rsi.py

@@ -0,0 +1,30 @@
+import numpy as np
+
+def rsi(dataset, period=14):
+    # Check if dataset contains the required columns
+    data = dataset['close'] if 'close' in dataset else None
+    # Check if data has sufficient length for RSI calculations
+    if len(data) < period:
+        return None
+
+    close_prices = [float(x) for x in data]
+    deltas = np.diff(close_prices)
+    seed = deltas[:period + 1]
+    up_periods = []
+    down_periods = []
+
+    up_periods.append(sum(x for x in seed if x >= 0) / period)
+    down_periods.append(abs(sum(x for x in seed if x < 0) / period))
+
+    for i in range(period + 1, len(close_prices)):
+        delta = deltas[i - 1]
+        if delta > 0:
+            up_periods.append((up_periods[-1] * (period - 1) + delta) / period)
+            down_periods.append((down_periods[-1] * (period - 1)) / period)
+        else:
+            up_periods.append((up_periods[-1] * (period - 1)) / period)
+            down_periods.append((down_periods[-1] * (period - 1) - delta) / period)
+
+    rs = np.array(up_periods) / np.array(down_periods)
+    rsi_line = 100 - (100 / (1 + rs))
+    return rsi_line