Technical Details

Monad Blockchain
       ↓
  Event Monitoring
       ↓
  Transaction Analysis
       ↓
  Notification Formatting
       ↓
  Telegram API
       ↓
  Your Community Group

# Get token decimals
decimals = token_contract.functions.decimals().call()
 
# Get token symbol
symbol = token_contract.functions.symbol().call()
 
# Get token name
name = token_contract.functions.name().call()
 
# Monitor Transfer events
transfer_filter = token_contract.events.Transfer.create_filter(
    fromBlock='latest',
    argument_filters={'from': pool_address}
)

event Transfer(
    address indexed from,
    address indexed to,
    uint256 value
)

# Get pool address for token pair
pool_address = factory.functions.getPair(
    token_address,
    wmon_address
).call()
 
# Get all pairs count
pairs_count = factory.functions.allPairsLength().call()
 
# Get specific pair by index
pair = factory.functions.allPairs(index).call()

FACTORIES = {
    'Uniswap V2': '0x...',
    'PancakeSwap V2': '0x...',
    'OctoSwap V2': '0x...',
    'TayaSwap V2': '0x...',
    'Purps Exchange V2': '0x...',
    'Madness Finance V2': '0x...'
}

# Get tokens in pair
token0 = pool.functions.token0().call()
token1 = pool.functions.token1().call()
 
# Get reserves
reserves = pool.functions.getReserves().call()
# Returns: (reserve0, reserve1, blockTimestampLast)
 
# Monitor Sync events
sync_filter = pool.events.Sync.create_filter(
    fromBlock='latest'
)

def monitor_blocks():
    while True:
        current_block = w3.eth.block_number
 
        if current_block > last_processed_block:
            process_blocks(last_processed_block + 1, current_block)
            last_processed_block = current_block
 
        time.sleep(2)  # Poll every 2 seconds

def detect_buys(pool_address, token_address):
    # Create event filter
    transfer_filter = token_contract.events.Transfer.create_filter(
        fromBlock=start_block,
        toBlock=end_block,
        argument_filters={
            'from': pool_address  # From pool = buy
        }
    )
 
    # Get all transfer events
    events = transfer_filter.get_all_entries()
 
    for event in events:
        # event['args']['from'] = pool_address
        # event['args']['to'] = buyer_address
        # event['args']['value'] = token_amount
 
        if is_buy_event(event):
            process_buy(event)

def analyze_buy(tx_hash):
    # Get transaction receipt
    receipt = w3.eth.get_transaction_receipt(tx_hash)
 
    # Extract buyer address
    buyer = receipt['from']
 
    # Get token amount from Transfer event
    token_amount = parse_transfer_amount(receipt)
 
    # Get MON spent from WMON Transfer
    mon_spent = parse_mon_spent(receipt)
 
    # Identify DEX used
    dex_name = identify_dex(pool_address)
 
    return {
        'buyer': buyer,
        'token_amount': token_amount,
        'mon_spent': mon_spent,
        'tx_hash': tx_hash,
        'dex': dex_name,
        'timestamp': receipt['blockNumber']
    }

def identify_dex(pool_address):
    # Check cache first
    if pool_address in dex_cache:
        return dex_cache[pool_address]
 
    # Query each factory
    for dex_name, factory_address in FACTORIES.items():
        factory = w3.eth.contract(
            address=factory_address,
            abi=FACTORY_ABI
        )
 
        # Get pair for tokens
        pair = factory.functions.getPair(
            token_address,
            wmon_address
        ).call()
 
        # Check if matches
        if pair.lower() == pool_address.lower():
            dex_cache[pool_address] = dex_name
            return dex_name
 
    return 'Unknown'

dex_cache = {
    '0x1234...': 'Uniswap V2',
    '0x5678...': 'PancakeSwap V2',
    '0x9abc...': 'OctoSwap V2'
}

