Price Action EMA + RSI + Bollinger Bands With Bots Testing

fastEMA:30 slowEMA:50 SOLUSDT SL=2.5, TP=1.8, return=13.29% in 1h.
fastEMA:30 slowEMA:50 SOLUSDT SL=2.5, TP=1.3, return=4.96% in 30m.
fastEMA:30 slowEMA:50 SOLUSDT SL=1.2, TP=1.1, return=-1.01% in 15m.
fastEMA:30 slowEMA:50 SOLUSDT SL=1.5, TP=2.5, return=-0.42% in 5m.
fastEMA:30 slowEMA:50 SOLUSDT SL=1.1, TP=1.4, return=0.66% in 4h.
fastEMA:20 slowEMA:40 SOLUSDT SL=2.4, TP=2.2, return=6.70% in 1h.
fastEMA:50 slowEMA:100 SOLUSDT SL=2.0, TP=2.5, return=8.82% in 1h.

fastEMA:30 slowEMA:50 ETHUSDT SL=2.1, TP=2.4, return=10.72% in 1h.
fastEMA:30 slowEMA:50 ETHUSDT SL=1.8, TP=1.3, return=1.06% in 30m.
fastEMA:30 slowEMA:50 ETHUSDT SL=1.1, TP=1.7, return=0.10% in 15m.
fastEMA:30 slowEMA:50 ETHUSDT SL=1.3, TP=2.5, return=0.69% in 5m.
fastEMA:30 slowEMA:50 ETHUSDT SL=1.0, TP=1.3, return=3.97% in 4h.
fastEMA:20 slowEMA:40 ETHUSDT SL=2.2, TP=2.4, return=9.95% in 1h.
fastEMA:50 slowEMA:100 ETHUSDT SL=1.3, TP=2.5, return=-2.24% in 1h.

def count_opened_trades():
    api_instance = c.API(access_token)
    config = api_instance.config
    api_key = config.get(“api_key_bybit”)
    api_secret = config.get(“api_secret_bybit”)
    session = HTTP(testnet=False, api_key=api_key, api_secret=api_secret)
    try:
        data = session.get_positions(category=”linear”, symbol=Symbol)
        size = data[‘result’][‘list’][0][‘size’]
        return float(size)
    except Exception as e:
        print(f”DGM Failed to get {Symbol} position: {e}”)
        return

def ema_signal(df, current_candle, backcandles):  
    df_slice = df.reset_index().copy()
    start = max(0, current_candle – backcandles)
    end = current_candle + 1
    relevant_rows = df_slice.iloc[start:end]
   
    # Check if all EMA_fast values are below EMA_slow values (buy signal)
    if (relevant_rows[‘EMA_fast’] < relevant_rows[‘EMA_slow’]).all():
        return 1
    # Check if all EMA_fast values are above EMA_slow values (sell signal)
    elif (relevant_rows[‘EMA_fast’] > relevant_rows[‘EMA_slow’]).all():
        return -1
    else:
        return 0
   
def total_signal(df, current_candle, backcandles):
    if isinstance(current_candle, pd.Timestamp):
        current_candle = df.index.get_loc(current_candle)

    ema_signal_result = ema_signal(df, current_candle, backcandles)

    candle_open_price = df[‘Open’].iloc[current_candle]
    bbl = df[‘BBL_15_1.5’].iloc[current_candle]
    bbu = df[‘BBU_15_1.5’].iloc[current_candle]

    if ema_signal_result == 1 and candle_open_price <= bbl:
        return 1
    if ema_signal_result == -1 and candle_open_price >= bbu:
        return -1
    return 0

def get_candles(symbol, interval, lookback):
    url = f”{API_URLv3}{library}?symbol={symbol}&interval={interval}&limit={lookback}”
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()

        if not data:
            print(f”No data received from {API_URLv3}{library} API.”)
            return None

        df = pd.DataFrame(data, columns=[
            ‘open_time’, ‘open’, ‘high’, ‘low’, ‘close’, ‘volume’,
            ‘close_time’, ‘quote_asset_volume’, ‘number_of_trades’,
            ‘taker_buy_base_asset_volume’, ‘taker_buy_quote_asset_volume’, ‘ignore’
        ])
        df[‘open_time’] = pd.to_datetime(df[‘open_time’], unit=’ms’)
        df.set_index(‘open_time’, inplace=True)
        df.rename(columns={‘open’: ‘Open’, ‘high’: ‘High’, ‘low’: ‘Low’, ‘close’: ‘Close’}, inplace=True)

        df[[‘Open’, ‘High’, ‘Low’, ‘Close’]] = df[[‘Open’, ‘High’, ‘Low’, ‘Close’]].astype(float)
        return df[[‘Open’, ‘High’, ‘Low’, ‘Close’]]
    except requests.RequestException as e:
        print(f”{symbol} Request failed: {e}”)
    except Exception as e:
        print(f”Failed to process {symbol} data: {e}”)
    return None

