Synced repo using 'sync_with_huggingface' Github Action

app.py

@@ -1,9 +1,13 @@
 import streamlit as st
+from src.colorer import get_logger
 
 def app():
     st.set_page_config(page_title="Algorithmic Trading Dashboard", layout="wide", initial_sidebar_state="auto",
                            menu_items=None, page_icon=":chart_with_upwards_trend:")
 
+    if "logger" not in st.session_state:
+        st.session_state.logger = get_logger()
+
     single_test = st.Page("page/single_backtest.py", title="Run Strategy")
     complete_test = st.Page("page/complete_backtest.py", title="Evaluate Strategy")
 
@@ -12,4 +16,5 @@ def app():
     pg.run()
 
 if __name__ == "__main__":
+    # run_logging()
     app()

indicators.py

@@ -610,7 +610,7 @@ def EMA(array, n):
     return pd.Series(array).ewm(span=n, adjust=False).mean()
 
 if __name__ == "__main__":
-    from utils import fetch
+    from src.utils import fetch
 
     data = fetch('ICICIBANK.NS', period='1mo', interval='15m')
     data = fetch('RELIANCE.NS', period='1mo', interval='15m')

page/complete_backtest.py

@@ -1,9 +1,9 @@
 import streamlit as st
 import pandas as pd
 import time
+
 from streamlit.components import v1 as components
-from utils import complete_test
-from os import cpu_count
+from src.utils import complete_test
 
 def complete_backtest():
     @st.cache_data

page/single_backtest.py

@@ -3,7 +3,7 @@ import streamlit as st
 from streamlit.components import v1 as components
 
 from indicators import SMC
-from utils import fetch, run_strategy
+from src.utils import fetch, run_strategy
 
 def algorithmic_trading_dashboard():
     @st.cache_data

src/utils.py

