Building custom notification bots for Twitter monitoring transforms how individuals and teams consume social media intelligence. While generic monitoring tools provide basic functionality, custom bots offer unlimited flexibility for filtering, formatting, and routing notifications based on your specific needs.
This comprehensive guide walks you through creating Twitter notification bots for both Telegram and Discord, from basic setup to advanced features like sentiment analysis, priority routing, and webhook integration. Whether you're monitoring crypto KOLs, tracking brand mentions, or building team alerting systems, these implementations provide the foundation for professional-grade social media monitoring.
Why Build Custom Notification Bots?
Custom bots offer significant advantages over existing monitoring solutions:
- Complete control over filtering logic. Implement complex rules that generic tools can't provide.
- Custom formatting and presentation. Design notifications that match your workflow and information needs.
- Multiple notification channels. Route different types of alerts to appropriate channels or users.
- Integration capabilities. Connect with your existing tools, databases, and workflows.
- Cost effectiveness. No per-user or per-notification pricing limits.
Telegram Bot Implementation
Telegram provides an excellent platform for notification bots with rich formatting, inline buttons, and reliable message delivery.
Setting Up Your Telegram Bot
- Create Bot with BotFather. Message @BotFather on Telegram, use /newbot command, and save the bot token.
- Get Chat ID. Add your bot to a chat and send /start. Use Telegram API to get chat ID for message targeting.
- Set bot permissions. Configure bot settings through BotFather including privacy mode and commands.
Basic Telegram Bot Code
Here's a complete Python implementation for a Twitter notification Telegram bot:
import asyncio
import aiohttp
from telegram import Bot, ParseMode
from telegram.error import TelegramError
import logging
from datetime import datetime
class TwitterTelegramBot:
def __init__(self, bot_token, default_chat_id):
self.bot = Bot(token=bot_token)
self.default_chat_id = default_chat_id
self.logger = logging.getLogger(__name__)
async def send_tweet_notification(self, tweet_data, chat_id=None):
"""Send formatted tweet notification"""
try:
chat_id = chat_id or self.default_chat_id
message = self._format_tweet_message(tweet_data)
await self.bot.send_message(
chat_id=chat_id,
text=message,
parse_mode=ParseMode.MARKDOWN,
disable_web_page_preview=False
)
except TelegramError as e:
self.logger.error(f"Failed to send telegram message: {e}")
def _format_tweet_message(self, tweet_data):
"""Format tweet data for Telegram display"""
account = tweet_data['account']
tweet = tweet_data['tweet']
# Emoji based on account verification
verified_emoji = "✅" if account.get('verified') else ""
# Priority indicator
priority_emoji = {
'high': '🚨',
'medium': '⚡',
'low': '💬'
}.get(tweet_data.get('priority', 'low'), '💬')
# Format message with Markdown
message = f"""
{priority_emoji} *{account['display_name']}* {verified_emoji}
@{account['username']} • {self._format_time(tweet['created_at'])}
{tweet['text']}
🔗 [View Tweet]({tweet['url']})
📊 Followers: {account['follower_count']:,}
"""
# Add sentiment if available
if 'sentiment' in tweet_data:
sentiment = tweet_data['sentiment']
sentiment_emoji = "😊" if sentiment['polarity'] > 0.3 else "😐" if sentiment['polarity'] > -0.3 else "😞"
message += f"\n{sentiment_emoji} Sentiment: {sentiment['label'].title()}"
return message.strip()
def _format_time(self, timestamp):
"""Format timestamp for display"""
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
return dt.strftime("%H:%M")
# Usage example
async def main():
bot = TwitterTelegramBot(
bot_token="YOUR_BOT_TOKEN",
default_chat_id="YOUR_CHAT_ID"
)
# Example tweet data
sample_tweet = {
'account': {
'username': 'elonmusk',
'display_name': 'Elon Musk',
'verified': True,
'follower_count': 175000000
},
'tweet': {
'text': 'Dogecoin to the moon! 🚀',
'url': 'https://twitter.com/elonmusk/status/1234567890',
'created_at': '2026-03-04T18:30:12.000Z'
},
'priority': 'high',
'sentiment': {
'polarity': 0.8,
'label': 'positive'
}
}
await bot.send_tweet_notification(sample_tweet)
if __name__ == "__main__":
asyncio.run(main())
Discord Bot Implementation
Discord bots excel at team notifications with rich embeds, role mentions, and channel organization.
Discord Bot Setup
- Create Discord Application. Visit Discord Developer Portal, create new application, and navigate to Bot section.
- Generate Bot Token. Click "Add Bot" and copy the token for authentication.
- Invite Bot to Server. Use OAuth2 URL generator with appropriate permissions.
Discord Bot Code
import discord
from discord.ext import commands
import asyncio
import logging
from datetime import datetime
class TwitterDiscordBot(commands.Bot):
def __init__(self, command_prefix='!'):
intents = discord.Intents.default()
intents.message_content = True
super().__init__(command_prefix=command_prefix, intents=intents)
self.logger = logging.getLogger(__name__)
async def on_ready(self):
print(f'{self.user} is connected to Discord!')
async def send_tweet_notification(self, tweet_data, channel_id):
"""Send formatted tweet notification as Discord embed"""
try:
channel = self.get_channel(channel_id)
if not channel:
self.logger.error(f"Channel {channel_id} not found")
return
embed = self._create_tweet_embed(tweet_data)
await channel.send(embed=embed)
except discord.DiscordException as e:
self.logger.error(f"Failed to send Discord message: {e}")
def _create_tweet_embed(self, tweet_data):
"""Create rich embed for tweet display"""
account = tweet_data['account']
tweet = tweet_data['tweet']
# Color based on priority
color_map = {
'high': 0xFF4444, # Red
'medium': 0xFFAA00, # Orange
'low': 0x00AA00 # Green
}
color = color_map.get(tweet_data.get('priority', 'low'), 0x1DA1F2)
embed = discord.Embed(
title=f"{account['display_name']} (@{account['username']})",
description=tweet['text'],
color=color,
url=tweet['url'],
timestamp=datetime.fromisoformat(tweet['created_at'].replace('Z', '+00:00'))
)
# Add author with verification status
verified_text = " ✅" if account.get('verified') else ""
embed.set_author(
name=f"{account['display_name']}{verified_text}",
icon_url=f"https://unavatar.io/twitter/{account['username']}"
)
# Add fields
embed.add_field(
name="👥 Followers",
value=f"{account['follower_count']:,}",
inline=True
)
if 'sentiment' in tweet_data:
sentiment = tweet_data['sentiment']
embed.add_field(
name="😊 Sentiment",
value=sentiment['label'].title(),
inline=True
)
# Add footer with metadata
embed.set_footer(text=f"Priority: {tweet_data.get('priority', 'low').title()}")
return embed
# Bot setup and event handlers
bot = TwitterDiscordBot()
@bot.command(name='test')
async def test_notification(ctx):
"""Test command for bot functionality"""
sample_tweet = {
'account': {
'username': 'VitalikButerin',
'display_name': 'Vitalik Buterin',
'verified': True,
'follower_count': 5200000
},
'tweet': {
'text': 'Ethereum 2.0 staking rewards are now live! 🎉',
'url': 'https://twitter.com/VitalikButerin/status/1234567890',
'created_at': '2026-03-04T18:30:12.000Z'
},
'priority': 'high',
'sentiment': {
'polarity': 0.9,
'label': 'positive'
}
}
await bot.send_tweet_notification(sample_tweet, ctx.channel.id)
# Run bot
# bot.run('YOUR_BOT_TOKEN')
Integration with Twitter Monitoring Services
To make your bots useful, integrate them with Twitter monitoring APIs or services like Xanguard.
Xanguard Integration Example
import asyncio
import aiohttp
from telegram import Bot
class XanguardTelegramIntegration:
def __init__(self, telegram_bot_token, telegram_chat_id, xanguard_api_key):
self.telegram_bot = Bot(token=telegram_bot_token)
self.chat_id = telegram_chat_id
self.xanguard_api_key = xanguard_api_key
self.webhook_url = None
async def setup_webhook(self, webhook_url):
"""Configure Xanguard webhook to point to your bot"""
self.webhook_url = webhook_url
async with aiohttp.ClientSession() as session:
headers = {'Authorization': f'Bearer {self.xanguard_api_key}'}
webhook_config = {
'url': webhook_url,
'filters': {
'accounts': ['elonmusk', 'VitalikButerin', 'naval'],
'keywords': ['bitcoin', 'ethereum', 'crypto'],
'priority': 'high'
}
}
async with session.post(
'https://api.xanguard.tech/v1/webhooks',
json=webhook_config,
headers=headers
) as response:
if response.status == 200:
print("Webhook configured successfully!")
else:
print(f"Webhook setup failed: {response.status}")
async def process_webhook(self, payload):
"""Process incoming webhook from Xanguard"""
try:
# Format message for Telegram
message = self._format_xanguard_payload(payload)
await self.telegram_bot.send_message(
chat_id=self.chat_id,
text=message,
parse_mode='Markdown',
disable_web_page_preview=False
)
except Exception as e:
print(f"Error processing webhook: {e}")
def _format_xanguard_payload(self, payload):
"""Format Xanguard webhook data for Telegram"""
account = payload['account']
tweet = payload['tweet']
priority_emoji = {
'high': '🚨',
'medium': '⚡',
'low': '💬'
}.get(payload.get('priority', 'low'), '💬')
return f"""
{priority_emoji} *{account['display_name']}*
@{account['username']} • Just now
{tweet['text']}
🔗 [View Tweet]({tweet['url']})
📈 Confidence: {payload.get('confidence_score', 0):.0%}
"""
Advanced Features
Smart Filtering and Routing
Implement intelligent routing based on content analysis:
class SmartNotificationRouter:
def __init__(self):
self.channels = {
'high_priority': 'URGENT_CHANNEL_ID',
'crypto_news': 'CRYPTO_CHANNEL_ID',
'general': 'GENERAL_CHANNEL_ID'
}
async def route_notification(self, tweet_data):
"""Intelligently route notifications based on content"""
# High priority accounts always go to urgent channel
if tweet_data['account']['username'] in ['elonmusk', 'VitalikButerin']:
await self.send_to_channel('high_priority', tweet_data)
return
# Crypto-related content
crypto_keywords = ['bitcoin', 'ethereum', 'crypto', 'defi', 'nft']
if any(keyword in tweet_data['tweet']['text'].lower() for keyword in crypto_keywords):
await self.send_to_channel('crypto_news', tweet_data)
return
# Default to general channel
await self.send_to_channel('general', tweet_data)
async def send_to_channel(self, channel_type, tweet_data):
"""Send notification to specific channel"""
channel_id = self.channels[channel_type]
# Implementation depends on platform (Telegram/Discord)
pass
Rate Limiting and Anti-Spam
import asyncio
from collections import defaultdict
from datetime import datetime, timedelta
class NotificationRateLimiter:
def __init__(self, max_messages=10, time_window=60):
self.max_messages = max_messages
self.time_window = time_window
self.message_history = defaultdict(list)
async def should_send_notification(self, chat_id):
"""Check if notification should be sent based on rate limits"""
now = datetime.now()
cutoff = now - timedelta(seconds=self.time_window)
# Clean old messages
self.message_history[chat_id] = [
timestamp for timestamp in self.message_history[chat_id]
if timestamp > cutoff
]
# Check current rate
if len(self.message_history[chat_id]) >= self.max_messages:
return False
# Record current message
self.message_history[chat_id].append(now)
return True
async def send_rate_limited_notification(self, bot, chat_id, message):
"""Send notification with rate limiting"""
if await self.should_send_notification(chat_id):
await bot.send_message(chat_id=chat_id, text=message)
else:
print(f"Rate limit exceeded for chat {chat_id}, message queued")
Production Deployment Tips
Use environment variables for all tokens and secrets. Implement proper logging and error handling. Set up monitoring for bot uptime and message delivery rates. Use webhooks instead of polling for better performance and lower resource usage.
Monitoring and Analytics
Track bot performance and notification effectiveness:
import sqlite3
from datetime import datetime
import asyncio
class BotAnalytics:
def __init__(self, db_path='bot_analytics.db'):
self.db_path = db_path
self.init_database()
def init_database(self):
"""Initialize analytics database"""
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS notifications (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
chat_id TEXT,
account_username TEXT,
tweet_id TEXT,
priority TEXT,
delivery_status TEXT,
response_time_ms INTEGER
)
''')
async def log_notification(self, chat_id, tweet_data, delivery_status, response_time):
"""Log notification for analytics"""
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
INSERT INTO notifications
(timestamp, chat_id, account_username, tweet_id, priority, delivery_status, response_time_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
datetime.now().isoformat(),
str(chat_id),
tweet_data['account']['username'],
tweet_data['tweet']['id'],
tweet_data.get('priority', 'normal'),
delivery_status,
int(response_time * 1000)
))
def get_performance_stats(self, hours=24):
"""Get bot performance statistics"""
cutoff = (datetime.now() - timedelta(hours=hours)).isoformat()
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute('''
SELECT
COUNT(*) as total_notifications,
AVG(response_time_ms) as avg_response_time,
SUM(CASE WHEN delivery_status = 'success' THEN 1 ELSE 0 END) as successful_deliveries
FROM notifications
WHERE timestamp > ?
''', (cutoff,))
return cursor.fetchone()
Best Practices and Security
- Secure token storage. Never hardcode API tokens. Use environment variables or secure key management.
- Input validation. Sanitize all incoming webhook data to prevent injection attacks.
- Rate limiting. Implement proper rate limiting to avoid overwhelming users or hitting API limits.
- Error handling. Gracefully handle network errors, API failures, and malformed data.
- Monitoring. Track bot uptime, message delivery rates, and error frequencies.
- Privacy compliance. Follow relevant data protection regulations for user data handling.
Conclusion
Custom Twitter notification bots provide unmatched flexibility for social media monitoring. By implementing the patterns and code examples in this guide, you can build sophisticated notification systems that scale with your needs and integrate seamlessly with existing workflows.
For teams that prefer turnkey solutions, Xanguard provides ready-made Twitter monitoring bots for both Telegram and Discord with enterprise-grade reliability and no coding required. The @Xanguard_bot can be configured in minutes and includes advanced features like priority filtering, sentiment analysis, and webhook integration.
Whether building custom solutions or using existing platforms, the key is choosing an approach that matches your technical capabilities and monitoring requirements while providing reliable, actionable intelligence from Twitter.