Upload 6 files

app.py

@@ -0,0 +1,148 @@
+from dash import Dash, Input, Output, State, callback, dcc, html
+import dash_bootstrap_components as dbc
+import pandas as pd
+from indicators import SMC
+from data_fetcher import fetch
+from strategies import smc_plot_backtest, smc_ema_plot_backtest
+
+# Load symbols data
+symbols = pd.read_csv('data/Ticker_List_NSE_India.csv')
+
+# Initialize the app with a Bootstrap theme
+app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
+
+# Function to create the layout
+app.layout = dbc.Container([
+    dbc.Row([
+        html.H1("Algorithmic Trading Dashboard", className="text-center mb-4")
+    ]),
+
+    dbc.Row([
+        html.Label("Select Company Name", className="form-label"),
+        dcc.Dropdown(
+            id="name",
+            options=[{"label": name, "value": name} for name in symbols['NAME OF COMPANY'].unique()],
+            value='',
+            placeholder="Select a company",
+            className="mb-3"
+        ),
+
+        html.Label("Select Strategy", className="form-label"),
+        dcc.Dropdown(
+            id="strategy",
+            options=['SMC', 'SMC with EMA'],
+            value='',
+            placeholder="Select Strategy",
+            className="mb-3"
+        ),
+
+        html.Label("Swing High/Low Window Size", className="form-label"),
+        dcc.Input(
+            id="window",
+            type="number",
+            value=10,
+            placeholder="Enter window size",
+            className="form-control mb-3"
+        ),
+
+    ]),
+
+    html.Div([
+        dbc.Row([
+            dbc.Col([
+                html.Label("Fast EMA Length: ", className="form-label"),
+                dcc.Input(
+                    id="ema1",
+                    type="number",
+                    value=9,
+                    placeholder="Enter EMA Length",
+                    # className="form-control mb-3"
+                    className = "text-nowrap"
+                ),
+            ], md=8),
+
+            dbc.Col([
+                html.Label("Slow EMA Length: ", className="form-label"),
+                dcc.Input(
+                    id="ema2",
+                    type="number",
+                    value=21,
+                    placeholder="Enter EMA size",
+                    # className="form-control mb-3"
+                    className="text-nowrap"
+                ),
+
+            dbc.Col([dcc.Checklist(['Close on EMA crossover'], id='closecross', className="text-nowrap")]),
+
+            ]),
+        ]),
+        ], style={'display': 'block'}, id='smc_ema'
+    ),
+
+    dbc.Button("Run", id="submit-button", color="primary", className="w-100 mb-4"),
+
+    dbc.Row([
+        html.H5("Order Block Chart", className="text-center mb-3"),
+        html.Iframe(
+            src="assets/SMC.html",
+            style={"height": "450px", "width": "95%", "border": "none"},
+            className="mb-4"
+        ),
+
+        html.H5("Backtest Results", className="text-center mb-3"),
+        html.Iframe(
+            src="assets/backtest_results.html",
+            style={"height": "1067px", "width": "95%", "border": "none"}
+        ),
+    ])
+], fluid=True)
+
+
+@callback(
+    Output("smc_ema", 'style'),
+    Input("strategy", 'value')
+)
+def update_layout(strategy):
+    if strategy=='SMC with EMA':
+        return {'display':'block'}
+    else:
+        return {'display':'none'}
+
+
+# Callback for updating the visualizations
+@callback(
+    Input("submit-button", "n_clicks"),
+    State("name", "value"),
+    State("window", "value"),
+    State("strategy", "value"),
+    State("ema1", "value"),
+    State("ema2", "value"),
+    State("closecross", "value")
+)
+def update_visuals(n_clicks, name, window, strategy, ema1, ema2, closecross):
+    if n_clicks <= 0 or not name:
+        return
+
+    # Clear existing files
+    open('assets/backtest_results.html', 'w').close()
+    open('assets/SMC.html', 'w').close()
+
+    ticker = symbols[symbols['NAME OF COMPANY'] == name]['YahooEquiv'].values[0]
+    data = fetch(ticker, '1mo', '15m')
+
+    fig = SMC(data=data, swing_hl_window_sz=window).plot(show=False).update_layout(title=dict(text=ticker))
+
+    print(strategy)
+    if strategy=='SMC':
+        smc_plot_backtest(data, 'assets/backtest_results.html', swing_hl=window)
+    elif strategy=='SMC with EMA':
+        smc_ema_plot_backtest(data, 'assets/backtest_results.html', ema1, ema2, closecross)
+
+    fig.write_html('assets/SMC.html')
+
+if __name__ == "__main__":
+    # Clear initial files
+    open('assets/backtest_results.html', 'w').close()
+    open('assets/SMC.html', 'w').close()
+
+    app.run(debug=True)

