mirror of
https://github.com/asavinov/intelligent-trading-bot.git
synced 2026-05-04 16:26:44 +00:00
108 lines
4.2 KiB
Python
108 lines
4.2 KiB
Python
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
|