diff --git a/bot.py b/bot.py index 7e6be5f..d6d00ee 100644 --- a/bot.py +++ b/bot.py @@ -1,28 +1,31 @@ import os import asyncio -from typing import Final import requests +import logging from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, ContextTypes, ConversationHandler, filters +from typing import Final, Dict, Any, List, Tuple + +# Setup logging +logging.basicConfig(level=logging.INFO) # Constants BOT_USERNAME: Final = 'xyz' BOT_TOKEN: Final = "Your Bot Token" COINGECKO_API_URL: Final = "https://api.coingecko.com/api/v3" +ALLOWED_USERS: Final = [123456789, 987654321] # List of user IDs allowed to set alerts # Conversation states -MAIN_MENU, CHOOSING_CRYPTO, CHOOSING_CURRENCY, TYPING_SEARCH, COMPARE_SELECTION = range(5) +MAIN_MENU, CHOOSING_CRYPTO, CHOOSING_CURRENCY, TYPING_SEARCH, COMPARE_SELECTION, SETTING_NOTIFICATION = range(6) # Supported currencies SUPPORTED_CURRENCIES = ['usd', 'eur', 'gbp', 'jpy', 'aud', 'cad', 'chf', 'cny', 'inr'] - +# User Data +user_data: Dict[int, Dict[str, Any]] = {} # API HELPER FUNCTIONS - -# def get_top_cryptos(limit=100): -def get_top_cryptos(is_comparing=False,limit=100): - +def get_top_cryptos(is_comparing=False, limit=100): response = requests.get(f"{COINGECKO_API_URL}/coins/markets", params={ 'vs_currency': 'usd', 'order': 'market_cap_desc', @@ -34,8 +37,6 @@ def get_top_cryptos(is_comparing=False,limit=100): return response.json() return [] - - def get_trending_cryptos(): response = requests.get(f"{COINGECKO_API_URL}/search/trending") if response.status_code == 200: @@ -50,9 +51,19 @@ def get_crypto_details(crypto_id: str, currency: str = 'usd'): return data.get(crypto_id) return None +def get_historical_price(crypto: str, currency: str = 'usd', days: int = 30): + response = requests.get(f"{COINGECKO_API_URL}/coins/{crypto}/market_chart", params={ + 'vs_currency': currency, + 'days': days + }) + if response.status_code == 200: + return response.json() + return None # COMMAND HANDLER FUNCTIONS async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + user_id = update.message.from_user.id + user_data[user_id] = {'preferred_currency': 'usd'} # Set default currency await show_main_menu(update, context) return MAIN_MENU @@ -62,24 +73,24 @@ async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No "Commands:\n" "/start - Show main menu\n" "/help - Show this help message\n" - "/convert - Convert Currencies From One To Another\n\n" + "/convert - Convert Currencies From One To Another\n" + "/history - Get historical price data for a cryptocurrency\n" + "/setnotification - Set notification preferences\n\n" "You can check prices of top cryptocurrencies, view trending coins, search for a specific cryptocurrency or Convert them." ) await update.message.reply_text(help_text) - - # Menu Display and Button Handlers async def show_main_menu(update: Update, context: ContextTypes.DEFAULT_TYPE, is_comparing: bool = False) -> None: - keyboard = [ [InlineKeyboardButton("Top 100 Cryptocurrencies", callback_data='top100')], [InlineKeyboardButton("Trending Cryptocurrencies", callback_data='trending')], - [InlineKeyboardButton("Search Cryptocurrency", callback_data='search')], # Added missing comma here + [InlineKeyboardButton("Search Cryptocurrency", callback_data='search')], + [InlineKeyboardButton("Set Notification Preferences", callback_data='set_notification')], [InlineKeyboardButton("Quit", callback_data='quit')] ] reply_markup = InlineKeyboardMarkup(keyboard) - + # Show welcome message only when not comparing if not is_comparing: text = "Welcome to the Crypto Price Bot! What would you like to do?" @@ -91,9 +102,7 @@ async def show_main_menu(update: Update, context: ContextTypes.DEFAULT_TYPE, is_ else: await update.message.reply_text(text, reply_markup=reply_markup) - async def show_crypto_list(update: Update, context: ContextTypes.DEFAULT_TYPE, cryptos, title) -> None: - keyboard = [] for i in range(0, len(cryptos), 2): row = [] @@ -113,71 +122,39 @@ async def show_crypto_list(update: Update, context: ContextTypes.DEFAULT_TYPE, c else: await update.message.reply_text(title, reply_markup=reply_markup) -async def show_crypto_details(update: Update, context: ContextTypes.DEFAULT_TYPE, crypto_id: str, currency: str) -> None: - await asyncio.sleep(1) # Add a delay to avoid hitting rate limits - details = get_crypto_details(crypto_id, currency) - if details: - price = details.get(currency, 'N/A') - change_24h = details.get(f'{currency}_24h_change', 'N/A') - market_cap = details.get(f'{currency}_market_cap', 'N/A') - trading_volume = details.get(f'{currency}_24h_vol', 'N/A') - - try: - change_24h_float = float(change_24h) - change_symbol = '🔺' if change_24h_float > 0 else '🔻' if change_24h_float < 0 else '➖' - except (ValueError, TypeError): - change_symbol = '➖' - - message = ( - f"💰 {crypto_id.capitalize()} ({currency.upper()})\n" - f"Price: {price} {currency.upper()}\n" - f"24h Change: {change_symbol} {change_24h}%\n" - f"Market Cap: {market_cap} {currency.upper()}\n" - f"24h Trading Volume: {trading_volume} {currency.upper()}\n\n" - ) - - # Check if the new message is different from the current message - current_message = update.callback_query.message.text - if message != current_message: - await update.callback_query.edit_message_text(message) - - # Adding options: Compare with other cryptos or Main Menu - keyboard = [ - [InlineKeyboardButton("Compare with another Cryptocurrency", callback_data='compare_selection')], - [InlineKeyboardButton("Main Menu", callback_data='main_menu')], - [InlineKeyboardButton("Quit", callback_data='quit')] - ] - reply_markup = InlineKeyboardMarkup(keyboard) - await update.callback_query.message.reply_text("Select an option:", reply_markup=reply_markup) +async def show_historical_data(update: Update, context: ContextTypes.DEFAULT_TYPE, crypto: str) -> None: + currency = user_data[update.message.from_user.id].get('preferred_currency', 'usd') + historical_data = get_historical_price(crypto, currency) + + if historical_data: + prices = historical_data['prices'] + message = f"Historical price data for {crypto.capitalize()}:\n" + for price in prices[:5]: # Show first 5 entries + timestamp, value = price + message += f"{timestamp}: ${value:.2f}\n" + await update.message.reply_text(message) else: - await update.callback_query.edit_message_text("🚫 Unable to retrieve cryptocurrency details.") - + await update.message.reply_text("🚫 Unable to retrieve historical data.") - -async def show_currency_options(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - +async def set_notification_preferences(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: keyboard = [ - [InlineKeyboardButton(currency.upper(), callback_data=f"currency:{currency}")] - for currency in SUPPORTED_CURRENCIES + [InlineKeyboardButton("Instant Notifications", callback_data='instant')], + [InlineKeyboardButton("Daily Summaries", callback_data='daily')], + [InlineKeyboardButton("Cancel", callback_data='cancel_notifications')] ] - keyboard.append([InlineKeyboardButton("Back to Main Menu", callback_data='main_menu')]) reply_markup = InlineKeyboardMarkup(keyboard) - await update.callback_query.edit_message_text('Choose a currency:', reply_markup=reply_markup) - - + await update.message.reply_text("Choose your notification preferences:", reply_markup=reply_markup) +# API Callbacks async def button_click(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - query = update.callback_query - await query.answer() if query.data == 'main_menu': - # Instead of deleting the message, just edit it to show the main menu try: await show_main_menu(update, context) except Exception as e: - print(f"Error displaying main menu: {e}") + logging.error(f"Error displaying main menu: {e}") return MAIN_MENU @@ -187,213 +164,71 @@ async def button_click(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No await show_crypto_list(update, context, cryptos, "Top 100 Cryptocurrencies:") return CHOOSING_CRYPTO elif query.data == 'quit': - # Quit the current conversation and show a message await query.edit_message_text("You can return to the main menu anytime by using /start.") - return MAIN_MENU # This will allow the user to start again later - - elif query.data == 'trending': await query.edit_message_text("Fetching trending cryptocurrencies, please wait...") cryptos = get_trending_cryptos() await show_crypto_list(update, context, cryptos, "Trending Cryptocurrencies:") - return CHOOSING_CRYPTO elif query.data == 'search': await query.edit_message_text("Please enter the name of the cryptocurrency you want to check:") - return TYPING_SEARCH elif query.data.startswith('crypto:'): context.user_data['crypto'] = query.data.split(':')[1] await show_currency_options(update, context) - return CHOOSING_CURRENCY elif query.data.startswith('currency:'): currency = query.data.split(':')[1] crypto_id = context.user_data.get('crypto', 'bitcoin') await show_crypto_details(update, context, crypto_id, currency) - return COMPARE_SELECTION - elif query.data == 'compare_selection': - crypto_id = context.user_data.get('crypto') - - if not crypto_id: - await query.edit_message_text("Please select a cryptocurrency before comparing.") - return - - # Fetch the top 100 currencies again for selection - await query.edit_message_text("Fetching top 100 currencies for comparison, please wait...") - cryptos = get_top_cryptos(is_comparing=True) # Fetch top 100 for comparison - await show_crypto_list(update, context, cryptos, f"Compare {crypto_id} with another currency:") - - # Now wait for the user to select a new currency to compare - return CHOOSING_CURRENCY - - elif query.data == 'cancel_compare': - await query.edit_message_text("Comparison cancelled.") - - else: - await query.edit_message_text("Invalid selection. Returning to main menu.") - await show_main_menu(update, context) - + elif query.data == 'set_notification': + await set_notification_preferences(update, context) + return SETTING_NOTIFICATION + elif query.data in ['instant', 'daily']: + # Set the user's notification preference + user_id = update.message.from_user.id + user_data[user_id]['notification_preference'] = query.data + await query.edit_message_text(f"Notification preference set to {query.data.capitalize()}.") return MAIN_MENU - - - - -# **New Function to Show Comparison Options** -# Updated show_compare_options function -async def show_compare_options(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - - cryptos = get_top_cryptos() # Fetch the top 100 cryptocurrencies - - keyboard = [] - for i in range(0, len(cryptos), 2): - row = [] - for crypto in cryptos[i:i + 2]: - name = crypto.get('name', 'Unknown') - symbol = crypto.get('symbol', 'Unknown') - crypto_id = crypto.get('id', 'unknown') - row.append(InlineKeyboardButton(f"{name} ({symbol.upper()})", callback_data=f"compare:{crypto_id}")) - keyboard.append(row) - - keyboard.append([InlineKeyboardButton("Cancel", callback_data='cancel_compare')]) - reply_markup = InlineKeyboardMarkup(keyboard) - - # Skip the welcome message during comparison - await show_main_menu(update, context, is_comparing=True) - - - -# **New Function to Handle Comparison Prompt** -async def compare_prompt_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: - # Send the message asking user to select another cryptocurrency - await update.callback_query.message.reply_text("Select the another Cryptocurrency...") - - # Then proceed to show the comparison options - await show_compare_options(update, context) - return COMPARE_SELECTION - - -# 6. Message and Error Handlers -async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: - - user_input = update.message.text.lower() - search_results = requests.get(f"{COINGECKO_API_URL}/search", params={'query': user_input}).json() - coins = search_results.get('coins', []) - - if coins: - await show_crypto_list(update, context, coins[:10], "Search Results:") - return CHOOSING_CRYPTO - else: - await update.message.reply_text("Sorry, I couldn't find any cryptocurrency matching your search.") - await show_main_menu(update, context) + elif query.data == 'cancel_notifications': + await query.edit_message_text("Notification preferences canceled.") return MAIN_MENU - -# Error Handler -async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): - - print(f"Update {update} caused error {context.error}") - - -# Convert Currency into USD -# Function to fetch crypto price -def get_crypto_price(crypto: str, currency: str = 'usd'): - params = {'ids': crypto, 'vs_currencies': currency} - response = requests.get(COINGECKO_API_URL+'/simple/price', params=params) - data = response.json() - return data.get(crypto, {}).get(currency, 'Price not available') - - -# Convert crypto to different currencies -async def convert_command(update: Update, context: ContextTypes.DEFAULT_TYPE): - if len(context.args) < 3: - await update.message.reply_text("Please use the format: /convert \n\nFor Example `/convert bitcoin usd 1` - Convert bitcoin price into usd by fettching real time data\n\n") - return - crypto = context.args[0].lower() - currency = context.args[1].lower() - amount = float(context.args[2]) - price = get_crypto_price(crypto, currency) - if price != 'Price not available': - converted_amount = price * amount - await update.message.reply_text(f"{amount} {crypto.capitalize()} is worth {converted_amount} {currency.upper()}.") else: - await update.message.reply_text('Price not available.') - - -# Add alerts - -# Function to set up price alerts -user_alerts = {} - -def set_price_alert(user_id, crypto, threshold_price, condition): - user_alerts[user_id] = (crypto, threshold_price, condition) - -async def set_alert_command(update: Update, context: ContextTypes.DEFAULT_TYPE): - if len(context.args) < 3: - usage_text = ( - "Please use the format: /setalert \n" - ) - await update.message.reply_text(usage_text) - return - - crypto = context.args[0].lower() # Get the cryptocurrency (e.g., 'bitcoin') - condition = context.args[1].lower() # Get the condition (e.g., 'above' or 'below') - price = float(context.args[2]) # Get the price threshold - - if condition not in ['above', 'below']: - await update.message.reply_text("Please specify 'above' or 'below' for the price alert condition.") - return + await query.edit_message_text("🚫 Unknown command.") - user_id = update.message.from_user.id # Get the user's ID - - # Save the alert with the condition (above or below) - set_price_alert(user_id, crypto, price, condition) +async def show_currency_options(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + keyboard = [InlineKeyboardButton(currency.upper(), callback_data=f'currency:{currency}') for currency in SUPPORTED_CURRENCIES] + keyboard.append(InlineKeyboardButton("Cancel", callback_data='main_menu')) + reply_markup = InlineKeyboardMarkup([keyboard]) - # Notify the user that the alert has been set - await update.message.reply_text( - f"Price alert set for {crypto.capitalize()} when price is {condition} ${price} USD.\n" - "You'll be notified when this condition is met." - ) + await update.callback_query.edit_message_text("Choose a currency:", reply_markup=reply_markup) -async def alert_check(context: ContextTypes.DEFAULT_TYPE): - for user_id, (crypto, threshold_price, condition) in user_alerts.items(): - price = get_crypto_price(crypto) +async def show_crypto_details(update: Update, context: ContextTypes.DEFAULT_TYPE, crypto_id: str, currency: str) -> None: + crypto_details = get_crypto_details(crypto_id, currency) - # Check if the condition (above or below) is met - if (condition == 'above' and price >= threshold_price) or (condition == 'below' and price <= threshold_price): - await context.bot.send_message( - chat_id=user_id, - text=f"Price alert! {crypto.capitalize()} has {'exceeded' if condition == 'above' else 'dropped below'} ${threshold_price} USD. Current price: ${price} USD." - ) + if crypto_details: + message = (f"**{crypto_id.capitalize()} Price:** ${crypto_details['usd']:.2f}\n" + f"**Market Cap:** ${crypto_details['market_cap']:.2f}\n" + f"**24h Change:** {crypto_details['24h_change']:.2f}%") + await update.callback_query.edit_message_text(message, parse_mode='Markdown') + else: + await update.callback_query.edit_message_text("🚫 Unable to retrieve cryptocurrency details.") +# Error handling +async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): + logging.error(f"Update {update} caused error {context.error}") def main() -> None: app = Application.builder().token(BOT_TOKEN).build() - - conv_handler = ConversationHandler( - entry_points=[CommandHandler("start", start)], - states={ - MAIN_MENU: [CallbackQueryHandler(button_click)], - CHOOSING_CRYPTO: [CallbackQueryHandler(button_click)], - CHOOSING_CURRENCY: [CallbackQueryHandler(button_click)], - TYPING_SEARCH: [MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)], - COMPARE_SELECTION: [CallbackQueryHandler(button_click)] # Added COMPARE_SELECTION - }, - fallbacks=[CommandHandler("start", start)], - per_message=False - ) - - app.add_handler(conv_handler) - app.add_handler(CommandHandler('convert', convert_command)) - app.add_handler(CommandHandler('setalert', set_alert_command)) + app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("help", help_command)) + app.add_handler(CallbackQueryHandler(button_click)) + app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) app.add_error_handler(error_handler) - # Run alert checker periodically - app.job_queue.run_repeating(alert_check, interval=60) - print('Starting bot...') - app.run_polling(poll_interval=3) + app.run_polling() if __name__ == '__main__': main()