data_fetcher.py

@@ -0,0 +1,6 @@
+import yfinance as yf
+
+def fetch(symbol, period, interval):
+    df = yf.download(symbol, period=period, interval=interval)
+    df.columns =df.columns.get_level_values(0)
+    return df

gradio_app.py

@@ -0,0 +1,65 @@
+import gradio as gr
+from data_fetcher import fetch
+from indicators import SMC
+from strategies import smc_plot_backtest, smc_ema_plot_backtest
+import pandas as pd
+
+symbols = pd.read_csv('data/Ticker_List_NSE_India.csv')
+
+def run(stock, strategy, swing_hl, ema1, ema2, cross_close):
+    ticker =  symbols[symbols['NAME OF COMPANY'] == stock]['YahooEquiv'].values[0]
+    data = fetch(ticker, period='1mo', interval='15m')
+
+    smc_plot_backtest(data, 'test.html', swing_hl)
+
+    signal_plot = SMC(data=data, swing_hl_window_sz=swing_hl).plot(show=False).update_layout(title=dict(text=ticker))
+    backtest_plot = gr.Plot()
+
+    if strategy == "SMC":
+        backtest_plot = smc_plot_backtest(data, 'test.html', swing_hl)
+
+    if strategy == "SMC with EMA":
+        backtest_plot = smc_ema_plot_backtest(data, 'test.html', ema1, ema2, cross_close)
+
+    return signal_plot, backtest_plot
+
+
+with gr.Blocks(fill_width=True) as app:
+    gr.Markdown(
+        '# Algorithmic Trading Dashboard'
+    )
+    stock = gr.Dropdown(symbols['NAME OF COMPANY'].unique().tolist(), label='Select Company', value=None)
+
+    with gr.Row():
+        strategy = gr.Dropdown(['SMC', 'SMC with EMA'], label='Strategy', value=None)
+        swing_hl = gr.Number(label="Swing High/Low Window Size", value=10, interactive=True)
+
+    @gr.render(inputs=[strategy])
+    def show_extra(strat):
+        if strat == "SMC with EMA":
+            with gr.Row():
+                ema1 = gr.Number(label='Fast EMA length', value=9)
+                ema2 = gr.Number(label='Slow EMA length', value=21)
+                cross_close = gr.Checkbox(label='Close trade on EMA crossover')
+            input = [stock, strategy, swing_hl, ema1, ema2, cross_close]
+
+        elif strat == "SMC":
+            input = [stock, strategy, swing_hl]
+        else:
+            input = []
+
+        btn.click(
+            run,
+            inputs=input,
+            outputs=[signal_plot, backtest_plot]
+        )
+
+    btn = gr.Button("Run")
+
+    with gr.Row():
+        signal_plot = gr.Plot(label='Signal plot')
+
+    with gr.Row():
+        backtest_plot = gr.Plot(label='Backtesting plot')
+
+app.launch(debug=True)

indicators.py

