""" Telegram Bot — User interface for portfolio monitoring and copy-trade ===================================================================== Provides: - /status: Current portfolio allocation and value - /yields: Live yield rates across assets - /risk: Risk assessment summary - /report: Latest strategy report - /history: Recent allocation history - /subscribe: Enable weekly report delivery """ import asyncio import json import logging import os from datetime import datetime, timezone from typing import Optional logger = logging.getLogger("telegram_bot") # Check if python-telegram-bot is available try: from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import ( Application, CommandHandler, CallbackQueryHandler, ContextTypes, ) TELEGRAM_AVAILABLE = True except ImportError: TELEGRAM_AVAILABLE = False logger.warning("python-telegram-bot not installed. Bot will run in mock mode.") class YieldRouterBot: """ Telegram bot interface for the Dynamic RWA Yield Router. Provides read-only access to portfolio state, yields, and reports. Copy-trade functionality allows users to mirror the agent's allocations. """ def __init__( self, token: Optional[str] = None, agent=None, # YieldRouterAgent instance ): self.token = token or os.getenv("TELEGRAM_BOT_TOKEN") self.agent = agent self.subscribers = set() self._app = None async def start(self): """Start the Telegram bot.""" if not TELEGRAM_AVAILABLE: logger.error("Cannot start bot: python-telegram-bot not installed") return if not self.token: logger.error("Cannot start bot: TELEGRAM_BOT_TOKEN not set") return self._app = Application.builder().token(self.token).build() # Register handlers self._app.add_handler(CommandHandler("start", self._cmd_start)) self._app.add_handler(CommandHandler("status", self._cmd_status)) self._app.add_handler(CommandHandler("yields", self._cmd_yields)) self._app.add_handler(CommandHandler("risk", self._cmd_risk)) self._app.add_handler(CommandHandler("report", self._cmd_report)) self._app.add_handler(CommandHandler("history", self._cmd_history)) self._app.add_handler(CommandHandler("subscribe", self._cmd_subscribe)) self._app.add_handler(CommandHandler("unsubscribe", self._cmd_unsubscribe)) self._app.add_handler(CommandHandler("help", self._cmd_help)) self._app.add_handler(CallbackQueryHandler(self._handle_callback)) logger.info("Telegram bot starting...") await self._app.initialize() await self._app.start() await self._app.updater.start_polling() logger.info("Telegram bot running.") async def stop(self): """Stop the bot gracefully.""" if self._app: await self._app.updater.stop() await self._app.stop() await self._app.shutdown() async def broadcast_report(self, report_text: str): """Send report to all subscribers.""" if not self._app: return for chat_id in self.subscribers: try: await self._app.bot.send_message( chat_id=chat_id, text=report_text, parse_mode="Markdown", ) except Exception as e: logger.warning(f"Failed to send to {chat_id}: {e}") # ─────────────────── Command Handlers ─────────────── async def _cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /start command.""" welcome = """🤖 **Dynamic RWA Yield Router** ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Welcome to the autonomous AI portfolio manager for Mantle L2 RWA assets. **Commands:** /status — Current portfolio allocation /yields — Live yield rates /risk — Risk assessment /report — Latest strategy report /history — Recent allocation changes /subscribe — Weekly report alerts /help — Full command list Built for the Mantle Turing Test Hackathon 🏗️ """ await update.message.reply_text(welcome, parse_mode="Markdown") async def _cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /status command — show portfolio state.""" if not self.agent: await update.message.reply_text("⚠️ Agent not connected.") return state = self.agent.state.to_dict() weights = state["current_weights"] msg = f"""📊 **Portfolio Status** ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 💰 Value: ${state['portfolio_value']:,.2f} 📈 Total Return: {state['total_return_pct']:+.2f}% **Allocation:** 🏛️ USDY (T-bills): {weights['USDY']*100:.1f}% ⟠ mETH (Staked ETH): {weights['mETH']*100:.1f}% 📊 MI4 (Index Fund): {weights['MI4']*100:.1f}% 🔄 Rebalances: {state['total_rebalances']} ⛽ Gas Spent: ${state['total_gas_spent']:.4f} ⏱️ Uptime: {state['uptime_hours']:.1f}h """ keyboard = [ [ InlineKeyboardButton("📈 Yields", callback_data="yields"), InlineKeyboardButton("🛡️ Risk", callback_data="risk"), ], [InlineKeyboardButton("📊 Full Report", callback_data="report")], ] reply_markup = InlineKeyboardMarkup(keyboard) await update.message.reply_text(msg, parse_mode="Markdown", reply_markup=reply_markup) async def _cmd_yields(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /yields command — show live yield rates.""" if not self.agent or not self.agent.latest_snapshot: await update.message.reply_text("⚠️ No yield data available yet.") return s = self.agent.latest_snapshot msg = f"""📈 **Live Yield Rates** ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 🏛️ USDY: {s.usdy_apy:.2f}% APY ⟠ mETH: {s.meth_apy:.2f}% APY 📊 MI4: {s.mi4_apy:.2f}% APY **Lending Yields (Aave V3):** 🏦 USDY Supply: {s.aave_usdy_supply_apy:.2f}% APY 🏦 mETH Supply: {s.aave_meth_supply_apy:.2f}% APY **Prices:** Ξ ETH: ${s.eth_price:,.0f} ₿ BTC: ${s.btc_price:,.0f} Ⓜ MNT: ${s.mnt_price:.3f} **Macro:** 🏦 Fed Rate: {s.fed_funds_rate:.2f}% 📊 BTC Dom: {s.btc_dominance:.1f}% 📉 ETH Vol (30d): {s.eth_volatility_30d:.2f} ⛽ Gas: {s.gas_price_gwei:.4f} Gwei """ await update.message.reply_text(msg, parse_mode="Markdown") async def _cmd_risk(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /risk command — show risk assessment.""" if not self.agent: await update.message.reply_text("⚠️ Agent not connected.") return risk = self.agent.risk_manager.get_risk_summary() cb_emoji = {"closed": "🟢", "open": "🔴", "half_open": "🟡"} risk_emoji = {"low": "🟢", "medium": "🟡", "high": "🟠", "critical": "🔴"} msg = f"""🛡️ **Risk Assessment** ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {risk_emoji.get(risk['latest_risk_level'], '⚪')} Risk Level: {risk['latest_risk_level'].upper()} 📊 Risk Score: {risk['latest_risk_score']:.3f} {cb_emoji.get(risk['circuit_breaker'], '⚪')} Circuit Breaker: {risk['circuit_breaker'].upper()} 📉 Current Drawdown: {risk['current_drawdown']*100:.2f}% 📈 Peak Value: ${risk['peak_value']:,.2f} 📋 Total Assessments: {risk['total_assessments']} ⚠️ Total Warnings: {risk['total_warnings']} """ await update.message.reply_text(msg, parse_mode="Markdown") async def _cmd_report(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /report command — show latest strategy report.""" if not self.agent: await update.message.reply_text("⚠️ Agent not connected.") return reports = self.agent.strategy_reporter.report_history if not reports: await update.message.reply_text("📊 No reports generated yet.") return latest = reports[-1] msg = latest.to_telegram_message() # Telegram has 4096 char limit if len(msg) > 4000: msg = msg[:3990] + "\n..." await update.message.reply_text(msg, parse_mode="Markdown") async def _cmd_history(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /history command — recent allocation changes.""" if not self.agent: await update.message.reply_text("⚠️ Agent not connected.") return plans = self.agent.state.executed_plans[-5:] if not plans: await update.message.reply_text("📜 No rebalancing history yet.") return msg = "📜 **Recent Allocations**\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n" for i, plan in enumerate(reversed(plans)): ts = datetime.fromtimestamp(plan.timestamp, tz=timezone.utc).strftime("%m/%d %H:%M") weights = plan.target_weights msg += ( f"\n{i+1}. {ts} UTC\n" f" USDY={weights[0]*100:.0f}% mETH={weights[1]*100:.0f}% MI4={weights[2]*100:.0f}%\n" f" Trades: {len(plan.trades)} | Gas: ${plan.estimated_gas_usd:.4f}\n" ) await update.message.reply_text(msg, parse_mode="Markdown") async def _cmd_subscribe(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /subscribe command.""" chat_id = update.effective_chat.id self.subscribers.add(chat_id) await update.message.reply_text( "✅ Subscribed to weekly strategy reports!\n" "Use /unsubscribe to stop." ) async def _cmd_unsubscribe(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /unsubscribe command.""" chat_id = update.effective_chat.id self.subscribers.discard(chat_id) await update.message.reply_text("❌ Unsubscribed from reports.") async def _cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /help command.""" msg = """🤖 **Command Reference** ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ /status — Portfolio value, allocation, performance /yields — Live yield rates for USDY, mETH, MI4 /risk — Risk score, circuit breaker, drawdown /report — Latest weekly strategy report /history — Recent 5 allocation changes /subscribe — Get weekly reports via Telegram /unsubscribe — Stop weekly reports /help — This message **About:** Dynamic RWA Yield Router is an autonomous AI agent that optimizes capital allocation across Mantle's RWA asset stack using reinforcement learning. 📄 [GitHub](https://github.com/example/mantle-rwa-yield-router) 🌐 [Dashboard](https://mantle-rwa-router.example.com) """ await update.message.reply_text(msg, parse_mode="Markdown") async def _handle_callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle inline keyboard callbacks.""" query = update.callback_query await query.answer() if query.data == "yields": await self._cmd_yields(update, context) elif query.data == "risk": await self._cmd_risk(update, context) elif query.data == "report": await self._cmd_report(update, context) # ─────────────────── Mock Bot for Testing ─────────────────── class MockBot: """Mock bot for testing without Telegram API.""" def __init__(self, agent=None): self.agent = agent self.messages = [] def process_command(self, command: str) -> str: """Process a command and return the response text.""" if command == "/status" and self.agent: state = self.agent.state.to_dict() return json.dumps(state, indent=2) elif command == "/yields" and self.agent and self.agent.latest_snapshot: s = self.agent.latest_snapshot return f"USDY: {s.usdy_apy:.2f}% | mETH: {s.meth_apy:.2f}% | MI4: {s.mi4_apy:.2f}%" elif command == "/risk" and self.agent: return json.dumps(self.agent.risk_manager.get_risk_summary(), indent=2) return f"Unknown command: {command}"