Розробка 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 та регулювання

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