Build a free crypto price alert bot for Telegram in ~150 lines of Python

/price BTC ETH SOL -> live price + 24h change /alert BTC above 70000 -> one-time alert /alerts -> list your alerts 

pip install "aiogram>=3.13,<4.0" "httpx>=0.27,<1.0" export TELEGRAM_BOT_TOKEN=123456:your-token 

API = "https://data-api.binance.vision" QUOTE = "USDT" async def fetch_24h(client: httpx.AsyncClient, symbol: str) -> dict | None: r = await client.get(f"{API}/api/v3/ticker/24hr", params={"symbol": symbol + QUOTE}) if r.status_code != 200: return None return r.json() 

@router.message(Command("price")) async def price_cmd(message: Message, command: CommandObject) -> None: symbols = [s.upper() for s in (command.args or "BTC").split()][:5] lines = [] async with httpx.AsyncClient(timeout=10) as client: for s in symbols: if not SYMBOL_RE.match(s): lines.append(f"{s}: not a valid symbol") continue d = await fetch_24h(client, s) if not d: lines.append(f"{s}: not found") continue chg = float(d["priceChangePercent"]) lines.append(f"{s}: {fmt_price(float(d['lastPrice']))} ({chg:+.2f}% 24h)") await message.answer("\n".join(lines)) 

async def alert_loop(bot: Bot) -> None: async with httpx.AsyncClient(timeout=10) as client: while True: await asyncio.sleep(CHECK_EVERY) for symbol in {a.symbol for a in ALERTS}: try: d = await fetch_24h(client, symbol) except httpx.HTTPError: continue if not d: continue price = float(d["lastPrice"]) for a in [a for a in ALERTS if a.symbol == symbol]: hit = price >= a.target if a.direction == "above" else price <= a.target if hit: ALERTS.remove(a) await bot.send_message( a.chat_id, f"🔔 {symbol} is {a.direction} {fmt_price(a.target)}: now {fmt_price(price)}", ) 

from __future__ import annotations import asyncio import logging import os import re from dataclasses import dataclass import httpx from aiogram import Bot, Dispatcher, Router from aiogram.filters import Command, CommandObject from aiogram.types import Message logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logging.getLogger("httpx").setLevel(logging.WARNING) API = os.getenv("BINANCE_API_BASE", "https://data-api.binance.vision") QUOTE = "USDT" CHECK_EVERY = int(os.getenv("ALERT_CHECK_SECONDS", "30")) SYMBOL_RE = re.compile(r"^[A-Z0-9]{2,10}$") router = Router() @dataclass class Alert: chat_id: int symbol: str direction: str # "above" | "below"  target: float ALERTS: list[Alert] = [] async def fetch_24h(client: httpx.AsyncClient, symbol: str) -> dict | None: r = await client.get(f"{API}/api/v3/ticker/24hr", params={"symbol": symbol + QUOTE}) if r.status_code != 200: return None return r.json() def fmt_price(p: float) -> str: return f"${p:,.2f}" if p >= 1 else f"${p:.6f}" @router.message(Command("start", "help")) async def help_cmd(message: Message) -> None: await message.answer( "Free Crypto Price Bot\n\n" "/price BTC ETH - live price\n" "/alert BTC above 70000 - price alert\n" "/alerts - your alerts\n\n" "Data: Binance public market data. Not financial advice." ) @router.message(Command("price")) async def price_cmd(message: Message, command: CommandObject) -> None: symbols = [s.upper() for s in (command.args or "BTC").split()][:5] lines = [] async with httpx.AsyncClient(timeout=10) as client: for s in symbols: if not SYMBOL_RE.match(s): lines.append(f"{s}: not a valid symbol") continue d = await fetch_24h(client, s) if not d: lines.append(f"{s}: not found") continue chg = float(d["priceChangePercent"]) lines.append(f"{s}: {fmt_price(float(d['lastPrice']))} ({chg:+.2f}% 24h)") await message.answer("\n".join(lines)) @router.message(Command("alert")) async def alert_cmd(message: Message, command: CommandObject) -> None: parts = (command.args or "").split() if len(parts) != 3 or parts[1].lower() not in ("above", "below"): await message.answer("Usage: /alert BTC above 70000") return try: target = float(parts[2].replace(",", "")) except ValueError: await message.answer("The price must be a number, e.g. /alert ETH below 3000") return symbol = parts[0].upper() if not SYMBOL_RE.match(symbol): await message.answer("That doesn't look like a coin symbol.") return if sum(a.chat_id == message.chat.id for a in ALERTS) >= 5: await message.answer("Max 5 alerts per chat.") return ALERTS.append(Alert(message.chat.id, symbol, parts[1].lower(), target)) await message.answer(f"OK, I'll tell you when {symbol} goes {parts[1].lower()} {fmt_price(target)}.") @router.message(Command("alerts")) async def alerts_cmd(message: Message) -> None: mine = [a for a in ALERTS if a.chat_id == message.chat.id] if not mine: await message.answer("No alerts. Try /alert BTC above 70000") return await message.answer("\n".join(f"{a.symbol} {a.direction} {fmt_price(a.target)}" for a in mine)) async def alert_loop(bot: Bot) -> None: async with httpx.AsyncClient(timeout=10) as client: while True: await asyncio.sleep(CHECK_EVERY) for symbol in {a.symbol for a in ALERTS}: try: d = await fetch_24h(client, symbol) except httpx.HTTPError: continue if not d: continue price = float(d["lastPrice"]) for a in [a for a in ALERTS if a.symbol == symbol]: hit = price >= a.target if a.direction == "above" else price <= a.target if hit: ALERTS.remove(a) await bot.send_message( a.chat_id, f"🔔 {symbol} is {a.direction} {fmt_price(a.target)}: now {fmt_price(price)}", ) async def main() -> None: token = os.getenv("TELEGRAM_BOT_TOKEN", "").strip() if not token: raise SystemExit("Set TELEGRAM_BOT_TOKEN (get one from @BotFather).") bot = Bot(token=token) dp = Dispatcher() dp.include_router(router) asyncio.create_task(alert_loop(bot)) await dp.start_polling(bot) if __name__ == "__main__": asyncio.run(main())