Build a stock and crypto price alert bot for Discord and Telegram

"""Stock and crypto price alerts for Discord and Telegram from one WebSocket."""
import asyncio
import json
import logging
import os
import random
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from urllib.parse import quote

import requests
import websockets  # pip install websockets requests

API = "https://api.tickerlayer.com"
KEY = os.environ["TICKERLAYER_API_KEY"]
STREAM = "wss://stream.tickerlayer.com/?apiKey=" + quote(KEY, safe="")
HEADERS = {"x-api-key": KEY}
log = logging.getLogger("alerts")


@dataclass
class Rule:
    symbol: str
    kind: str  # "above", "below" or "move" (percent versus previous close)
    value: float
    cooldown_s: int = 1800
    last_fired: float = 0.0

    def check(self, prev, price, ref):
        if time.time() - self.last_fired < self.cooldown_s:
            return None
        if self.kind == "above" and prev is not None and prev < self.value <= price:
            text = f"{self.symbol} crossed above {self.value:,.2f} (last {price:,.2f})"
        elif self.kind == "below" and prev is not None and prev > self.value >= price:
            text = f"{self.symbol} crossed below {self.value:,.2f} (last {price:,.2f})"
        elif self.kind == "move" and ref:
            pct = (price - ref) / ref * 100
            if abs(pct) < self.value:
                return None
            text = f"{self.symbol} {pct:+.2f}% vs previous close {ref:,.2f} (last {price:,.2f})"
        else:
            return None
        self.last_fired = time.time()
        return text


RULES = [
    Rule("BTCUSD", "above", 90_000),
    Rule("BTCUSD", "move", 1.5),
    Rule("ETHUSD", "move", 3.0),
    Rule("US:KO", "below", 87.50),
    Rule("US:KO", "move", 2.0),
]

last_price = {}  # symbol -> last trade price seen
prev_close = {}  # symbol -> reference price for "move" rules
us_open = False  # True only during the regular US session

def asset_of(symbol):
    return "stocks" if ":" in symbol else "crypto"


def load_prev_closes():
    for symbol in sorted({r.symbol for r in RULES}):
        resp = requests.get(f"{API}/{asset_of(symbol)}/snapshot/{symbol}", headers=HEADERS, timeout=10)
        resp.raise_for_status()
        prev_close[symbol] = float(resp.json()["prev_close"])
    log.info("reference closes: %s", prev_close)


def us_status():
    resp = requests.get(
        f"{API}/markets/status", params={"asset": "stocks", "market": "US"}, headers=HEADERS, timeout=10
    )
    resp.raise_for_status()
    return resp.json()


def seconds_to_next_boundary(status):
    now = time.time()
    waits = [3600.0]  # re-check at least hourly
    for key in ("next_open", "next_close"):
        if status.get(key):
            at = datetime.fromisoformat(status[key].replace("Z", "+00:00")).timestamp()
            if at > now:
                waits.append(at - now + 2)
    return min(waits)


async def watch_market_hours():
    """One status call per session change, not one per minute."""
    global us_open
    loaded_for = None
    while True:
        try:
            status = await asyncio.to_thread(us_status)
            was_open, us_open = us_open, status.get("status") == "open"
            today = datetime.now(timezone.utc).date()
            if (us_open and not was_open) or loaded_for != today:
                await asyncio.to_thread(load_prev_closes)
                loaded_for = today
            wait = seconds_to_next_boundary(status)
            log.info("US stocks %s; next check in %.0f min", status.get("status"), wait / 60)
        except requests.RequestException as exc:
            log.warning("market status failed: %s", type(exc).__name__)
            wait = 60
        await asyncio.sleep(wait)

def on_trade(msg, outbox):
    symbol, price = msg["symbol"], float(msg["price"])  # numbers arrive as strings
    prev = last_price.get(symbol)
    last_price[symbol] = price
    if msg.get("snapshot"):
        return  # a replayed last value seeds state; never alert on it
    if msg.get("asset") == "stocks" and not us_open:
        return  # pre-market, after-hours, weekends and holidays stay quiet
    for rule in RULES:
        if rule.symbol == symbol:
            text = rule.check(prev, price, prev_close.get(symbol))
            if text:
                outbox.put_nowait(text)


