β ΠΠ°Π·Π°Π΄"""
Grid Bot β Telegram Bot v2
=============================
Commands for monitoring and control.
Updated for v2 (inventory, spacing, chop).
"""
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from src.config import (
TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID,
GRID_LEVELS, GRID_SPACING_PCT, LEVERAGE, POSITION_SIZE_USD,
)
logger = logging.getLogger("telegram")
class TelegramBot:
def __init__(self, grid_bot=None):
self.grid_bot = grid_bot
self.app = None
self._chat_id = TELEGRAM_CHAT_ID
async def start(self):
if not TELEGRAM_BOT_TOKEN:
logger.warning("Telegram disabled (no token)")
return
self.app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
self.app.add_handler(CommandHandler("start", self._cmd_start))
self.app.add_handler(CommandHandler("help", self._cmd_help))
self.app.add_handler(CommandHandler("status", self._cmd_status))
self.app.add_handler(CommandHandler("grid", self._cmd_grid))
self.app.add_handler(CommandHandler("balance", self._cmd_balance))
self.app.add_handler(CommandHandler("bal", self._cmd_balance))
self.app.add_handler(CommandHandler("pnl", self._cmd_pnl))
self.app.add_handler(CommandHandler("scan", self._cmd_scan))
self.app.add_handler(CommandHandler("stop", self._cmd_stop))
await self.app.initialize()
await self.app.start()
await self.app.updater.start_polling(drop_pending_updates=True)
logger.info("Telegram bot started")
async def stop(self):
if self.app:
await self.app.updater.stop()
await self.app.stop()
await self.app.shutdown()
async def send_message(self, text):
if not self.app or not self._chat_id:
return
try:
await self.app.bot.send_message(
chat_id=self._chat_id, text=text, parse_mode="Markdown",
)
except Exception as e:
logger.error(f"Send message error: {e}")
try:
await self.app.bot.send_message(chat_id=self._chat_id, text=text)
except Exception:
pass
# ============================================================
# COMMANDS
# ============================================================
async def _cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
f"π€ *Grid Bot v2*\n"
f"Auto-rotating grid + inventory mgmt\n"
f"Spacing: ~{GRID_SPACING_PCT}% ATR | "
f"Levels: {GRID_LEVELS}+{GRID_LEVELS} | {LEVERAGE}x\n\n"
f"/help β commands",
parse_mode="Markdown",
)
async def _cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"π *Commands:*\n\n"
"/status β bot status\n"
"/grid β active grid details\n"
"/balance (/bal) β USDT balance\n"
"/pnl β daily PnL summary\n"
"/scan β screener top coins\n"
"/stop β stop all grids\n",
parse_mode="Markdown",
)
async def _cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.grid_bot:
await update.message.reply_text("No bot connected")
return
grid_status = self.grid_bot.grid_manager.get_status()
paused, reason = self.grid_bot.risk.is_paused()
status_str = f"βΈοΈ {reason}" if paused else "β
Running"
await update.message.reply_text(
f"π€ *Grid Bot v2 Status*\n\n"
f"Status: {status_str}\n"
f"{grid_status}",
parse_mode="Markdown",
)
async def _cmd_grid(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.grid_bot:
await update.message.reply_text("No bot connected")
return
sessions = self.grid_bot.grid_manager.sessions
if not sessions:
await update.message.reply_text("π No active grids")
return
for symbol, session in sessions.items():
s = session.get_session_summary()
open_orders = session.get_open_order_count()
waiting = session.get_filled_unpaired_count()
emoji = "π’" if s["net_pnl"] >= 0 else "π΄"
inv = s['inventory_imbalance']
inv_str = f"+{inv}L" if inv > 0 else f"{inv}S" if inv < 0 else "0"
await update.message.reply_text(
f"π *{symbol} Grid*\n\n"
f"Center: ${s['center_price']:.4f}\n"
f"Spacing: {s['spacing_pct']:.2f}%\n"
f"π Round-trips: {s['round_trips']}\n"
f"{emoji} PnL: ${s['net_pnl']:+.4f}\n"
f"π¦ Inventory: {inv_str} (peak: {s['peak_inventory']})\n"
f"π§ Partial: {s['partial_closes']} | Unstuck: {s['unstuck_closes']}\n"
f"β±οΈ Duration: {s['duration_min']:.0f}min\n"
f"π Open: {open_orders} | Waiting: {waiting}",
parse_mode="Markdown",
)
async def _cmd_balance(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.grid_bot:
await update.message.reply_text("No bot connected")
return
try:
balance = self.grid_bot.exchange.get_balance()
avail = self.grid_bot.exchange.get_available_balance()
await update.message.reply_text(
f"π° Balance: ${balance:.2f}\n"
f"π΅ Available: ${avail:.2f}",
parse_mode="Markdown",
)
except Exception as e:
await update.message.reply_text(f"Error: {e}")
async def _cmd_pnl(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.grid_bot:
await update.message.reply_text("No bot connected")
return
msg = self.grid_bot.risk.get_daily_summary()
await update.message.reply_text(msg, parse_mode="Markdown")
async def _cmd_scan(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.grid_bot:
await update.message.reply_text("No bot connected")
return
self.grid_bot.screener.scan()
msg = self.grid_bot.screener.get_scan_summary()
await update.message.reply_text(msg, parse_mode="Markdown")
async def _cmd_stop(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.grid_bot:
await update.message.reply_text("No bot connected")
return
sessions = list(self.grid_bot.grid_manager.sessions.keys())
if not sessions:
await update.message.reply_text("π No active grids to stop")
return
for sym in sessions:
summary = self.grid_bot.grid_manager.stop_grid(sym, reason="manual_stop")
if summary:
self.grid_bot.risk.on_session_closed(summary)
await update.message.reply_text(
f"π Stopped {len(sessions)} grid(s): {', '.join(sessions)}"
)