def get_candles_frame(lookback):
    candles = get_candles(Symbol, Interval, lookback)

    if candles is None:
        print(f”Failed to retrieve {Interval} {Symbol} candle data.”)
        return None

    dfstream = candles.copy()

    dfstream[‘ATR’] = ta.atr(dfstream[‘High’], dfstream[‘Low’], dfstream[‘Close’], length=7)
    dfstream[‘EMA_fast’] = ta.ema(dfstream[‘Close’], length=30)
    dfstream[‘EMA_slow’] = ta.ema(dfstream[‘Close’], length=50)
    dfstream[‘RSI’] = ta.rsi(dfstream[‘Close’], length=10)
    my_bbands = ta.bbands(dfstream[‘Close’], length=15, std=1.5)
    dfstream = dfstream.join(my_bbands)
    if not isinstance(dfstream.index, pd.DatetimeIndex):
        dfstream.index = pd.to_datetime(dfstream.index)

    dfstream[‘TotalSignal’] = dfstream.apply(lambda row: total_signal(dfstream, row.name, 7), axis=1)
    dfstream.to_csv(‘dfstream.csv’)
    return dfstream

def optimization():
    slatrcoef = 0
    TPSLRatio_coef = 0

    dfstream = get_candles_frame(lookback)
    if dfstream is None:
        print(f”No candle data for {Symbol} fitting optimization job.”)
        return

    def SIGNAL():
        return dfstream[‘TotalSignal’]

    class MyStrat(Strategy):
        mysize = 3000
        slcoef = 1.3
        TPSLRatio = 2.5

        def init(self):
            self.signal1 = self.I(SIGNAL)

        def next(self):
            slatr = self.slcoef * self.data.ATR[-1]
            TPSLRatio = self.TPSLRatio

            if self.signal1[-1] == 2 and len(self.trades) == 0:
                sl1 = self.data.Close[-1] – slatr
                tp1 = self.data.Close[-1] + slatr * TPSLRatio
                self.buy(sl=sl1, tp=tp1, size=self.mysize)

            elif self.signal1[-1] == 1 and len(self.trades) == 0:
                sl1 = self.data.Close[-1] + slatr
                tp1 = self.data.Close[-1] – slatr * TPSLRatio
                self.sell(sl=sl1, tp=tp1, size=self.mysize)

    bt = Backtest(dfstream, MyStrat, cash=100000, margin=0.01, commission=0.00055)
    stats, heatmap = bt.optimize(slcoef=[i/10 for i in range(10, 26)],
                                 TPSLRatio=[i/10 for i in range(10, 26)],
                                 maximize=’Return [%]’, max_tries=300,
                                 random_state=0,
                                 return_heatmap=True)
    print(stats)
   
    slatrcoef = stats[“_strategy”].slcoef
    TPSLRatio_coef = stats[“_strategy”].TPSLRatio
    print(f”{Symbol} SL = {slatrcoef}, TP = {TPSLRatio_coef}, expected return, {stats[‘Return [%]’]:.2f}% in {Interval} interval.\n”)
   
    with open(“fitting_data_file.txt”, “a”) as file:
        file.write(f”{Symbol} SL = {slatrcoef}, TP = {TPSLRatio_coef}, expected return, {stats[‘Return [%]’]:.2f}% in {Interval} interval.\n”)
    return slatrcoef, TPSLRatio_coef

