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

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

Telegram-канал с автоматическими торговыми сигналами — это интеграция алгоритмической системы генерации сигналов с Telegram Bot API. Автоматизация включает публикацию сигналов, обновление статусов (вход достигнут, TP сработал, SL сработал), статистику по итогам периода.

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

# Основные компоненты
SignalGenerator → SignalFormatter → TelegramPublisher → MessageTracker
                                                    ↓
                                           StatusUpdater (при изменении рынка)

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.