Spaces:
Runtime error
Runtime error
| 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() |