Разработка Telegram-канала торговых сигналов с автоматизацией

Разработка Telegram-канала торговых сигналов с автоматизацией

Торговый аналитик вручную постит сигналы в Telegram — и забывает обновить статус по TP. Подписчики путаются, доверие падает. В день уходит до 3 часов на мониторинг и правку сообщений. Мы автоматизируем весь цикл: интеграция алгоритмической генерации сигналов с Telegram Bot API, публикация, обновление статусов (вход, TP, SL) и еженедельная статистика. Наш опыт — 5+ лет в разработке торговых ботов, более 10 успешных проектов.

Архитектура системы

SignalGenerator → SignalFormatter → TelegramPublisher → MessageTracker ↓ StatusUpdater (при изменении рынка) 

Как автоматизировать обновление статусов?

Бот слушает цены через WebSocket (Binance, Kraken) или Chainlink oracle. При каждом обновлении цены проверяются условия: вход, TP, SL. Если условие выполнено, редактируется исходное сообщение в канале с помощью edit_message_text. Задержка обновления — менее 100 мс. Это избавляет аналитика от ручного мониторинга — статусы обновляются автоматически в реальном времени.

Telegram Bot Setup

from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup from telegram.constants import ParseMode class TelegramSignalChannel: def __init__(self, bot_token: str, channel_id: str): self.bot = Bot(token=bot_token) self.channel_id = channel_id self.published_messages: dict[str, int] = {} # signal_id → message_id async def publish_signal(self, signal: TradingSignal) -> int: text = self.format_signal_message(signal) message = await self.bot.send_message( chat_id=self.channel_id, text=text, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ) # Запоминаем message_id для последующих обновлений self.published_messages[signal.id] = message.message_id return message.message_id async def update_signal_status(self, signal_id: str, status: str, details: str): """Редактирует оригинальное сообщение с обновлённым статусом""" message_id = self.published_messages.get(signal_id) if not message_id: return original_signal = await self.signal_repo.get(signal_id) updated_text = self.format_signal_message(original_signal, status=status, details=details) await self.bot.edit_message_text( chat_id=self.channel_id, message_id=message_id, text=updated_text, parse_mode=ParseMode.HTML, ) def format_signal_message(self, signal: TradingSignal, status: str = None, details: str = None) -> str: direction_emoji = "🟢" if signal.direction == "LONG" else "🔴" status_line = "" if status == "ENTRY_HIT": status_line = "\n\n✅ <b>Вход достигнут</b>" elif status == "TP1": status_line = "\n\n🎯 <b>TP1 сработал!</b>" elif status == "TP2": status_line = "\n\n🎯🎯 <b>TP2 сработал!</b>" elif status == "SL": status_line = "\n\n🛑 <b>Stop Loss сработал</b>" elif status == "CLOSED": status_line = f"\n\n📊 <b>Закрыт: {details}</b>" tps = "\n".join(f" 📍 TP{i+1}: <code>${tp:,.2f}</code>" for i, tp in enumerate(signal.take_profit_levels)) return f"""{direction_emoji} <b>{signal.symbol}</b> — {signal.direction} 💰 Вход: <code>${signal.entry_price:,.2f}</code> {tps} 🛑 Stop: <code>${signal.stop_loss:,.2f}</code> 📊 Таймфрейм: {signal.timeframe} ⚡️ Риск: {signal.risk_pct or 1}% от депозита 📝 {signal.rationale}{status_line}""" 

Мониторинг цены и обновление статусов

