DEV Community

Steve Jefferson
Steve Jefferson

Posted on

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

Most "build a Telegram bot" tutorials stop at an echo bot. This one builds something you'd actually leave running: a bot that answers /price BTC ETH with live prices and pings you when a coin crosses a price you set.

No API keys (besides your bot token), no database, one file. Full code is below, MIT-licensed, copy it freely.

What we're building

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

Stack:

  • aiogram 3: a modern async Telegram bot framework
  • httpx: async HTTP client
  • Binance public market data (data-api.binance.vision): free, no key needed for read-only market data

1. Get a bot token

Open Telegram, message @botfather, send /newbot, pick a name and a username ending in bot. BotFather replies with a token like 123456:ABC.... Treat it like a password.

2. Install

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

Python 3.11+ recommended.

3. Fetch prices

Binance's 24h ticker endpoint gives last price and 24h % change in one call:

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()
Enter fullscreen mode Exit fullscreen mode

We validate symbols with a regex (^[A-Z0-9]{2,10}$) before sending them anywhere, so users can't inject odd strings into the request.

4. The /price command

aiogram 3 uses routers and filters. CommandObject gives you everything after the command:

@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))
Enter fullscreen mode Exit fullscreen mode

Capping at 5 symbols keeps one message from fanning out into dozens of requests.

5. Alerts with a background loop

Alerts live in a list of dataclasses. A background task checks prices every 30 seconds, and, importantly, fetches each symbol once, no matter how many users are watching it:

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)}",
                        )
Enter fullscreen mode Exit fullscreen mode

Two details worth copying into any bot:

  1. Dedupe outbound calls. The set comprehension means 1,000 users watching BTC still costs one request per cycle.
  2. Never let one failure kill the loop. A network hiccup on one symbol just skips that symbol this round.

6. Full code

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())
Enter fullscreen mode Exit fullscreen mode

Run it with python bot.py, open your bot in Telegram, and send /price BTC.

Limits of this version (and how to fix them)

  • Alerts are in memory. Restart the bot and they're gone. Swap the list for SQLite (aiosqlite) when you care.
  • Polling, not webhooks. Fine for a hobby bot; for production behind HTTPS, switch to aiogram's webhook mode.
  • Binance symbols only. Anything not listed against USDT on Binance returns "not found".
  • Not financial advice. Public data can lag; say so in your bot, like /help does here.

Going further

If you want to turn a bot like this into something people pay for (license keys locked to one Telegram account, access removed automatically when a subscription is cancelled, plus more templates and a deploy guide), I packaged that into a kit: The Python Bot Monetization Kit. Disclosure: it's mine and it's paid. The bot above is free either way.

Questions or bugs? Drop them in the comments and I'll answer.

Top comments (0)