Bot · PY

"""
Solana Meme Coin Bot — Core Logic
Monitors DexScreener for new tokens, applies filters,
executes swaps via Jupiter, and reports via Telegram.
"""

import asyncio
import logging
from datetime import datetime, timedelta

import aiohttp
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.ext import (
    Application,
    CallbackQueryHandler,
    CommandHandler,
    ContextTypes,
)

import config

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────
# Bot class
# ─────────────────────────────────────────────────────────────

class SolanaBot:
    def __init__(self, telegram_token: str, chat_id: str) -> None:
        self.telegram_token = telegram_token
        self.chat_id = chat_id
        self.is_running: bool = False
        self.positions: dict = {}          # address → position dict
        self.daily_pnl: float = 0.0
        self.total_trades: int = 0
        self.session: aiohttp.ClientSession | None = None

    # ── Lifecycle ─────────────────────────────────────────────

    async def start(self) -> None:
        self.session = aiohttp.ClientSession()

        self.app = Application.builder().token(self.telegram_token).build()
        self._register_handlers()

        await self.app.initialize()
        await self.app.start()
        await self.app.updater.start_polling()

        await self.send_message(
            "🚀 *Solana Bot Started*\n\n"
            f"✅ Scanning Raydium & Pump.fun\n"
            f"⚙️ Max per trade : `{config.MAX_SOL_PER_TRADE}` SOL\n"
            f"🛡️ Daily loss cap : `{config.DAILY_LOSS_LIMIT_SOL}` SOL\n\n"
            "Use /help for all commands."
        )

        await self.monitor_loop()

    def _register_handlers(self) -> None:
        handlers = [
            CommandHandler("start",   self.cmd_start),
            CommandHandler("status",  self.cmd_status),
            CommandHandler("balance", self.cmd_balance),
            CommandHandler("pnl",     self.cmd_pnl),
            CommandHandler("buy",     self.cmd_buy),
            CommandHandler("sell",    self.cmd_sell),
            CommandHandler("stop",    self.cmd_stop),
            CommandHandler("help",    self.cmd_help),
            CallbackQueryHandler(self.handle_callback),
        ]
        for h in handlers:
            self.app.add_handler(h)

    # ── Monitoring loop ───────────────────────────────────────

    async def monitor_loop(self) -> None:
        self.is_running = True
        logger.info("Monitor loop started.")

        while self.is_running:
            try:
                tokens = await self.scan_new_tokens()
                for token in tokens:
                    if self.should_buy(token):
                        await self.execute_buy(token)

                await self.check_open_positions()
            except Exception as exc:
                logger.error(f"Monitor loop error: {exc}")

            await asyncio.sleep(config.SCAN_INTERVAL_SECONDS)

    # ── DexScreener ───────────────────────────────────────────

    async def scan_new_tokens(self) -> list[dict]:
        try:
            url = f"{config.DEXSCREENER_API}/tokens/solana"
            async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp:
                if resp.status != 200:
                    return []
                data = await resp.json()

            cutoff = datetime.utcnow() - timedelta(minutes=config.MAX_TOKEN_AGE_MINUTES)
            result = []

            for pair in data.get("pairs", []):
                created_ts = (pair.get("pairCreatedAt") or 0) / 1000
                if created_ts < cutoff.timestamp():
                    continue

                result.append({
                    "address":        pair.get("baseToken", {}).get("address", ""),
                    "symbol":         pair.get("baseToken", {}).get("symbol", "???"),
                    "name":           pair.get("baseToken", {}).get("name", ""),
                    "price_usd":      float(pair.get("priceUsd") or 0),
                    "liquidity_usd":  (pair.get("liquidity") or {}).get("usd", 0),
                    "volume_h1":      (pair.get("volume") or {}).get("h1", 0),
                    "price_change_5m": (pair.get("priceChange") or {}).get("m5", 0),
                    "pair_address":   pair.get("pairAddress", ""),
                    "created_at":     created_ts,
                })

            logger.info(f"Scanned {len(result)} new token(s).")
            return result

        except Exception as exc:
            logger.error(f"DexScreener error: {exc}")
            return []

    async def get_token_price(self, token_address: str) -> float | None:
        try:
            url = f"{config.DEXSCREENER_API}/tokens/{token_address}"
            async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                if resp.status != 200:
                    return None
                data = await resp.json()
                pairs = data.get("pairs") or []
                if pairs:
                    return float(pairs[0].get("priceUsd") or 0)
        except Exception as exc:
            logger.error(f"Price fetch error ({token_address[:8]}…): {exc}")
        return None

    # ── Jupiter ───────────────────────────────────────────────

    async def get_jupiter_quote(
        self, input_mint: str, output_mint: str, amount_lamports: int
    ) -> dict | None:
        try:
            url = f"{config.JUPITER_API}/quote"
            params = {
                "inputMint":   input_mint,
                "outputMint":  output_mint,
                "amount":      amount_lamports,
                "slippageBps": 500,
            }
            async with self.session.get(
                url, params=params, timeout=aiohttp.ClientTimeout(total=5)
            ) as resp:
                if resp.status == 200:
                    return await resp.json()
        except Exception as exc:
            logger.error(f"Jupiter quote error: {exc}")
        return None

    # ── Buy / Sell ────────────────────────────────────────────

    def should_buy(self, token: dict) -> bool:
        if not token.get("address"):
            return False
        if token["address"] in self.positions:
            return False
        if self.daily_pnl <= -config.DAILY_LOSS_LIMIT_SOL:
            logger.warning("Daily loss limit reached — skipping buy.")
            return False
        if token["liquidity_usd"] < config.MIN_LIQUIDITY_USD:
            return False
        if token["price_change_5m"] < config.MIN_PRICE_CHANGE_5M:
            return False

        logger.info(
            f"✅ Buy signal: {token['symbol']} | "
            f"Liq ${token['liquidity_usd']:,.0f} | "
            f"+{token['price_change_5m']}% 5m"
        )
        return True

    async def execute_buy(self, token: dict) -> None:
        symbol = token["symbol"]
        amount_sol = config.MAX_SOL_PER_TRADE

        quote = await self.get_jupiter_quote(
            input_mint="So11111111111111111111111111111111111111112",
            output_mint=token["address"],
            amount_lamports=int(amount_sol * 1e9),
        )
        if not quote:
            logger.warning(f"No Jupiter quote for {symbol}.")
            return

        self.positions[token["address"]] = {
            "symbol":        symbol,
            "buy_price_usd": token["price_usd"],
            "amount_sol":    amount_sol,
            "buy_time":      datetime.utcnow(),
            "pair_address":  token["pair_address"],
        }
        self.total_trades += 1

        await self.send_message(
            f"🟢 *BUY*\n\n"
            f"🪙 `{symbol}`\n"
            f"💰 {amount_sol} SOL\n"
            f"💵 ${token['price_usd']:.8f}\n"
            f"🏊 Liq ${token['liquidity_usd']:,.0f}\n"
            f"📈 5m +{token['price_change_5m']}%\n"
            f"🔗 [DexScreener](https://dexscreener.com/solana/{token['pair_address']})"
        )
        logger.info(f"Bought {symbol} for {amount_sol} SOL.")

    async def check_open_positions(self) -> None:
        for addr, pos in list(self.positions.items()):
            price = await self.get_token_price(addr)
            if price is None:
                continue

            change_pct = ((price - pos["buy_price_usd"]) / pos["buy_price_usd"]) * 100

            if change_pct >= config.TAKE_PROFIT_PERCENT:
                await self.execute_sell(addr, reason=f"🎯 Take profit +{change_pct:.1f}%")
            elif change_pct <= -config.STOP_LOSS_PERCENT:
                await self.execute_sell(addr, reason=f"🛑 Stop loss {change_pct:.1f}%")

    async def execute_sell(self, token_address: str, reason: str = "Manual") -> None:
        pos = self.positions.pop(token_address, None)
        if not pos:
            return

        price = await self.get_token_price(token_address)
        pnl_pct = ((price - pos["buy_price_usd"]) / pos["buy_price_usd"]) * 100 if price else 0
        pnl_sol = pos["amount_sol"] * (pnl_pct / 100)
        self.daily_pnl += pnl_sol

        emoji = "🟢" if pnl_pct > 0 else "🔴"
        await self.send_message(
            f"{emoji} *SELL*\n\n"
            f"🪙 `{pos['symbol']}`\n"
            f"📊 {reason}\n"
            f"💹 PnL: {pnl_pct:+.1f}% ({pnl_sol:+.4f} SOL)\n"
            f"📅 Daily PnL: {self.daily_pnl:+.4f} SOL"
        )
        logger.info(f"Sold {pos['symbol']}: {pnl_pct:+.1f}% ({pnl_sol:+.4f} SOL).")

    # ── Telegram commands ─────────────────────────────────────

    async def cmd_start(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        kb = [
            [InlineKeyboardButton("📊 Status",  callback_data="status"),
             InlineKeyboardButton("💰 Balance", callback_data="balance")],
            [InlineKeyboardButton("📈 PnL",     callback_data="pnl"),
             InlineKeyboardButton("❓ Help",    callback_data="help")],
        ]
        await update.message.reply_text(
            "🚀 *Solana Meme Bot Active*\n\n"
            "Monitoring Raydium & Pump.fun for new launches.\n"
            "Tap a button or type /help for commands.",
            reply_markup=InlineKeyboardMarkup(kb),
            parse_mode="Markdown",
        )

    async def cmd_status(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        state = "🟢 Running" if self.is_running else "🔴 Stopped"
        text  = (
            f"*Bot Status*\n\n"
            f"State       : {state}\n"
            f"Open pos.   : {len(self.positions)}\n"
            f"Total trades: {self.total_trades}\n"
            f"Daily PnL   : {self.daily_pnl:+.4f} SOL\n"
        )
        if self.positions:
            text += "\n*Open Positions:*\n"
            for addr, pos in self.positions.items():
                age = (datetime.utcnow() - pos["buy_time"]).seconds // 60
                text += f"• `{pos['symbol']}` — {age}m ago\n"
        await update.message.reply_text(text, parse_mode="Markdown")

    async def cmd_balance(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        await update.message.reply_text(
            "💰 Wallet balance check coming soon.\n"
            "_(Connect your Solana RPC to enable this.)_",
            parse_mode="Markdown",
        )

    async def cmd_pnl(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        await update.message.reply_text(
            f"📈 *PnL Report*\n\n"
            f"Today        : {self.daily_pnl:+.4f} SOL\n"
            f"Total trades : {self.total_trades}",
            parse_mode="Markdown",
        )

    async def cmd_buy(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        args = ctx.args
        if not args:
            await update.message.reply_text("Usage: `/buy <token_address>`", parse_mode="Markdown")
            return
        await update.message.reply_text(
            f"🔄 Fetching quote for `{args[0][:12]}…`", parse_mode="Markdown"
        )

    async def cmd_sell(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        args = ctx.args
        if not args:
            await update.message.reply_text("Usage: `/sell <token_address>`", parse_mode="Markdown")
            return
        await self.execute_sell(args[0], reason="Manual sell")

    async def cmd_stop(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        self.is_running = False
        await update.message.reply_text("🛑 Auto-trading paused. Use /start to resume.")

    async def cmd_help(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        await update.message.reply_text(
            "*Commands*\n\n"
            "/status — Status & open positions\n"
            "/balance — SOL wallet balance\n"
            "/pnl — Profit/loss report\n"
            "/buy `<addr>` — Manual buy\n"
            "/sell `<addr>` — Manual sell\n"
            "/stop — Pause auto-trading\n"
            "/help — Show this menu",
            parse_mode="Markdown",
        )

    async def handle_callback(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
        q = update.callback_query
        await q.answer()
        update.message = q.message
        dispatch = {
            "status":  self.cmd_status,
            "balance": self.cmd_balance,
            "pnl":     self.cmd_pnl,
            "help":    self.cmd_help,
        }
        if q.data in dispatch:
            await dispatch[q.data](update, ctx)

    # ── Helper ────────────────────────────────────────────────

    async def send_message(self, text: str) -> None:
        try:
            await self.app.bot.send_message(
                chat_id=self.chat_id,
                text=text,
                parse_mode="Markdown",
                disable_web_page_preview=True,
            )
        except Exception as exc:
            logger.error(f"Telegram send error: {exc}")