Разработка 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 автоматизация
| Параметр | Ручной способ | Автоматизация |
|---|---|---|
| Время на публикацию | 5-10 минут на сигнал | 1-2 секунды (в 300 раз быстрее) |
| Обновление статусов | Вручную, часто забывается | Автоматически в реальном времени (задержка <100 мс) |
| Статистика | Собирается в Excel или не собирается | Еженедельный отчёт автоматически |
| Ошибки | Человеческий фактор (до 20% сигналов с ошибкой) | Исключён (0% ошибок) |
| Затраты времени аналитика | 2-3 часа в день | 0 — система работает автономно |
Сравнение методов мониторинга цен
| Метод | Задержка | Надёжность | Сложность |
|---|---|---|---|
| WebSocket | <50 мс | 99.9% | Средняя |
| REST polling (1 сек) | 1-2 сек | 99% | Низкая |
| REST polling (10 сек) | 10-20 сек | 98% | Низкая |
WebSocket даёт минимальную задержку и подходит для трейдинга. Мы используем его с fallback на REST при обрыве соединения.
Почему Telegram Bot API — не единственное ограничение?
Самый частый сценарий — сигналы публикуются в канал, но подписчики хотят получать уведомления в личку. Для этого нужно либо пересылать сообщения через бота (но это спам), либо сделать отдельный бот, который подписывает пользователя на рассылку. Второй вариант требует реализации inline-кнопок "Подписаться" и "Отписаться", а также хранения состояний подписок в базе данных. Мы это тоже делаем — минимальная нагрузка на сервер и 99.9% uptime гарантированы.
Обработка ошибок и логирование
Система включает автоматическое переподключение при обрыве WebSocket, запись всех событий в structured logs (JSON), и уведомления администратора при падении канала. Каждый сигнал логируется с timestamp и результатом.Что входит в работу
- Репозиторий с кодом на GitHub (Python, asyncio, python-telegram-bot)
- Документация по развёртыванию и настройке
- Доступ к серверу и мониторингу (логи, метрики)
- Обучение команды работе с системой
- Гарантия 30 дней на исправление ошибок
Оценим ваш проект за 1 рабочий день. Получите консультацию по вашему проекту — бесплатно. Закажите разработку и начните экономить время уже на этой неделе.