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
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
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()
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))
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)}",
)
Two details worth copying into any bot:
- Dedupe outbound calls. The set comprehension means 1,000 users watching BTC still costs one request per cycle.
- 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())
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
/helpdoes 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)