telegram-bot-python

import ccxt import pandas as pd import numpy as np import logging from telegram import Update from telegram.ext import Application, CommandHandler, CallbackContext

Logging configuration

logging.basicConfig( format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO ) logger = logging.getLogger(name)

Exchange API setup

exchange = ccxt.binance({ 'enableRateLimit': True, # Important to avoid rate limits 'options': { 'defaultType': 'spot', # Focus on spot trading } })

List of cryptocurrency pairs

SYMBOLS = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'ADA/USDT', 'DOGE/USDT'] TIMEFRAMES = ['1h', '1d'] # Timeframes to check RSI_PERIOD = 14 # Standard RSI period RSI_OVERBOUGHT = 70 # Overbought threshold RSI_OVERSOLD = 30 # Oversold threshold

def calculate_rsi(data, period=RSI_PERIOD): """Calculate RSI for given data.""" if len(data) < period: return None # Not enough data

delta = data['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()

rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi

def get_market_data(symbol, timeframe='1h', limit=100): """Fetch market data for a symbol.""" try: ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit) df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') return df except Exception as e: logger.error(f"Error fetching data for {symbol}: {str(e)}") return None

def check_rsi(symbol, timeframe): """Check RSI levels for a symbol.""" data = get_market_data(symbol, timeframe) if data is None or len(data) < RSI_PERIOD: return None

rsi = calculate_rsi(data)
if rsi is None or rsi.isna().all():
    return None

last_rsi = rsi.iloc[-1]
if pd.isna(last_rsi):
    return None

if last_rsi > RSI_OVERBOUGHT:
    return f'{symbol} ({timeframe}): Overbought (RSI: {last_rsi:.2f})'
elif last_rsi < RSI_OVERSOLD:
    return f'{symbol} ({timeframe}): Oversold (RSI: {last_rsi:.2f})'
return None

async def start(update: Update, context: CallbackContext): """Send a welcome message when the command /start is issued.""" await update.message.reply_text( 'Welcome to the RSI Analysis Bot!\n\n' 'Available commands:\n' '/start - Show this message\n' '/check_rsi - Check RSI levels for major crypto pairs' )

async def check_rsi_command(update: Update, context: CallbackContext): """Check RSI levels for all configured symbols.""" await update.message.reply_text("Checking RSI levels, please wait...")

results = []
for symbol in SYMBOLS:
    for timeframe in TIMEFRAMES:
        try:
            result = check_rsi(symbol, timeframe)
            if result:
                results.append(result)
        except Exception as e:
            logger.error(f"Error processing {symbol} {timeframe}: {e}")
            continue

if results:
    message = "Current RSI alerts:\n\n" + "\n".join(results)
else:
    message = "No crypto pairs are currently in overbought or oversold conditions."

await update.message.reply_text(message)

def main(): """Start the bot.""" # Replace with your actual Telegram bot token token = "7693970339:AAH1VFF1J_9U9M8B6UTcqFHhQKpnXArQ8fg"

try:
    app = Application.builder().token(token).build()
    
    # Add command handlers
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("check_rsi", check_rsi_command))
    
    logger.info("Bot is running...")
    app.run_polling()
except Exception as e:
    logger.error(f"Bot failed: {e}")

if name == "main": main()