Python trading bot for crypto: from live WebSocket data to signals

"""Paper-trading signal engine on a live crypto stream. Educational: it never places an order."""
import asyncio
import csv
import json
import logging
import os
import random
import time
from collections import deque
from datetime import datetime, timedelta, 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="")
SYMBOL = "BTCUSD"
BAR_MINUTES = 1  # 1, 5 or 15
BAR_MS = BAR_MINUTES * 60_000
FAST, SLOW, ATR_N = 9, 21, 14

START_EQUITY = 10_000.0
RISK_PER_TRADE = 0.005  # lose at most 0.5% of equity if the stop is hit
STOP_ATR = 2.0  # stop distance = 2 x ATR(14)
MAX_EXPOSURE = 0.25  # never hold more than 25% of equity in the position
FEE_BPS = 10  # assumed fee per side, in basis points
MAX_SPREAD_BPS = 5.0
MAX_QUOTE_AGE_MS = 5_000
DAILY_LOSS_LIMIT = 0.02  # no new entries after a 2% loss on the UTC day

logging.Formatter.converter = time.gmtime  # log in UTC
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(message)s",
    handlers=[logging.StreamHandler(), logging.FileHandler("paper_bot.log")],
)
log = logging.getLogger("bot")

class Indicators:
    """Incremental EMA(FAST), EMA(SLOW) and Wilder ATR(ATR_N) over closed bars."""

    def __init__(self):
        self.n, self.window = 0, deque(maxlen=SLOW)
        self.fast = self.slow = self.atr = self.prev_close = None

    def update(self, bar):
        h, l, c = bar["h"], bar["l"], bar["c"]
        tr = h - l if self.prev_close is None else max(h - l, abs(h - self.prev_close), abs(l - self.prev_close))
        self.atr = tr if self.atr is None else (self.atr * (ATR_N - 1) + tr) / ATR_N
        self.prev_close = c
        self.n += 1
        self.window.append(c)
        self.fast = self._ema(self.fast, c, FAST)
        self.slow = self._ema(self.slow, c, SLOW)

    def _ema(self, prev, price, n):
        if self.n < n:
            return None
        if prev is None:
            return sum(list(self.window)[-n:]) / n  # seed with a simple average
        return prev + 2 / (n + 1) * (price - prev)

    @property
    def ready(self):
        return self.n > SLOW + ATR_N

class PaperBroker:
    def __init__(self):
        self.cash, self.qty, self.stop = START_EQUITY, 0.0, None
        self.day, self.day_start_equity = None, START_EQUITY
        self.file = open("paper_trades.csv", "a", newline="")
        self.journal = csv.writer(self.file)

    def equity(self, bid):
        return self.cash + self.qty * bid

    def buy(self, qty, ask, stop, ts):
        cost = qty * ask * (1 + FEE_BPS / 10_000)
        self.cash -= cost
        self.qty, self.stop = qty, stop
        self._record("BUY", qty, ask, ts)

    def sell(self, bid, ts, why):
        self.cash += self.qty * bid * (1 - FEE_BPS / 10_000)
        self._record(f"SELL ({why})", self.qty, bid, ts)
        self.qty, self.stop = 0.0, None

    def _record(self, side, qty, price, ts):
        when = datetime.fromtimestamp(ts / 1000, tz=timezone.utc).isoformat()
        self.journal.writerow([when, SYMBOL, side, f"{qty:.6f}", f"{price:.2f}", f"{self.cash:.2f}"])
        self.file.flush()
        log.info("%s %.6f %s at %.2f, cash %.2f", side, qty, SYMBOL, price, self.cash)

