Development of Telegram Channel with Automated Trading Signals
A Telegram channel with automated trading signals is an integration of an algorithmic signal generation system with Telegram Bot API. Automation includes signal publishing, status updates (entry reached, TP triggered, SL triggered), end-of-period statistics.
System Architecture
# Main components
SignalGenerator → SignalFormatter → TelegramPublisher → MessageTracker
↓
StatusUpdater (on market change)
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,
)
# Remember message_id for subsequent updates
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):
"""Edit original message with updated status"""
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>Entry reached</b>"
elif status == "TP1":
status_line = "\n\n🎯 <b>TP1 triggered!</b>"
elif status == "TP2":
status_line = "\n\n🎯🎯 <b>TP2 triggered!</b>"
elif status == "SL":
status_line = "\n\n🛑 <b>Stop Loss triggered</b>"
elif status == "CLOSED":
status_line = f"\n\n📊 <b>Closed: {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}
💰 Entry: <code>${signal.entry_price:,.2f}</code>
{tps}
🛑 Stop: <code>${signal.stop_loss:,.2f}</code>
📊 Timeframe: {signal.timeframe}
⚡️ Risk: {signal.risk_pct or 1}% of deposit
📝 {signal.rationale}{status_line}"""
Price Monitoring and Status Update
class SignalStatusMonitor:
async def monitor_signal(self, signal: TradingSignal):
"""Track signal until close"""
async for price in self.price_stream.subscribe(signal.symbol):
# Check entry
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
# Check TP levels
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}%"
)
# Check 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
Automatic Weekly Reports
async def send_weekly_report(bot: Bot, channel_id: str, stats: WeeklyStats):
report = f"""
📊 <b>Weekly Summary</b>
Total signals: {stats.total}
✅ Profitable: {stats.profitable} ({stats.win_rate:.0%})
❌ Losing: {stats.losing}
💰 Average result: {stats.avg_result:+.1f}%
📈 Best signal: {stats.best_symbol} ({stats.best_pnl:+.1f}%)
📉 Worst signal: {stats.worst_symbol} ({stats.worst_pnl:+.1f}%)
🏆 Win streak: {stats.current_win_streak}
"""
await bot.send_message(channel_id, report, parse_mode=ParseMode.HTML)
Rate Limits and Throttling
Telegram Bot API has limitations: 30 messages per second in personal chats, 1 message per second in channel. When broadcasting to individual chats (not channel) — add queue with throttling via asyncio.Semaphore or Redis-based rate limiter.