Разработка Telegram-интерфейса управления AI-трейдинг-ботом
Telegram-бот — наиболее удобный мобильный интерфейс для управления торговым ботом. Доступен с телефона везде, Push-уведомления в реальном времени, команды для управления.
Функциональность Telegram-интерфейса
Мониторинг
-
/status— текущий P&L, открытые позиции, статус бота -
/positions— таблица открытых позиций -
/trades [last N]— история последних сделок -
/stats [daily|weekly|monthly]— статистика доходности -
/equity— скриншот equity curve
Управление
-
/pause//resume— пауза/возобновление торговли -
/stop— экстренная остановка (с подтверждением) -
/risk {0.5}— изменение множителя риска -
/close {BTCUSDT}— закрыть конкретную позицию
Уведомления (push)
- 🟢 Открытие позиции: символ, направление, цена, размер
- 🔴 Закрытие позиции: P&L по сделке
- ⚠️ Drawdown alert при превышении порога
- 🚨 Ошибки соединения с биржей
Реализация
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ContextTypes
import asyncio
TOKEN = "your_telegram_bot_token"
ALLOWED_USERS = [123456789] # Telegram user IDs
def auth_required(func):
async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.effective_user.id not in ALLOWED_USERS:
await update.message.reply_text("⛔ Unauthorized")
return
return await func(update, context)
return wrapper
@auth_required
async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
metrics = get_bot_metrics()
positions = get_open_positions()
text = f"""
📊 *Bot Status*
Status: {'🟢 Running' if metrics['running'] else '🔴 Paused'}
Daily P&L: `{metrics['daily_pnl']:+.2f}%`
Total P&L: `{metrics['total_pnl']:+.2f}%`
Open Positions: {len(positions)}
*Active Positions:*
"""
for pos in positions:
text += f"• {pos['symbol']}: {pos['side']} {pos['size']} @ {pos['entry']} ({pos['unrealized_pnl']:+.2f}%)\n"
keyboard = [
[InlineKeyboardButton("⏸ Pause", callback_data='pause'),
InlineKeyboardButton("▶️ Resume", callback_data='resume')],
[InlineKeyboardButton("🔄 Refresh", callback_data='refresh_status')]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(text, parse_mode='Markdown', reply_markup=reply_markup)
@auth_required
async def stop_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
keyboard = [[
InlineKeyboardButton("✅ Confirm STOP", callback_data='confirm_stop'),
InlineKeyboardButton("❌ Cancel", callback_data='cancel_stop')
]]
await update.message.reply_text(
"⚠️ *Emergency Stop*\nThis will cancel all orders and close all positions. Confirm?",
parse_mode='Markdown',
reply_markup=InlineKeyboardMarkup(keyboard)
)
async def send_trade_notification(bot, trade_data):
"""Отправка уведомления о сделке"""
emoji = "🟢" if trade_data['side'] == 'buy' else "🔴"
text = f"""{emoji} *Trade Executed*
Symbol: `{trade_data['symbol']}`
Side: {trade_data['side'].upper()}
Price: `${trade_data['price']:,.2f}`
Size: `{trade_data['quantity']}`
{'P&L: ' + f"`{trade_data['pnl']:+.2f}%`" if 'pnl' in trade_data else ''}"""
for user_id in ALLOWED_USERS:
await bot.send_message(user_id, text, parse_mode='Markdown')
# Запуск
def main():
application = Application.builder().token(TOKEN).build()
application.add_handler(CommandHandler("status", status_command))
application.add_handler(CommandHandler("stop", stop_command))
application.add_handler(CallbackQueryHandler(button_callback))
application.run_polling()
Безопасность
- ALLOWED_USERS whitelist — только авторизованные Telegram ID
- Подтверждение для деструктивных операций (stop, close all)
- Логирование всех команд
- Rate limiting на команды (не чаще N раз в минуту)
Срок разработки: 1–2 недели для полнофункционального Telegram-интерфейса.