intelligent-trading-bot/service/notifier.py

import os
import sys
import asyncio
import requests

from service.App import *
from common.utils import *

import logging
log = logging.getLogger('notifier')


async def notify_telegram():
    await simulate_trade()

    symbol = App.config["symbol"]

    status = App.status
    signal = App.signal
    signal_side = signal.get("side")
    close_price = signal.get('close_price')
    buy_score = signal.get('buy_score')
    sell_score = signal.get('sell_score')
    close_time = signal.get('close_time')

    model = App.config["signal_model"]
    buy_notify_threshold = model["buy_notify_threshold"]
    sell_notify_threshold = model["sell_notify_threshold"]
    buy_signal_threshold = model["buy_signal_threshold"]
    sell_signal_threshold = model["sell_signal_threshold"]
    trade_icon_step = model.get("trade_icon_step", 0)
    notify_frequency_minutes = model.get("notify_frequency_minutes", 1)

    # Crypto Currency Symbols: https://github.com/yonilevy/crypto-currency-symbols
    if symbol == "BTCUSDT":
        symbol_char = "₿"
    elif symbol == "ETHUSDT":
        symbol_char = "Ξ"
    else:
        symbol_char = symbol

    # Notification logic:
    # 1. Trade signal in the case it is suggested to really buy or sell: BUY or SELL and one corresponding score
    # 2. Notification signal simply to provide information (separate criteria): both scores
    # Icons: down: 📉, ⬇ ⬇️🔴 (red), up:  📈, ⬆,  ⬆️ ↗️ 🟢 (green)
    message = ""
    if signal_side == "BUY":
        score_steps = (np.abs(buy_score - buy_signal_threshold) // trade_icon_step) if trade_icon_step else 0
        message = "🟢"*int(score_steps+1) + f" *BUY: {symbol_char} {int(close_price):,} Buy score: {buy_score:+.2f}*"
    elif signal_side == "SELL":
        score_steps = (np.abs(sell_score - sell_signal_threshold) // trade_icon_step) if trade_icon_step else 0
        message = "🔴"*int(score_steps+1) + f" *SELL: {symbol_char} {int(close_price):,} Sell score: {-sell_score:+.2f}*"
    elif (close_time.minute % notify_frequency_minutes) == 0:  # Info message with custom frequency
        if buy_score > sell_score:
            message = f"{symbol_char} {int(close_price):,} 📈{buy_score:+.2f}"
        else:
            message = f"{symbol_char} {int(close_price):,} 📉{-sell_score:+.2f}"
    message = message.replace("+", "%2B")  # For Telegram to display plus sign

    if not message:
        return
    if buy_score < buy_notify_threshold and sell_score < sell_notify_threshold:
        return  # Do not send notifications with low notification threshold (also no buy/sell notifications)

    bot_token = App.config["telegram_bot_token"]
    chat_id = App.config["telegram_chat_id"]

    try:
        url = 'https://api.telegram.org/bot' + bot_token + '/sendMessage?chat_id=' + chat_id + '&parse_mode=markdown&text=' + message
        response = requests.get(url)
        response_json = response.json()
        if not response_json.get('ok'):
            log.error(f"Error sending notification.")
    except Exception as e:
        log.error(f"Error sending notification: {e}")


async def simulate_trade():
    symbol = App.config["symbol"]

    status = App.status
    signal = App.signal
    signal_side = signal.get("side")
    close_price = signal.get('close_price')
    buy_score = signal.get('buy_score')
    sell_score = signal.get('sell_score')
    close_time = signal.get('close_time')

    t_status = App.transaction.get("status")
    t_price = App.transaction.get("price")
    if signal_side == "BUY" and (not t_status or t_status == "BUYING"):
        profit = t_price - close_price if t_price else 0.0
        t_dict = dict(timestamp=str(close_time), price=close_price, profit=profit, status="SELLING")
    elif signal_side == "SELL" and (not t_status or t_status == "SELLING"):
        profit = close_price - t_price if t_price else 0.0
        t_dict = dict(timestamp=str(close_time), price=close_price, profit=profit, status="BUYING")
    else:
        return

    # Save this transaction
    App.transaction = t_dict
    transaction_file = Path("transactions.txt")
    with open(transaction_file, 'a+') as f:
        f.write(",".join([str(v) for v in t_dict.values()]) + "\n")


if __name__ == '__main__':
    pass