@@ -0,0 +1,121 @@
+import yfinance as yf
+from backtesting import Backtest
+import pandas as pd
+import os
+
+from multiprocessing import Pool
+from itertools import repeat
+from functools import partial
+from strategies import SMC_test, SMC_ema, SMCStructure
+from src.colorer import get_logger, start_end_log
+
+logger = get_logger()
+
+@start_end_log
+def fetch(symbol, period, interval):
+    logger.info(f"Fetching {symbol} for interval {interval} and period {period}")
+    df = yf.download(symbol, period=period, interval=interval, progress=False)
+    df.columns =df.columns.get_level_values(0)
+    return df
+
+@start_end_log
+def smc_backtest(data, filename, **kwargs):
+    bt = Backtest(data, SMC_test, cash=kwargs['cash'], commission=kwargs['commission'])
+    results = bt.run(swing_window=kwargs['swing_hl'])
+    bt.plot(filename=filename, open_browser=False)
+    return results
+
+@start_end_log
+def smc_ema_backtest(data, filename, **kwargs):
+    bt = Backtest(data, SMC_ema, cash=kwargs['cash'], commission=kwargs['commission'])
+    results = bt.run(swing_window=kwargs['swing_hl'], ema1=kwargs['ema1'], ema2=kwargs['ema2'], close_on_crossover=kwargs['cross_close'])
+    bt.plot(filename=filename, open_browser=False)
+    return results
+
+@start_end_log
+def smc_structure_backtest(data, filename, **kwargs):
+    bt = Backtest(data, SMCStructure, cash=kwargs['cash'], commission=kwargs['commission'])
+    results = bt.run(swing_window=kwargs['swing_hl'])
+    bt.plot(filename=filename, open_browser=False)
+    return results
+
+@start_end_log
+def run_strategy(ticker_symbol, strategy, period, interval, **kwargs):
+    logger.info(f'Running {strategy} for {ticker_symbol}')
+    # Fetching ohlc of random ticker_symbol.
+    retries = 3
+    for i in range(retries):
+        try:
+            data = fetch(ticker_symbol, period, interval)
+        except:
+            raise Exception(f"{ticker_symbol} data fetch failed")
+
+        if len(data) == 0:
+            if i < retries - 1:
+                print(f"Attempt{i + 1}: {ticker_symbol} ohlc is empty")
+            else:
+                raise Exception(f"{ticker_symbol} ohlc is empty")
+        else:
+            break
+
+    filename = f'{ticker_symbol}.html'
+
+    if strategy == "Order Block":
+        backtest_results = smc_backtest(data, filename, **kwargs)
+    elif strategy == "Order Block with EMA":
+        backtest_results = smc_ema_backtest(data, filename, **kwargs)
+    elif strategy == "Structure trading":
+        backtest_results = smc_structure_backtest(data, filename, **kwargs)
+    else:
+        raise Exception('Strategy not found')
+
+    with open(filename, 'r', encoding='utf-8') as f:
+        plot = f.read()
+
+    os.remove(filename)
+
+    # Converting pd.Series to pd.Dataframe
+    backtest_results = backtest_results.to_frame().transpose()
+
+    backtest_results['stock'] = ticker_symbol
+    backtest_results['plot'] = plot
+
+    # Reordering columns.
+    cols = ['stock', 'Start', 'End', 'Return [%]', 'Equity Final [$]', 'Buy & Hold Return [%]', '# Trades',
+            'Win Rate [%]', 'Best Trade [%]', 'Worst Trade [%]', 'Avg. Trade [%]', 'plot']
+    backtest_results = backtest_results[cols]
+
+    backtest_results = backtest_results.rename(columns = {'Equity Final [$]': 'Equity Final [₹]'})
+
+    return backtest_results
+
+@start_end_log
+def complete_test(strategy: str, period: str, interval: str, multiprocess=True, **kwargs):
+    nifty50 = pd.read_csv("data/ind_nifty50list.csv")
+    ticker_list = pd.read_csv("data/Ticker_List_NSE_India.csv")
+
+    # Merging nifty50 and ticker_list dataframes to get 'YahooEquiv' column.
+    nifty50 = nifty50.merge(ticker_list, "inner", left_on=['Symbol'], right_on=['SYMBOL'])
+
+    if multiprocess:
+        with Pool() as p:
+            result = p.starmap(partial(run_strategy, **kwargs), zip(nifty50['YahooEquiv'].values, repeat(strategy), repeat(period), repeat(interval)))
+    else:
+        result = [run_strategy(nifty50['YahooEquiv'].values[i], strategy, period, interval, **kwargs) for i in range(len(nifty50))]
+
+    df = pd.concat(result)
+
+    df['plot'] = df['plot'].astype(str)
+    df = df.sort_values(by=['Return [%]'], ascending=False)
+
+    return df.reset_index().drop(columns=['index'])
+
+
+if __name__ == "__main__":
+    # random_testing("")
+    # data = fetch('RELIANCE.NS', period='1y', interval='15m')
+    # df = yf.download('RELIANCE.NS', period='1yr', interval='15m')
+
+    rt = complete_test("Order Block", '1mo', '15m', swing_hl=20)
+    rt.to_excel('test/all_testing_2.xlsx', index=False)
+    print(rt)

strategies.py

@@ -4,6 +4,10 @@ from indicators import SMC, EMA
 import pandas as pd
 import numpy as np
 
+from src.colorer import get_logger
+
+logger = get_logger()
+
 class SMC_test(Strategy):
     swing_window = 10
     def init(self):
@@ -117,10 +121,9 @@ class SMCStructure(TrailingStrategy):
             try:
                 self.buy(sl=stoploss, tp=target)
             except:
-                print('Buying failed')
+                logger.warning(f'Buying failed at {price} with {stoploss=} and {target=}')
         if self.smc_s[-1] == 1:
             nearest = self.nearest_swing(self.data.df, self.swing_window)
-            print(self.data.df.iloc[-1])
             if nearest > price:
                 target = price - ((nearest - price) * .414)
                 stoploss = price + (price - target)
@@ -128,7 +131,7 @@ class SMCStructure(TrailingStrategy):
                 try:
                     self.sell(sl=stoploss, tp=target, limit=float(price))
                 except:
-                    print("Selling failed")
+                    logger.warning(f'Selling failed at {price} with {stoploss=} and {target=}')
 
         # Additionally, set aggressive stop-loss on trades that have been open
         # for more than two days
@@ -154,7 +157,7 @@ class SMCStructure(TrailingStrategy):
 strategies = {'Order Block': SMC_test, 'Order Block with EMA': SMC_ema , 'Structure trading': SMCStructure}
 
 if __name__ == "__main__":
-    from utils import fetch
+    from src.utils import fetch
     # data = fetch('ICICIBANK.NS', period='1mo', interval='15m')
     data = fetch('RELIANCE.NS', period='1mo', interval='15m')
     # data = fetch('AXISBANK.NS', period='1mo', interval='15m')