def trading_job():
    dfstream = get_candles_frame(lookback)
    if dfstream is None:
        print(f”No {Symbol} candle data for trading job.”)
        return

    signal = total_signal(dfstream, len(dfstream) – 1, 7)



    # now = datetime.now()
    # if now.weekday() == 0 and now.hour < 7 and now.minute < 5:  # Monday before 07:05
    slatrcoef, TPSLRatio_coef = optimization()
    print(f”Optimize SL = {slatrcoef}, and TP = {TPSLRatio_coef}.”)

    slatr = slatrcoef * dfstream[‘ATR’].iloc[-1]
    TPSLRatio = TPSLRatio_coef
    max_spread = 16e-5

    last_candle = get_candles(Symbol, Interval, 1).iloc[-1]
    candle_open_bid = float(last_candle[‘Open’])
    candle_open_ask = candle_open_bid
    spread = candle_open_ask – candle_open_bid

    SLBuy = candle_open_bid – slatr – spread
    SLSell = candle_open_ask + slatr + spread

    TPBuy = candle_open_ask + slatr * TPSLRatio + spread
    TPSell = candle_open_bid – slatr * TPSLRatio – spread

    print(“SLBuy = “, SLBuy)
    print(“SLSell = “, SLSell)
    print(“TPBuy  = “, TPBuy )
    print(“TPSell = “, TPSell)
   
    # # Sell
    # if signal == -1 and count_opened_trades() == 0.0 and spread < max_spread:
    #     print(“Sell Signal Found…”)
    #     trade_crypto = c.TradeCrypto(EXCHANGE, Symbol, ‘sell’)
    #     message, MyTradePrice = trade_crypto.TradeQty(quantity)
    #     print(message)
    #     with open(“trading_data_file.txt”, “a”) as file:
    #         file.write(f”SL = {SLSell}, TP = {TPSell}, Trade Price = {MyTradePrice}\n”)

    # # Buy
    # elif signal == 1 and count_opened_trades() == 0.0 and spread < max_spread:
    #     print(“Buy Signal Found…”)
    #     trade_crypto = c.TradeCrypto(EXCHANGE, Symbol, ‘buy’)
    #     message, MyTradePrice = trade_crypto.TradeQty(quantity)
    #     print(message)
    #     with open(“trading_data_file.txt”, “a”) as file:
    #         file.write(f”SL = {SLBuy}, TP = {TPBuy}, Trade Price = {MyTradePrice}\n”)


if __name__ == “__main__”:
    optimization()

# scheduler = BlockingScheduler()
# scheduler.add_job(trading_job, ‘cron’, day_of_week=’mon-fri’, hour=’07-18′, minute=’1, 6, 11, 16, 21, 26, 31, 36, 41, 46, 51, 56′, timezone=’Asia/Beirut’, misfire_grace_time=15)
# scheduler.start()

In 1993, Buffett spoke to Columbia University’s Business School graduates. Asked about his method for evaluating risk, he said, “Risk comes from not knowing what you’re doing.” This quote reflects Buffett’s investment philosophy, highlighting the crucial role of knowledge and understanding in reducing risk.

The biggest risk is not taking any risk… In a world that changing really quickly, the only strategy that is guaranteed to fail is not taking risks.” Mark Zuckerberg



Tips:

Despite of the crypto dump recently on all the alt coins after SEC announcement to sue Binance and Coinbase. Guess what? My Ai Trading Strategies are making shit ton of USDT from the crazy markets. Well there is a secret and cannot tell you unless…Anyway, I have given you the formula to copy and it is up to you to trade manually with stress and sleepless nights or ride on the trend of Ai trading today ⬇️⬇️⬇️


AI Sleeping Income With DGM System

The SECRET is to marry between Ai trading strategies and an income generated exchange platform

  • Ai trading strategies

  • An income generated exchange platform

How It Works?


LET’S KEEP IN TOUCH!

We’d love to keep you updated with our latest news and offers 😎

We don’t spam! Read our privacy policy for more info.

LUV IT -

Written by 

🎮 Daily Game Moments ⬇️ 🔒 | Register as An Author 🎬 Account To Publish Your Trading Journal Daily. ⬇️⬇️ Why you want to do that? Keep it simple so that we can learn more efficiently and effectively by posting out our weaknesses and failures should be celebrated. Every failure is one step toward your success and DGM can set your course for success. ⬇️⬇️⬇️ ACTION MORE 🕹️ A Day Without Gaming 🤠, Staking 😇 or Trading 🤓 Is A Day Wasted 🏆🎯 @DailyGameMoments has Infinite Possibility

Leave a Reply