def format_notification(buy_data):
    # Calculate green circles
    circles = min(int(buy_data['mon_spent'] / 0.1), 10)
    indicator = '🟢' * circles
 
    # Format amounts
    token_amount = format_number(buy_data['token_amount'])
    mon_spent = f"{buy_data['mon_spent']:.4f}"
 
    # Shorten address
    buyer = shorten_address(buy_data['buyer'])
 
    # Build message
    message = f"""
{indicator} New Buy!
 
💰 Amount: {token_amount} {symbol}
💸 Spent: {mon_spent} MON
👤 Buyer: [{buyer}]({explorer_url}/address/{buy_data['buyer']})
🔍 Tx: [View on Explorer]({explorer_url}/tx/{buy_data['tx_hash']})
🔄 DEX: [{buy_data['dex']}]({dexscreener_url})
 
{format_social_links(config['social_links'])}
 
Powered by 🤖 Mobot's Monad Buy Bot
[Purchase on MoBots]({mobots_url})
"""
 
    return message

def send_with_image(chat_id, message, image_url):
    try:
        # Try sending with photo
        bot.send_photo(
            chat_id=chat_id,
            photo=image_url,
            caption=message,
            parse_mode='MarkdownV2'
        )
    except Exception as e:
        # Fallback to text-only
        bot.send_message(
            chat_id=chat_id,
            text=message,
            parse_mode='MarkdownV2',
            disable_web_page_preview=False
        )

from telegram import Bot
from telegram.ext import Updater, CommandHandler
 
# Initialize bot
bot = Bot(token=TELEGRAM_BOT_TOKEN)
 
# Send notification
def send_notification(chat_id, message):
    bot.send_message(
        chat_id=chat_id,
        text=message,
        parse_mode='MarkdownV2',
        disable_web_page_preview=False
    )

{
  _id: ObjectId,
  chat_id: Number,                    // Telegram group ID
  chat_title: String,                 // Group name
  creator_id: Number,                 // Creator's Telegram ID
  token_address: String,              // Monitored token
  pool_addresses: [String],           // Pool addresses
  dex_names: [String],                // DEX names per pool
  notification_image: String,         // Image URL
  minimum_amount: Number,             // Min buy threshold
  social_links: {
    discord: String,
    telegram: String,
    twitter: String,
    website: String
  },
  active: Boolean,                    // Active/inactive
  created_at: Date,
  updated_at: Date
}

{
  chat_id: 1;
} // Quick lookup by group
{
  creator_id: 1;
} // List user's bots
{
  active: 1;
} // Filter active bots

from concurrent.futures import ThreadPoolExecutor
 
# Create thread pool
executor = ThreadPoolExecutor(max_workers=32)
 
# Process monitors concurrently
for monitor in active_monitors:
    executor.submit(run_monitor, monitor)

# Sliding window cache
processed_txs = {}  # {tx_hash: block_number}
CACHE_SIZE = 100  # blocks
 
def is_processed(tx_hash, current_block):
    if tx_hash in processed_txs:
        return True
 
    # Add to cache
    processed_txs[tx_hash] = current_block
 
    # Cleanup old entries
    if len(processed_txs) > 1000:
        cleanup_cache(current_block - CACHE_SIZE)
 
    return False
 
def cleanup_cache(min_block):
    for tx_hash, block in list(processed_txs.items()):
        if block < min_block:
            del processed_txs[tx_hash]

def verify_creator(user_id, chat_id):
    config = db.group_configs.find_one({'chat_id': chat_id})
 
    if not config:
        return False
 
    return config['creator_id'] == user_id
 
# Usage
if not verify_creator(message.from_user.id, chat_id):
    return "❌ Only the creator can manage this bot"

def safe_process_buy(buy_data):
    try:
        # Format notification
        message = format_notification(buy_data)
 
        # Send to Telegram
        send_notification(chat_id, message, image_url)
 
    except TelegramError as e:
        logger.error(f"Telegram error: {e}")
        # Retry with text-only
        send_notification(chat_id, message, image_url=None)
 
    except Exception as e:
        logger.error(f"Processing error: {e}")
        # Log but don't crash monitor

def retry_operation(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

web3==6.0.0
python-telegram-bot==20.0
pymongo==4.0
requests==2.31.0