Spaces:
Runtime error
Runtime error
| 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() |