import os import logging import requests import yfinance as yf import json from telegram import Update from telegram.ext import Application, CommandHandler, ContextTypes logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO ) logger = logging.getLogger(__name__) TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") if not TOKEN: raise ValueError("No TELEGRAM_BOT_TOKEN found") try: from upstash_redis import Redis redis = Redis( url=os.getenv("UPSTASH_REDIS_REST_URL"), token=os.getenv("UPSTASH_REDIS_REST_TOKEN") ) logger.info("Redis connected successfully") except Exception as e: logger.error(f"Redis connection failed: {e}") redis = None def get_price(ticker): ticker = ticker.upper().strip() crypto_ids = { "BTC": "bitcoin", "ETH": "ethereum", "SOL": "solana", "XRP": "ripple", "DOGE": "dogecoin", "ADA": "cardano", "AVAX": "avalanche-2", "LINK": "chainlink", "BNB": "binancecoin", "DOT": "polkadot", "MATIC": "matic-network", "SHIB": "shiba-inu" } if ticker in crypto_ids: try: coin = crypto_ids[ticker] url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin}&vs_currencies=usd" r = requests.get(url, timeout=10) return r.json()[coin]["usd"] except: return None try: hist = yf.Ticker(ticker).history(period="1d") if hist.empty: return None return float(hist["Close"].iloc[-1]) except: return None def get_data(ticker): ticker = ticker.upper().strip() crypto_ids = { "BTC": "bitcoin", "ETH": "ethereum", "SOL": "solana", "XRP": "ripple", "DOGE": "dogecoin", "ADA": "cardano", "AVAX": "avalanche-2", "LINK": "chainlink", "BNB": "binancecoin", "DOT": "polkadot", "MATIC": "matic-network", "SHIB": "shiba-inu" } if ticker in crypto_ids: try: coin = crypto_ids[ticker] url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin}&vs_currencies=usd&include_24hr_change=true" r = requests.get(url, timeout=10) data = r.json() if coin not in data: return f"❌ No data found for {ticker}" price = data[coin]["usd"] change = data[coin].get("usd_24h_change", 0) if change > 2: trend = "Strong bullish 📈" elif change > 0.5: trend = "Mildly bullish 📈" elif change < -2: trend = "Strong bearish 📉" elif change < -0.5: trend = "Mildly bearish 📉" else: trend = "Sideways ↔️" return ( f"🦍 *Big Brain Ape Analysis*\n\n" f"*{ticker}*\n" f"💰 Price: `${price:,.2f}`\n" f"📊 24h Change: `{change:+.2f}%`\n" f"📈 Trend: {trend}" ) except Exception as e: logger.error(e) return f"❌ Error fetching crypto data for {ticker}" try: hist = yf.Ticker(ticker).history(period="5d") if hist.empty: return f"❌ No data found for {ticker}" price = float(hist["Close"].iloc[-1]) prev = float(hist["Close"].iloc[-2]) if len(hist) > 1 else price change = price - prev pct = (change / prev) * 100 if prev != 0 else 0 if pct > 2: trend = "Strong bullish 📈" elif pct > 0.5: trend = "Mildly bullish 📈" elif pct < -2: trend = "Strong bearish 📉" elif pct < -0.5: trend = "Mildly bearish 📉" else: trend = "Sideways ↔️" return ( f"🦍 *Big Brain Ape Analysis*\n\n" f"*{ticker}*\n" f"💰 Price: `${price:,.2f}`\n" f"📊 Change: `{change:+.2f}` (`{pct:+.2f}%`)\n" f"📈 Trend: {trend}" ) except Exception as e: logger.error(e) return f"❌ Error fetching data for {ticker}" async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): user = update.effective_user text = ( f"🦍 Welcome to *Big Brain Ape*, {user.first_name}!\n\n" "Your AI-powered market analysis assistant.\n\n" "• /free TICKER — Free analysis\n" "• /alert TICKER PRICE — Set price alert\n" "• /alerts — View your alerts\n" "• /delalert NUMBER — Remove an alert\n" "• /paid — Deeper analysis\n" "• /hire — Hire the bot\n" "• /help — All commands" ) await update.message.reply_text(text, parse_mode="Markdown") async def free_analysis(update: Update, context: ContextTypes.DEFAULT_TYPE): if not context.args: await update.message.reply_text( "Please provide a ticker.\n\nExamples:\n`/free BTC`\n`/free NVDA`\n`/free ETH`", parse_mode="Markdown" ) return await update.message.reply_text("🔍 Fetching live data...") result = get_data(context.args[0]) await update.message.reply_text(result, parse_mode="Markdown") async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE): text = ( "🦍 *Big Brain Ape Commands*\n\n" "/free TICKER — Free analysis\n" "/alert TICKER PRICE — Set alert\n" "/alerts — List your alerts\n" "/delalert NUMBER — Delete an alert\n" "/paid — Paid deep analysis\n" "/hire — Hire the bot\n" "/help — Show this message" ) await update.message.reply_text(text, parse_mode="Markdown") async def paid_analysis(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text( "💎 Paid analysis coming soon.\nFor now use `/free TICKER`.", parse_mode="Markdown" ) async def hire(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text( "🤝 *Hire Big Brain Ape*\n\n" "Performance-only model — free to start.\n" "You only pay when the bot makes you profit.\n\n" "Coming soon.", parse_mode="Markdown" ) async def set_alert(update: Update, context: ContextTypes.DEFAULT_TYPE): if redis is None: await update.message.reply_text("❌ Alerts are temporarily unavailable.") return if not context.args or len(context.args) < 2: await update.message.reply_text( "Usage:\n`/alert BTC 95000`\n`/alert SOL 8%`", parse_mode="Markdown" ) return ticker = context.args[0].upper() target = context.args[1] user_id = update.effective_user.id alert = {"ticker": ticker, "target": target, "user_id": user_id} key = f"alerts:{user_id}" alerts = redis.get(key) alerts = json.loads(alerts) if alerts else [] alerts.append(alert) redis.set(key, json.dumps(alerts)) await update.message.reply_text( f"✅ Alert set: *{ticker}* → `{target}`", parse_mode="Markdown" ) async def list_alerts(update: Update, context: ContextTypes.DEFAULT_TYPE): if redis is None: await update.message.reply_text("❌ Alerts are temporarily unavailable.") return user_id = update.effective_user.id key = f"alerts:{user_id}" alerts = redis.get(key) if not alerts: await update.message.reply_text("You have no active alerts.") return alerts = json.loads(alerts) text = "📋 *Your Alerts:*\n\n" for i, a in enumerate(alerts, 1): text += f"{i}. *{a['ticker']}* → `{a['target']}`\n" await update.message.reply_text(text, parse_mode="Markdown") async def delete_alert(update: Update, context: ContextTypes.DEFAULT_TYPE): if redis is None: await update.message.reply_text("❌ Alerts are temporarily unavailable.") return if not context.args: await update.message.reply_text("Usage: `/delalert 1`", parse_mode="Markdown") return try: index = int(context.args[0]) - 1 except: await update.message.reply_text("Please give a number, example: `/delalert 1`", parse_mode="Markdown") return user_id = update.effective_user.id key = f"alerts:{user_id}" alerts = redis.get(key) if not alerts: await update.message.reply_text("You have no alerts.") return alerts = json.loads(alerts) if index < 0 or index >= len(alerts): await update.message.reply_text("Invalid number.") return removed = alerts.pop(index) redis.set(key, json.dumps(alerts)) await update.message.reply_text( f"🗑 Removed: *{removed['ticker']}* → `{removed['target']}`", parse_mode="Markdown" ) async def check_alerts(context: ContextTypes.DEFAULT_TYPE): if redis is None: return try: keys = redis.keys("alerts:*") if not keys: return for key in keys: alerts_data = redis.get(key) if not alerts_data: continue alerts = json.loads(alerts_data) user_id = int(key.split(":")[1]) remaining = [] for alert in alerts: ticker = alert["ticker"] target = alert["target"] price = get_price(ticker) if price is None: remaining.append(alert) continue triggered = False if "%" not in str(target): try: if price >= float(target): triggered = True except: remaining.append(alert) continue else: remaining.append(alert) continue if triggered: try: await context.bot.send_message( chat_id=user_id, text=f"🚨 *ALERT*\n\n*{ticker}* hit `${price:,.2f}`\nTarget: `{target}`", parse_mode="Markdown" ) except Exception as e: logger.error(e) else: remaining.append(alert) if remaining: redis.set(key, json.dumps(remaining)) else: redis.delete(key) except Exception as e: logger.error(f"check_alerts error: {e}") def main(): app = Application.builder().token(TOKEN).build() app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("free", free_analysis)) app.add_handler(CommandHandler("help", help_command)) app.add_handler(CommandHandler("paid", paid_analysis)) app.add_handler(CommandHandler("hire", hire)) app.add_handler(CommandHandler("alert", set_alert)) app.add_handler(CommandHandler("alerts", list_alerts)) app.add_handler(CommandHandler("delalert", delete_alert)) if app.job_queue: app.job_queue.run_repeating(check_alerts, interval=60, first=10) logger.info("Bot starting...") app.run_polling() if __name__ == "__main__": main()