78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""
|
||
Telegram alert system. All sends are fire-and-forget — a send failure
|
||
never crashes the main loop.
|
||
"""
|
||
|
||
import logging
|
||
|
||
from . import config
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_bot = None
|
||
_chat_id = None
|
||
|
||
|
||
def _get_bot():
|
||
global _bot, _chat_id
|
||
if _bot is None and config.TELEGRAM_BOT_TOKEN and config.TELEGRAM_CHAT_ID:
|
||
try:
|
||
from telegram import Bot
|
||
_bot = Bot(token=config.TELEGRAM_BOT_TOKEN)
|
||
_chat_id = config.TELEGRAM_CHAT_ID
|
||
except ImportError:
|
||
logger.warning("python-telegram-bot not installed — Telegram disabled")
|
||
return _bot
|
||
|
||
|
||
async def send(text: str) -> None:
|
||
bot = _get_bot()
|
||
if bot is None:
|
||
return
|
||
try:
|
||
await bot.send_message(chat_id=_chat_id, text=text, parse_mode="HTML")
|
||
except Exception as exc:
|
||
logger.warning("Telegram send failed: %s", exc)
|
||
|
||
|
||
async def notify_open(symbol: str, direction: str, entry: float,
|
||
sl: float, tp: float, lots: float) -> None:
|
||
arrow = "🟢 BUY" if direction == "long" else "🔴 SELL"
|
||
mode = "PAPER" if config.PAPER_TRADING else "LIVE"
|
||
text = (
|
||
f"<b>[{mode}] {arrow} {symbol}</b>\n"
|
||
f"Entry : <code>{entry:.5f}</code>\n"
|
||
f"SL : <code>{sl:.5f}</code>\n"
|
||
f"TP : <code>{tp:.5f}</code>\n"
|
||
f"Size : {lots:.2f} lots\n"
|
||
f"R:R : 1 : {config.TP_ATR / config.SL_ATR:.1f}"
|
||
)
|
||
logger.info("OPEN %s %s @ %.5f SL=%.5f TP=%.5f %.2f lots",
|
||
symbol, direction.upper(), entry, sl, tp, lots)
|
||
await send(text)
|
||
|
||
|
||
async def notify_close(symbol: str, direction: str, entry: float,
|
||
exit_price: float, pnl_pips: float, reason: str) -> None:
|
||
sign = "✅" if pnl_pips > 0 else "❌"
|
||
text = (
|
||
f"<b>{sign} CLOSED {symbol}</b>\n"
|
||
f"Dir : {direction.upper()}\n"
|
||
f"Entry : <code>{entry:.5f}</code> → <code>{exit_price:.5f}</code>\n"
|
||
f"P&L : <b>{pnl_pips:+.1f} pips</b>\n"
|
||
f"Reason: {reason}"
|
||
)
|
||
logger.info("CLOSE %s %s %+.1f pips (%s)", symbol, direction.upper(),
|
||
pnl_pips, reason)
|
||
await send(text)
|
||
|
||
|
||
async def notify_status(message: str) -> None:
|
||
logger.info(message)
|
||
await send(f"ℹ️ {message}")
|
||
|
||
|
||
async def notify_error(message: str) -> None:
|
||
logger.error(message)
|
||
await send(f"🚨 <b>ERROR</b>: {message}")
|