class Engine:
    def __init__(self):
        self.ind, self.broker = Indicators(), PaperBroker()
        self.bar, self.bid, self.ask, self.quote_ts, self.first_live_bar = None, None, None, 0, 0

    def warm_up(self):
        """Rebuild indicators from the latest closed bars (one REST call)."""
        now = datetime.now(timezone.utc)
        start = (now - timedelta(days=2)).date().isoformat()
        resp = requests.get(
            f"{API}/crypto/agg/{SYMBOL}/{BAR_MINUTES}/minute/{start}/{now.date().isoformat()}",
            params={"sort": "desc", "limit": 120}, headers={"x-api-key": KEY}, timeout=15)
        resp.raise_for_status()
        this_minute = int(now.timestamp() * 1000) // BAR_MS * BAR_MS
        closed = [b for b in reversed(resp.json()["results"]) if b["t"] < this_minute]
        self.ind = Indicators()
        for bar in closed:
            self.ind.update(bar)
        # We join the current minute part-way through, so its bar would be incomplete:
        # skip it and build the first live bar from the next full minute.
        self.bar, self.first_live_bar = None, this_minute + BAR_MS
        log.info("warm-up: %d closed bars, EMA%d %.2f, EMA%d %.2f, ATR%d %.2f",
                 len(closed), FAST, self.ind.fast, SLOW, self.ind.slow, ATR_N, self.ind.atr)

    def on_quote(self, msg):
        self.bid, self.ask, self.quote_ts = float(msg["bid"]), float(msg["ask"]), msg["ts"]

    def on_trade(self, msg):
        price, ts = float(msg["price"]), msg["ts"]  # trade ts is the event time
        if self.broker.qty and price <= self.broker.stop and self.bid:
            self.broker.sell(self.bid, ts, "stop")
        start = ts // BAR_MS * BAR_MS
        if start < self.first_live_bar:
            return
        if self.bar is None or start > self.bar["t"]:
            if self.bar is not None:
                self.on_bar_close(self.bar, ts)
            self.bar = {"t": start, "o": price, "h": price, "l": price, "c": price}
        elif start == self.bar["t"]:
            self.bar["h"], self.bar["l"] = max(self.bar["h"], price), min(self.bar["l"], price)
            self.bar["c"] = price

    def on_bar_close(self, bar, ts):
        was_above = self.ind.fast is not None and self.ind.fast > self.ind.slow
        self.ind.update(bar)
        if not self.ind.ready:
            return
        is_above = self.ind.fast > self.ind.slow
        stamp = datetime.fromtimestamp(bar["t"] / 1000, tz=timezone.utc).strftime("%H:%M")
        log.info("bar %s close %.2f  EMA%d %.2f  EMA%d %.2f  ATR %.2f  qty %.6f",
                 stamp, bar["c"], FAST, self.ind.fast, SLOW, self.ind.slow, self.ind.atr, self.broker.qty)
        if is_above and not was_above and not self.broker.qty:
            self.try_enter(ts)
        elif was_above and not is_above and self.broker.qty and self.bid:
            self.broker.sell(self.bid, ts, "cross down")

    def try_enter(self, ts):
        b = self.broker
        if not self.ask or time.time() * 1000 - self.quote_ts > MAX_QUOTE_AGE_MS:
            log.info("skip entry: no fresh quote")
            return
        spread_bps = (self.ask - self.bid) / ((self.ask + self.bid) / 2) * 10_000
        if spread_bps > MAX_SPREAD_BPS:
            log.info("skip entry: spread %.1f bps", spread_bps)
            return
        day = datetime.now(timezone.utc).date()
        if b.day != day:
            b.day, b.day_start_equity = day, b.equity(self.bid)
        equity = b.equity(self.bid)
        if equity < b.day_start_equity * (1 - DAILY_LOSS_LIMIT):
            log.info("skip entry: daily loss limit reached")
            return
        stop_distance = STOP_ATR * self.ind.atr
        if stop_distance <= 0:
            return
        qty = min(equity * RISK_PER_TRADE / stop_distance, equity * MAX_EXPOSURE / self.ask)
        b.buy(qty, self.ask, self.ask - stop_distance, ts)

async def run():
    engine, backoff = Engine(), 1.0
    while True:
        try:
            engine.warm_up()  # also closes any gap left by a disconnect
            async with websockets.connect(STREAM, compression=None, open_timeout=10) as ws:
                if json.loads(await ws.recv()).get("event") != "ready":
                    raise RuntimeError("no ready frame")
                await ws.send(json.dumps({"action": "subscribe",
                                          "channels": ["crypto.quotes", "crypto.trades"], "symbols": [SYMBOL]}))
                backoff = 1.0
                async for raw in ws:
                    msg = json.loads(raw)
                    if msg.get("type") == "quote":
                        engine.on_quote(msg)
                    elif msg.get("type") == "trade":
                        engine.on_trade(msg)
                    elif msg.get("type") == "error":
                        log.error("stream error: %s", msg.get("code"))
        except websockets.InvalidStatus as exc:
            if exc.response.status_code in (401, 403):
                raise
            log.warning("upgrade refused: HTTP %s", exc.response.status_code)
        except (websockets.ConnectionClosed, OSError, asyncio.TimeoutError, requests.RequestException) as exc:
            log.warning("connection problem: %s", type(exc).__name__)
        await asyncio.sleep(backoff + random.uniform(0, backoff / 2))
        backoff = min(backoff * 2, 30.0)


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

2026-09-28 10:58:51,345 warm-up: 119 closed bars, EMA9 82966.95, EMA21 82935.04, ATR14 37.70
# from here, one "bar" line per closed minute, and BUY or SELL lines when a rule fires

"""Replay synthetic frames through the engine: no network, no key needed."""
import os
import time

os.environ.setdefault("TICKERLAYER_API_KEY", "offline")
import paper_bot as pb  # noqa: E402

engine = pb.Engine()
now_ms = int(time.time() * 1000)
t0 = now_ms // 60_000 * 60_000 - 100 * 60_000
price = 83_000.0
for i in range(60):  # a slow decline seeds the indicators with EMA9 below EMA21
    price -= 5
    engine.ind.update({"t": t0 + i * 60_000, "h": price + 10, "l": price - 10, "c": price})


def tick(p, ts):
    engine.on_quote({"bid": f"{p - 0.01:.2f}", "ask": f"{p:.2f}", "ts": int(time.time() * 1000)})
    engine.on_trade({"price": f"{p:.2f}", "ts": ts})


start = t0 + 60 * 60_000
for m in range(40):  # 20 minutes of rally, then a sharp drop
    move = m * 8 if m < 20 else 160 - (m - 20) * 25
    for k in range(6):
        tick(price + move + k, start + m * 60_000 + k * 5_000)

print(open("paper_trades.csv").read().strip())

11:17:19 BUY 0.030201 BTCUSD at 82780.00, cash 7497.50
11:17:19 SELL (stop) 0.030201 BTCUSD at 82734.99, cash 9993.64
2026-09-28T10:47:00+00:00,BTCUSD,BUY,0.030201,82780.00,7497.50
2026-09-28T11:02:00+00:00,BTCUSD,SELL (stop),0.030201,82734.99,9993.64