@@ -0,0 +1,252 @@
+import pandas as pd
+import numpy as np
+import plotly.graph_objects as go
+from plotly.subplots import make_subplots
+
+class SMC:
+    def __init__(self, data, swing_hl_window_sz=10):
+        self.data = data
+        self.data['Date'] = self.data.index.to_series()
+        self.swing_hl_window_sz = swing_hl_window_sz
+        self.order_blocks = self.order_block()
+
+    def backtest_buy_signal(self):
+        bull_ob = self.order_blocks[(self.order_blocks['OB']==1) & (self.order_blocks['MitigatedIndex']!=0)]
+        arr = np.zeros(len(self.data))
+        arr[bull_ob['MitigatedIndex'].apply(lambda x: int(x))] = 1
+        return arr
+
+    def backtest_sell_signal(self):
+        bear_ob = self.order_blocks[(self.order_blocks['OB'] == -1) & (self.order_blocks['MitigatedIndex'] != 0)]
+        arr = np.zeros(len(self.data))
+        arr[bear_ob['MitigatedIndex'].apply(lambda x: int(x))] = -1
+        return arr
+
+    def swing_highs_lows(self, window_size):
+        l = self.data['Low'].reset_index(drop=True)
+        h = self.data['High'].reset_index(drop=True)
+        swing_highs = (h.rolling(window_size, center=True).max() / h == 1.)
+        swing_lows = (l.rolling(window_size, center=True).min() / l == 1.)
+        return pd.DataFrame({'Date':self.data.index.to_series(), 'highs':swing_highs.values, 'lows':swing_lows.values})
+
+    def fvg(self):
+        """
+        FVG - Fair Value Gap
+        A fair value gap is when the previous high is lower than the next low if the current candle is bullish.
+        Or when the previous low is higher than the next high if the current candle is bearish.
+
+        parameters:
+
+        returns:
+        FVG = 1 if bullish fair value gap, -1 if bearish fair value gap
+        Top = the top of the fair value gap
+        Bottom = the bottom of the fair value gap
+        MitigatedIndex = the index of the candle that mitigated the fair value gap
+        """
+
+        fvg = np.where(
+            (
+                    (self.data["High"].shift(1) < self.data["Low"].shift(-1))
+                    & (self.data["Close"] > self.data["Open"])
+            )
+            | (
+                    (self.data["Low"].shift(1) > self.data["High"].shift(-1))
+                    & (self.data["Close"] < self.data["Open"])
+            ),
+            np.where(self.data["Close"] > self.data["Open"], 1, -1),
+            np.nan,
+        )
+
+        top = np.where(
+            ~np.isnan(fvg),
+            np.where(
+                self.data["Close"] > self.data["Open"],
+                self.data["Low"].shift(-1),
+                self.data["Low"].shift(1),
+            ),
+            np.nan,
+        )
+
+        bottom = np.where(
+            ~np.isnan(fvg),
+            np.where(
+                self.data["Close"] > self.data["Open"],
+                self.data["High"].shift(1),
+                self.data["High"].shift(-1),
+            ),
+            np.nan,
+        )
+
+        mitigated_index = np.zeros(len(self.data), dtype=np.int32)
+        for i in np.where(~np.isnan(fvg))[0]:
+            mask = np.zeros(len(self.data), dtype=np.bool_)
+            if fvg[i] == 1:
+                mask = self.data["Low"][i + 2:] <= top[i]
+            elif fvg[i] == -1:
+                mask = self.data["High"][i + 2:] >= bottom[i]
+            if np.any(mask):
+                j = np.argmax(mask) + i + 2
+                mitigated_index[i] = j
+
+        mitigated_index = np.where(np.isnan(fvg), np.nan, mitigated_index)
+
+        return pd.concat(
+            [
+                pd.Series(fvg.flatten(), name="FVG"),
+                pd.Series(top.flatten(), name="Top"),
+                pd.Series(bottom.flatten(), name="Bottom"),
+                pd.Series(mitigated_index.flatten(), name="MitigatedIndex"),
+            ],
+            axis=1,
+        )
+
+    def order_block(self, imb_perc=.1, join_consecutive=True):
+        hl = self.swing_highs_lows(self.swing_hl_window_sz)
+
+        ob = np.where(
+            (
+                    ((self.data["High"]*((100+imb_perc)/100)) < self.data["Low"].shift(-2))
+                    & ((hl['lows']==True) | (hl['lows'].shift(1)==True))
+            )
+            | (
+                    (self.data["Low"] > (self.data["High"].shift(-2)*((100+imb_perc)/100)))
+                    & ((hl['highs']==True) | (hl['highs'].shift(1)==True))
+            ),
+            np.where(((hl['lows']==True) | (hl['lows'].shift(1)==True)), 1, -1),
+            np.nan,
+        )
+
+        # print(ob)
+
+        top = np.where(
+            ~np.isnan(ob),
+            np.where(
+                self.data["Close"] > self.data["Open"],
+                self.data["Low"].shift(-2),
+                self.data["Low"],
+            ),
+            np.nan,
+        )
+
+        bottom = np.where(
+            ~np.isnan(ob),
+            np.where(
+                self.data["Close"] > self.data["Open"],
+                self.data["High"],
+                self.data["High"].shift(-2),
+            ),
+            np.nan,
+        )
+
+        # if join_consecutive:
+        #     for i in range(len(ob) - 1):
+        #         if ob[i] == ob[i + 1]:
+        #             top[i + 1] = max(top[i], top[i + 1])
+        #             bottom[i + 1] = min(bottom[i], bottom[i + 1])
+        #             ob[i] = top[i] = bottom[i] = np.nan
+
+        mitigated_index = np.zeros(len(self.data), dtype=np.int32)
+        for i in np.where(~np.isnan(ob))[0]:
+            mask = np.zeros(len(self.data), dtype=np.bool_)
+            if ob[i] == 1:
+                mask = self.data["Low"][i + 3:] <= top[i]
+            elif ob[i] == -1:
+                mask = self.data["High"][i + 3:] >= bottom[i]
+            if np.any(mask):
+                j = np.argmax(mask) + i + 3
+                mitigated_index[i] = int(j)
+        ob = ob.flatten()
+        mitigated_index1 = np.where(np.isnan(ob), np.nan, mitigated_index)
+
+        return pd.concat(
+            [
+                pd.Series(ob.flatten(), name="OB"),
+                pd.Series(top.flatten(), name="Top"),
+                pd.Series(bottom.flatten(), name="Bottom"),
+                pd.Series(mitigated_index1.flatten(), name="MitigatedIndex"),
+            ],
+            axis=1,
+        ).dropna(subset=['OB'])
+
+    def plot(self, swing_hl=True, show=True):
+        fig = make_subplots(1, 1)
+
+        # plot the candle stick graph
+        fig.add_trace(go.Candlestick(x=self.data.index.to_series(),
+                        open=self.data['Open'],
+                        high=self.data['High'],
+                        low=self.data['Low'],
+                        close=self.data['Close'],
+                        name='ohlc'))
+
+        # grab first and last observations from df.date and make a continuous date range from that
+        dt_all = pd.date_range(start=self.data['Date'].iloc[0], end=self.data['Date'].iloc[-1], freq='5min')
+
+        # check which dates from your source that also accur in the continuous date range
+        dt_obs = [d.strftime("%Y-%m-%d %H:%M:%S") for d in self.data['Date']]
+
+        # isolate missing timestamps
+        dt_breaks = [d for d in dt_all.strftime("%Y-%m-%d %H:%M:%S").tolist() if not d in dt_obs]
+
+        # adjust xaxis for rangebreaks
+        fig.update_xaxes(rangebreaks=[dict(dvalue=5 * 60 * 1000, values=dt_breaks)])
+
+        print(self.order_blocks.head())
+        print(self.order_blocks.index.to_list())
+
+        ob_df = self.data.iloc[self.order_blocks.index.to_list()]
+        # print(ob_df)
+
+        fig.add_trace(go.Scatter(
+                        x=ob_df['Date'],
+                        y=ob_df['Low'],
+                        name="Order Block",
+                        mode='markers',
+                        marker_symbol='diamond-dot',
+                        marker_size=13,
+                        marker_line_width=2,
+                        # offsetgroup=0,
+                       ))
+
+        if swing_hl:
+            hl = self.swing_highs_lows(self.swing_hl_window_sz)
+            h = hl[(hl['highs']==True)]
+            l = hl[hl['lows']==True]
+            # print(h)
+            # exit(0)
+            fig.add_trace(go.Scatter(
+                x=h['Date'],
+                y=self.data[self.data.Date.isin(h['Date'])]['High']*(100.1/100),
+                mode='markers',
+                marker_symbol="triangle-up-dot",
+                marker_size=10,
+                name='Swing High',
+                # offsetgroup=2,
+            ))
+            fig.add_trace(go.Scatter(
+                x=l['Date'],
+                y=self.data[self.data.Date.isin(l['Date'])]['Low']*(99.9/100),
+                mode='markers',
+                marker_symbol="triangle-down-dot",
+                marker_size=10,
+                name='Swing Low',
+                marker_color='red',
+                # offsetgroup=2,
+            ))
+
+        fig.update_layout(xaxis_rangeslider_visible=False)
+        if show:
+            fig.show()
+        return fig
+
+
+def EMA(array, n):
+    return pd.Series(array).ewm(span=n, adjust=False).mean()
+
+if __name__ == "__main__":
+    from data_fetcher import fetch
+    data = fetch('ICICIBANK.NS', period='1mo', interval='15m')
+    # data = fetch('RELIANCE.NS', period='1mo', interval='15m')
+
+    # print(SMC(data).backtest_buy_signal())
+    SMC(data).plot()