class SignalStatusMonitor: async def monitor_signal(self, signal: TradingSignal): """Отслеживает сигнал до закрытия""" async for price in self.price_stream.subscribe(signal.symbol): # Проверяем вход if not signal.entry_hit: if self.is_entry_triggered(signal, price): signal.entry_hit = True signal.entry_time = datetime.utcnow() await self.channel.update_signal_status(signal.id, "ENTRY_HIT", "") continue # Проверяем TP уровни for i, tp in enumerate(signal.take_profit_levels): if not signal.tp_hit[i]: if (signal.direction == "LONG" and price >= tp) or \ (signal.direction == "SHORT" and price <= tp): signal.tp_hit[i] = True pnl = ((tp - signal.entry_price) / signal.entry_price * 100) if signal.direction == "SHORT": pnl = -pnl await self.channel.update_signal_status( signal.id, f"TP{i+1}", f"+{pnl:.1f}%" ) # Проверяем SL if (signal.direction == "LONG" and price <= signal.stop_loss) or \ (signal.direction == "SHORT" and price >= signal.stop_loss): pnl = ((signal.stop_loss - signal.entry_price) / signal.entry_price * 100) if signal.direction == "LONG": pnl = -abs(pnl) await self.channel.update_signal_status(signal.id, "SL", f"{pnl:.1f}%") break 

Автоматические еженедельные отчёты

async def send_weekly_report(bot: Bot, channel_id: str, stats: WeeklyStats): report = f""" 📊 <b>Итоги недели</b> Всего сигналов: {stats.total} ✅ Прибыльных: {stats.profitable} ({stats.win_rate:.0%}) ❌ Убыточных: {stats.losing} 💰 Средний результат: {stats.avg_result:+.1f}% 📈 Лучший сигнал: {stats.best_symbol} ({stats.best_pnl:+.1f}%) 📉 Худший сигнал: {stats.worst_symbol} ({stats.worst_pnl:+.1f}%) 🏆 Серия побед: {stats.current_win_streak} """ await bot.send_message(channel_id, report, parse_mode=ParseMode.HTML) 

Rate Limits и throttling

Telegram Bot API имеет ограничения: 30 сообщений в секунду в личные чаты, 1 сообщение в секунду в канал. При массовой рассылке через отдельные чаты (не канал) — добавляем очередь с throttling через asyncio.Semaphore или Redis-based rate limiter. Нагрузка до 1000 подписчиков не вызывает задержек.

Хранение истории сигналов и аналитика

Все опубликованные сигналы и их статусы сохраняются в PostgreSQL. Таблица signals хранит symbol, direction, entry_price, tp_levels, sl и rationale с временной меткой публикации. Таблица signal_events фиксирует каждое изменение статуса: вход, TP1, TP2, SL — с ценой в момент события и рассчитанным P&L.

Такая схема позволяет строить отчёты за любой период: win rate по инструментам, средний P&L на таймфрейм, сравнение эффективности стратегий между собой. Индекс по (symbol, published_at) ускоряет выборку истории. При 30–50 сигналах в день объём за год не превышает 20 000 записей — для PostgreSQL это минимальная нагрузка.

Данные из базы подаются как в автоматические еженедельные отчёты, так и в дашборд аналитика. Дашборд реализуем на Grafana или на лёгком веб-интерфейсе — выбор зависит от ваших требований к визуализации и доступа команды.

Сравнение: ручная публикация vs автоматизация

Сравнение методов мониторинга цен

WebSocket даёт минимальную задержку и подходит для трейдинга. Мы используем его с fallback на REST при обрыве соединения.

Почему Telegram Bot API — не единственное ограничение?

Самый частый сценарий — сигналы публикуются в канал, но подписчики хотят получать уведомления в личку. Для этого нужно либо пересылать сообщения через бота (но это спам), либо сделать отдельный бот, который подписывает пользователя на рассылку. Второй вариант требует реализации inline-кнопок "Подписаться" и "Отписаться", а также хранения состояний подписок в базе данных. Мы это тоже делаем — минимальная нагрузка на сервер и 99.9% uptime гарантированы.

Что входит в работу

• Репозиторий с кодом на GitHub (Python, asyncio, python-telegram-bot)
• Документация по развёртыванию и настройке
• Доступ к серверу и мониторингу (логи, метрики)
• Обучение команды работе с системой
• Гарантия 30 дней на исправление ошибок

Оценим ваш проект за 1 рабочий день. Получите консультацию по вашему проекту — бесплатно. Закажите разработку и начните экономить время уже на этой неделе.