Автоматизація Telegram-каналу торгових сигналів

Розробка Telegram-каналу торгових сигналів з автоматизацією

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

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

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 робочий день. Отримайте консультацію по вашому проекту — безкоштовно. Замовте розробку та почніть економити час вже на цьому тижні.