File size: 6,561 Bytes
aab0a9b a047e89 aab0a9b 026493b aab0a9b 026493b aab0a9b d3b64d7 aab0a9b d3b64d7 aab0a9b a047e89 aab0a9b a047e89 aab0a9b a047e89 aab0a9b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
from backtesting import Backtest, Strategy
from backtesting.lib import SignalStrategy, TrailingStrategy
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):
super().init()
# Setting smc buy and sell indicators.
self.smc_b = self.I(self.smc_buy, data=self.data.df, swing_hl=self.swing_window)
self.smc_s = self.I(self.smc_sell, data=self.data.df, swing_hl=self.swing_window)
def next(self):
price = self.data.Close[-1]
current_time = self.data.index[-1]
# If buy signal, set target 5% above price and stoploss 5% below price.
if self.smc_b[-1] == 1:
self.buy(sl=.95 * price, tp=1.05 * price)
# If sell signal, set targe 5% below price and stoploss 5% above 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_ob()
def smc_sell(self, data, swing_hl):
return SMC(data, swing_hl).backtest_sell_signal_ob()
class SMC_ema(SignalStrategy, TrailingStrategy):
swing_window = 10
ema1 = 9
ema2 = 21
close_on_crossover = False
def init(self):
super().init()
# Setting smc buy and sell indicators.
self.smc_b = self.I(self.smc_buy, data=self.data.df, swing_hl=self.swing_window)
self.smc_s = self.I(self.smc_sell, data=self.data.df, swing_hl=self.swing_window)
close = self.data.Close
# Setting up EMAs.
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 buy signal and short moving average is above long moving average.
if self.smc_b[-1] == 1 and self.ma1 > self.ma2:
self.buy(sl=.95 * price, tp=1.05 * price)
# If sell signal and short moving average is below long moving average.
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, swing_hl):
return SMC(data, swing_hl).backtest_buy_signal_ob()
def smc_sell(self, data, swing_hl):
return SMC(data, swing_hl).backtest_sell_signal_ob()
class SMCStructure(TrailingStrategy):
swing_window = 20
def init(self):
super().init()
self.smc_b = self.I(self.smc_buy, data=self.data.df, swing_hl=self.swing_window)
self.smc_s = self.I(self.smc_sell, data=self.data.df, swing_hl=self.swing_window)
self.set_trailing_sl(2)
# self.swing = self.I(self.nearest_swing, data=self.data.df, swing_hl)
def next(self):
price = self.data.Close[-1]
current_time = self.data.index[-1]
if self.smc_b[-1] == 1:
nearest = self.nearest_swing(self.data.df, self.swing_window)
target = price + ((price - nearest)* .414)
stoploss = price - (target-price)
# print(f"buy: {current_time}, {price}, {nearest}, {target}, {stoploss}")
try:
self.buy(sl=stoploss, tp=target)
except:
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)
if nearest > price:
target = price - ((nearest - price) * .414)
stoploss = price + (price - target)
# print(f"sell: {current_time}, {price}, {nearest}, {target}, {stoploss}")
try:
self.sell(sl=stoploss, tp=target, limit=float(price))
except:
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
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_structure()
def smc_sell(self, data, swing_hl):
return SMC(data, swing_hl).backtest_sell_signal_structure()
def nearest_swing(self, data, swing_hl):
# Get swing high/low nearest to current price.
swings = SMC(data, swing_hl).swing_hl
swings = swings[~np.isnan(swings['Level'])]
return swings['Level'].iloc[-2]
strategies = {'Order Block': SMC_test, 'Order Block with EMA': SMC_ema , 'Structure trading': SMCStructure}
if __name__ == "__main__":
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')
# bt = Backtest(data, SMC_ema, commission=.002)
# bt.run(ema1 = 9, ema2 = 21, close_on_crossover=True)
bt = Backtest(data, SMCStructure, commission = .002, trade_on_close=True)
print(bt.run())
# bt.plot() |