requirements.txt

@@ -0,0 +1,8 @@
+backtesting==0.3.3
+numpy==2.2.0
+pandas==2.2.3
+bokeh==3.1.0
+yfinance==0.2.50
+plotly==5.24.1
+dash==2.18.2
+dash-bootstrap-components==1.6.0

strategies.py

@@ -0,0 +1,108 @@
+from backtesting import Backtest, Strategy
+from backtesting.lib import SignalStrategy, TrailingStrategy
+from indicators import SMC, EMA
+from data_fetcher import fetch
+import pandas as pd
+
+class SMC_test(Strategy):
+    swing_hl = 10
+    def init(self):
+        super().init()
+
+        self.smc_b = self.I(self.smc_buy, data=self.data.df, swing_hl=self.swing_hl)
+        self.smc_s = self.I(self.smc_sell, data=self.data.df, swing_hl=self.swing_hl)
+
+
+    def next(self):
+        price = self.data.Close[-1]
+        current_time = self.data.index[-1]
+
+        if self.smc_b[-1] == 1:
+            self.buy(sl=.95 * price, tp=1.05 * price)
+        if self.smc_s[-1] == -1:
+            self.sell(tp=.95 * price, sl=1.05 * price)
+
+        # Additionally, set aggressive stop-loss on trades that have been open
+        # for more than two days
+        for trade in self.trades:
+            if current_time - trade.entry_time > pd.Timedelta('2 days'):
+                if trade.is_long:
+                    trade.sl = max(trade.sl, self.data.Low[-1])
+                else:
+                    trade.sl = min(trade.sl, self.data.High[-1])
+
+
+    def smc_buy(self, data, swing_hl):
+        return SMC(data, swing_hl).backtest_buy_signal()
+
+    def smc_sell(self, data, swing_hl):
+        return SMC(data, swing_hl).backtest_sell_signal()
+
+class SMC_ema(SignalStrategy, TrailingStrategy):
+    ema1 = 9
+    ema2 = 21
+    close_on_crossover = False
+
+    def init(self):
+        super().init()
+
+        self.smc_b = self.I(self.smc_buy, self.data.df)
+        self.smc_s = self.I(self.smc_sell, self.data.df)
+
+
+        close = self.data.Close
+
+        self.ma1 = self.I(EMA, close, self.ema1)
+        self.ma2 = self.I(EMA, close, self.ema2)
+
+
+    def next(self):
+        price = self.data.Close[-1]
+        current_time = self.data.index[-1]
+
+        if self.smc_b[-1] == 1 and self.ma1 > self.ma2:
+            self.buy(sl=.95 * price, tp=1.05 * price)
+        if self.smc_s[-1] == -1 and self.ma1 < self.ma2:
+            self.sell(tp=.95 * price, sl=1.05 * price)
+
+        # Additionally, set aggressive stop-loss on trades that have been open
+        # for more than two days
+        for trade in self.trades:
+            if current_time - trade.entry_time > pd.Timedelta('2 days'):
+                if trade.is_long:
+                    trade.sl = max(trade.sl, self.data.Low[-1])
+                else:
+                    trade.sl = min(trade.sl, self.data.High[-1])
+
+        # Close the trade if there is a moving average crossover in opposite direction
+        if self.close_on_crossover:
+            for trade in self.trades:
+                if trade.is_long and self.ma1 < self.ma2:
+                    trade.close()
+                if trade.is_short and self.ma1 > self.ma2:
+                    trade.close()
+
+    def smc_buy(self, data):
+        return SMC(data).backtest_buy_signal()
+
+    def smc_sell(self, data):
+        return SMC(data).backtest_sell_signal()
+
+def smc_plot_backtest(data, filename, swing_hl, **kwargs):
+    bt = Backtest(data, SMC_test, **kwargs)
+    bt.run(swing_hl=swing_hl)
+    print('runned')
+    return bt.plot(filename=filename, open_browser=False)
+
+def smc_ema_plot_backtest(data, filename, ema1, ema2, closecross, **kwargs):
+    bt = Backtest(data, SMC_ema, **kwargs)
+    bt.run(ema1=ema1, ema2=ema2, close_on_crossover=closecross)
+    return bt.plot(filename=filename, open_browser=False)
+
+
+if __name__ == "__main__":
+    data = fetch('ICICIBANK.NS', period='1mo', interval='15m')
+    bt = Backtest(data, SMC_ema, commission=.002)
+
+    bt.run(ema1 = 9, ema2 = 21, close_on_crossover=True)
+    bt.plot()