async def stream(outbox):
    subs = {}
    for rule in RULES:
        subs.setdefault(f"{asset_of(rule.symbol)}.trades", set()).add(rule.symbol)
    backoff = 1.0
    while True:
        try:
            async with websockets.connect(STREAM, compression=None, open_timeout=10) as ws:
                ready = json.loads(await ws.recv())
                if ready.get("event") != "ready":
                    raise RuntimeError(f"expected ready, got {ready}")
                for channel, symbols in subs.items():
                    await ws.send(json.dumps({"action": "subscribe", "channels": [channel], "symbols": sorted(symbols)}))
                backoff = 1.0
                async for raw in ws:
                    msg = json.loads(raw)
                    if msg.get("type") == "trade":
                        on_trade(msg, outbox)
                    elif msg.get("type") == "error":
                        log.error("stream error: %s %s", msg.get("code"), msg.get("message"))
        except websockets.InvalidStatus as exc:
            if exc.response.status_code in (401, 403):
                raise  # wrong key or no WebSocket entitlement: retrying will not help
            log.warning("upgrade refused with HTTP %s", exc.response.status_code)
        except (websockets.ConnectionClosed, OSError, asyncio.TimeoutError) as exc:
            log.warning("disconnected: %s", type(exc).__name__)
        await asyncio.sleep(backoff + random.uniform(0, backoff / 2))
        backoff = min(backoff * 2, 30.0)

class Discord:
    def __init__(self, webhook_url):
        self.url = webhook_url

    def send(self, text):
        for _ in range(3):
            resp = requests.post(self.url, json={"content": text[:2000]}, timeout=10)
            if resp.status_code == 429:
                time.sleep(float(resp.json().get("retry_after", 1)))
                continue
            resp.raise_for_status()
            return


class Telegram:
    def __init__(self, token, chat_id):
        self.url = f"https://api.telegram.org/bot{token}/sendMessage"
        self.chat_id = chat_id

    def send(self, text):
        for _ in range(3):
            resp = requests.post(self.url, json={"chat_id": self.chat_id, "text": text}, timeout=10)
            if resp.status_code == 429:
                time.sleep(float(resp.json().get("parameters", {}).get("retry_after", 1)))
                continue
            resp.raise_for_status()
            return


class Console:
    def send(self, text):
        print("ALERT", text, flush=True)


def sinks_from_env():
    sinks = []
    if os.environ.get("DISCORD_WEBHOOK_URL"):
        sinks.append(Discord(os.environ["DISCORD_WEBHOOK_URL"]))
    if os.environ.get("TELEGRAM_BOT_TOKEN") and os.environ.get("TELEGRAM_CHAT_ID"):
        sinks.append(Telegram(os.environ["TELEGRAM_BOT_TOKEN"], os.environ["TELEGRAM_CHAT_ID"]))
    return sinks or [Console()]  # no chat configured: dry run to the terminal


async def deliver(outbox, sinks):
    while True:
        text = await outbox.get()
        for sink in sinks:
            try:
                await asyncio.to_thread(sink.send, text)
            except requests.RequestException as exc:
                # Log the status only: the exception text contains the webhook URL or bot token.
                status = getattr(exc.response, "status_code", None)
                log.error("%s delivery failed (HTTP %s)", type(sink).__name__, status)


async def main():
    logging.Formatter.converter = time.gmtime  # log in UTC
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    outbox = asyncio.Queue()
    await asyncio.gather(watch_market_hours(), stream(outbox), deliver(outbox, sinks_from_env()))


if __name__ == "__main__":
    asyncio.run(main())

2026-09-28 10:56:11,703 INFO reference closes: {'BTCUSD': 84472.0, 'ETHUSD': 2688.65, 'US:KO': 87.81}
2026-09-28 10:56:11,703 INFO US stocks pre_market; next check in 60 min
ALERT BTCUSD -1.76% vs previous close 84,472.00 (last 82,985.21)