""" AlphaPulse Bot — Command & Callback Handlers """ import logging from telegram import Update from telegram.ext import ContextTypes from ai import AICommentary from fetcher import ContentFetcher logger = logging.getLogger(__name__) class CommandHandlers: """All /command and inline button callback handlers.""" def __init__(self, fetcher: ContentFetcher): self.fetcher = fetcher # ── Slash commands ──────────────────────────────────────────────────────── async def cmd_price(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/price — Live BTC + ETH + SOL prices.""" prices = await self.fetcher.fetch_prices() if not prices: await update.message.reply_text("⚠️ Could not fetch prices right now. Try again later.") return btc_a = '▲' if prices['btc_change'] > 0 else '▼' eth_a = '▲' if prices['eth_change'] > 0 else '▼' sol_a = '▲' if prices.get('sol_change', 0) > 0 else '▼' text = ( f"💹 <b>Live Prices</b>\n\n" f"₿ BTC <b>${prices['btc_price']:,.0f}</b> {btc_a} {prices['btc_change']:+.1f}%\n" f"⟠ ETH <b>${prices['eth_price']:,.0f}</b> {eth_a} {prices['eth_change']:+.1f}%\n" f"◎ SOL <b>${prices['sol_price']:,.2f}</b> {sol_a} {prices.get('sol_change', 0):+.1f}%\n\n" f"<i>Source: CoinGecko</i>" ) await update.message.reply_text(text, parse_mode='HTML') async def cmd_fear(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/fear — Current Fear & Greed Index.""" fg = await self.fetcher.fetch_fear_greed() if not fg: await update.message.reply_text("⚠️ Could not fetch Fear & Greed index. Try again later.") return text = AICommentary.fear_greed_text(fg['value'], fg['classification']) await update.message.reply_text( f"🎚️ <b>Fear & Greed Index</b>\n\n{text}\n\n<i>Source: alternative.me</i>", parse_mode='HTML' ) async def cmd_stats(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/stats — Top 5 gainers & losers (24h).""" gainers, losers = await self.fetcher.fetch_top_movers() if not gainers and not losers: await update.message.reply_text("⚠️ Could not fetch top movers. Try again later.") return lines = ["🏆 <b>Top Movers (24h)</b>", ""] if gainers: lines.append("🟢 <b>Gainers</b>") for c in gainers[:5]: pct = c['price_change_percentage_24h'] or 0 lines.append(f" {c['symbol'].upper()} +{pct:.1f}%") if losers: lines += ["", "🔴 <b>Losers</b>"] for c in losers[:5]: pct = c['price_change_percentage_24h'] or 0 lines.append(f" {c['symbol'].upper()} {pct:.1f}%") await update.message.reply_text('\n'.join(lines), parse_mode='HTML') async def cmd_trending(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/trending — Today's top trending coins.""" coins = await self.fetcher.fetch_trending() if not coins: await update.message.reply_text("⚠️ Could not fetch trending coins. Try again later.") return lines = ["🔥 <b>Trending Now</b>", ""] medals = ['🥇', '🥈', '🥉', '4️⃣', '5️⃣', '6️⃣', '7️⃣'] for i, c in enumerate(coins[:7]): rank = f"#{c['rank']}" if c['rank'] != '?' else '' lines.append(f"{medals[i]} <b>{c['symbol']}</b> {c['name']} {rank}") lines.append("\n<i>Source: CoinGecko</i>") await update.message.reply_text('\n'.join(lines), parse_mode='HTML') async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/help — Available commands.""" text = ( "🤖 <b>AlphaPulseXP Commands</b>\n\n" "/price — Live BTC + ETH prices\n" "/fear — Crypto Fear & Greed Index\n" "/stats — Top 5 gainers & losers (24h)\n" "/trending — Today's trending coins\n" "/help — This message\n\n" "📢 Channel: @alphapulsexp" ) await update.message.reply_text(text, parse_mode='HTML') # ── Inline button callbacks ─────────────────────────────────────────────── async def btn_price(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Inline button: 📊 Price.""" query = update.callback_query prices = await self.fetcher.fetch_prices() if not prices: await query.answer("⚠️ Could not fetch prices", show_alert=True) return btc_a = '▲' if prices['btc_change'] > 0 else '▼' eth_a = '▲' if prices['eth_change'] > 0 else '▼' sol_a = '▲' if prices.get('sol_change', 0) > 0 else '▼' text = ( f"₿ BTC ${prices['btc_price']:,.0f} {btc_a}{prices['btc_change']:+.1f}%\n" f"⟠ ETH ${prices['eth_price']:,.0f} {eth_a}{prices['eth_change']:+.1f}%\n" f"◎ SOL ${prices['sol_price']:,.2f} {sol_a}{prices.get('sol_change', 0):+.1f}%" ) await query.answer(text, show_alert=True) async def btn_fear(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Inline button: 😱 Fear.""" query = update.callback_query fg = await self.fetcher.fetch_fear_greed() if not fg: await query.answer("⚠️ Could not fetch Fear & Greed", show_alert=True) return text = AICommentary.fear_greed_text(fg['value'], fg['classification']) await query.answer(text.replace('\n', ' | '), show_alert=True) async def btn_trending(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Inline button: 🔥 Trending.""" query = update.callback_query coins = await self.fetcher.fetch_trending() if not coins: await query.answer("⚠️ Could not fetch trending coins", show_alert=True) return medals = ['🥇', '🥈', '🥉', '4️⃣', '5️⃣', '6️⃣', '7️⃣'] lines = [f"{medals[i]} {c['symbol']} {c['name']}" for i, c in enumerate(coins[:5])] await query.answer('\n'.join(lines), show_alert=True)