Building a Crypto Trading Bot in 2026: Complete Technical Guide

import asyncio
import aiohttp
import numpy as np
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Deque
import time
import logging
import json

logger = logging.getLogger("pipeline")

@dataclass
class Candle:
    timestamp: int
    open: float
    high: float
    low: float
    close: float
    volume: float
    complete: bool = True

@dataclass
class OrderBook:
    bids: List[List[float]]  # [[price, size], ...]
    asks: List[List[float]]  # [[price, size], ...]
    timestamp: float

    @property
    def mid_price(self) -> float:
        return (self.bids[0][0] + self.asks[0][0]) / 2

    @property
    def spread_bps(self) -> float:
        spread = self.asks[0][0] - self.bids[0][0]
        return (spread / self.mid_price) * 10000

    def depth_at_bps(self, bps: float) -> tuple[float, float]:
        """Total bid/ask volume within bps of mid price."""
        mid = self.mid_price
        threshold = mid * bps / 10000
        bid_depth = sum(s for p, s in self.bids if mid - p <= threshold)
        ask_depth = sum(s for p, s in self.asks if p - mid <= threshold)
        return bid_depth, ask_depth

class MarketDataFeed:
    """
    Resilient market data manager for Purple Flea Trading API.
    Handles reconnection, sequence gap detection, and data staleness.
    """

    def __init__(self, api_key: str, symbols: List[str]):
        self.api_key = api_key
        self.symbols = symbols
        self.base_url = "https://purpleflea.com/trading-api"
        self.ws_url = "wss://purpleflea.com/trading-api/ws"

        # Storage: rolling windows
        self.candles: Dict[str, Deque[Candle]] = {
            s: deque(maxlen=500) for s in symbols
        }
        self.books: Dict[str, Optional[OrderBook]] = {s: None for s in symbols}
        self.last_update: Dict[str, float] = {s: 0 for s in symbols}

        # Callbacks by event type
        self._callbacks: Dict[str, List[Callable]] = {
            'candle': [], 'book': [], 'trade': [], 'stale': []
        }

        self._running = False
        self._reconnect_delay = 1.0

    def on(self, event: str, callback: Callable):
        self._callbacks[event].append(callback)

    async def _emit(self, event: str, data):
        for cb in self._callbacks[event]:
            try:
                if asyncio.iscoroutinefunction(cb):
                    await cb(data)
                else:
                    cb(data)
            except Exception as e:
                logger.error(f"Callback error in {event}: {e}")

    async def load_historical(self, symbol: str, limit: int = 200):
        """Warm up candle buffers with historical data on startup."""
        url = f"{self.base_url}/candles/{symbol}?interval=1m&limit={limit}"
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers={"X-API-Key": self.api_key}) as resp:
                data = await resp.json()

        for c in data['candles']:
            self.candles[symbol].append(Candle(
                timestamp=c['t'], open=c['o'], high=c['h'],
                low=c['l'], close=c['c'], volume=c['v']
            ))
        logger.info(f"Loaded {len(data['candles'])} candles for {symbol}")

    async def _ws_listener(self):
        while self._running:
            try:
                subscribe_msg = {
                    "action": "subscribe",
                    "channels": ["candle_1m", "orderbook_l2", "trades"],
                    "symbols": self.symbols,
                }
                async with aiohttp.ClientSession() as session:
                    async with session.ws_connect(
                        self.ws_url,
                        headers={"X-API-Key": self.api_key},
                        heartbeat=30,
                    ) as ws:
                        await ws.send_json(subscribe_msg)
                        self._reconnect_delay = 1.0  # Reset on success

                        async for msg in ws:
                            if msg.type == aiohttp.WSMsgType.TEXT:
                                await self._handle_message(json.loads(msg.data))

            except Exception as e:
                logger.warning(f"WS disconnected: {e}. Reconnecting in {self._reconnect_delay}s")
                await asyncio.sleep(self._reconnect_delay)
                self._reconnect_delay = min(self._reconnect_delay * 2, 30)

    async def _handle_message(self, msg: dict):
        ch = msg.get('channel')
        symbol = msg.get('symbol')
        if not symbol: return

        self.last_update[symbol] = time.time()

        if ch == 'candle_1m':
            c = msg['data']
            candle = Candle(
                timestamp=c['t'], open=c['o'], high=c['h'],
                low=c['l'], close=c['c'], volume=c['v'],
                complete=c.get('complete', True),
            )
            if candle.complete:
                self.candles[symbol].append(candle)
            await self._emit('candle', {'symbol': symbol, 'candle': candle})

        elif ch == 'orderbook_l2':
            d = msg['data']
            self.books[symbol] = OrderBook(
                bids=d['bids'], asks=d['asks'], timestamp=time.time()
            )
            await self._emit('book', {'symbol': symbol, 'book': self.books[symbol]})

        elif ch == 'trades':
            await self._emit('trade', {'symbol': symbol, 'trade': msg['data']})

    async def _staleness_watchdog(self):
        """Alert if any symbol has not updated in >5 seconds."""
        while self._running:
            await asyncio.sleep(5)
            now = time.time()
            for symbol in self.symbols:
                age = now - self.last_update[symbol]
                if age > 10:
                    await self._emit('stale', {'symbol': symbol, 'age_secs': age})

    def closes(self, symbol: str) -> np.ndarray:
        """Return close price array for indicator computation."""
        return np.array([c.close for c in self.candles[symbol]])

    def volumes(self, symbol: str) -> np.ndarray:
        return np.array([c.volume for c in self.candles[symbol]])

    async def start(self):
        self._running = True
        for symbol in self.symbols:
            await self.load_historical(symbol)
        await asyncio.gather(
            self._ws_listener(),
            self._staleness_watchdog(),
        )

    async def stop(self):
        self._running = False

import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional
from abc import ABC, abstractmethod

@dataclass
class Signal:
    symbol: str
    direction: float   # -1.0 to +1.0, signed strength
    confidence: float  # 0.0 to 1.0
    source: str
    metadata: dict

@dataclass
class OrderIntent:
    symbol: str
    side: str           # 'buy' or 'sell'
    size_usd: float
    order_type: str    # 'market', 'limit', 'twap'
    limit_price: Optional[float]
    urgency: str        # 'low', 'normal', 'high'
    signal_sources: List[str]

class BaseSignal(ABC):
    def __init__(self, name: str, weight: float = 1.0):
        self.name = name
        self.weight = weight

    @abstractmethod
    def compute(self, closes: np.ndarray, volumes: np.ndarray, **kwargs) -> Optional[Signal]:
        pass

class EMAMomentumSignal(BaseSignal):
    """EMA crossover with volume confirmation."""

    def __init__(self, fast: int = 9, slow: int = 21, weight: float = 1.0):
        super().__init__("ema_momentum", weight)
        self.fast = fast
        self.slow = slow

    def _ema(self, data: np.ndarray, period: int) -> np.ndarray:
        ema = np.zeros_like(data)
        ema[period-1] = data[:period].mean()
        k = 2.0 / (period + 1)
        for i in range(period, len(data)):
            ema[i] = data[i] * k + ema[i-1] * (1 - k)
        return ema

    def compute(self, closes, volumes, **kwargs) -> Optional[Signal]:
        if len(closes) < self.slow + 5: return None

        fast_ema = self._ema(closes, self.fast)
        slow_ema = self._ema(closes, self.slow)
        spread = (fast_ema - slow_ema) / slow_ema  # normalized spread

        # Z-score of spread over 50 periods
        spread_window = spread[-50:]
        z = (spread[-1] - spread_window.mean()) / (spread_window.std() + 1e-8)

        # Volume confirmation: current vol vs 20-period avg
        vol_ratio = volumes[-1] / (volumes[-20:].mean() + 1e-8)

        direction = np.clip(z / 3.0, -1.0, 1.0)
        confidence = min(abs(z) / 3.0, 1.0) * min(vol_ratio, 2.0) / 2.0

        return Signal(
            symbol=kwargs.get('symbol', ''),
            direction=round(direction, 4),
            confidence=round(min(confidence, 1.0), 4),
            source=self.name,
            metadata={'z_score': round(z, 3), 'vol_ratio': round(vol_ratio, 3)},
        )

class RSIMeanReversionSignal(BaseSignal):
    """RSI-based mean reversion for ranging markets."""

    def __init__(self, period: int = 14, ob: int = 70, os: int = 30, weight=0.7):
        super().__init__("rsi_mean_rev", weight)
        self.period = period
        self.overbought = ob
        self.oversold = os

    def _rsi(self, closes: np.ndarray) -> float:
        deltas = np.diff(closes[-(  self.period + 1):])
        gains = np.where(deltas > 0, deltas, 0)
        losses = np.where(deltas < 0, -deltas, 0)
        avg_gain = gains.mean()
        avg_loss = losses.mean()
        if avg_loss == 0: return 100.0
        rs = avg_gain / avg_loss
        return 100 - (100 / (1 + rs))

    def compute(self, closes, volumes, **kwargs) -> Optional[Signal]:
        if len(closes) < self.period + 2: return None
        rsi = self._rsi(closes)

        if rsi > self.overbought:
            direction = -(min(rsi - self.overbought, 30) / 30)
            confidence = (rsi - self.overbought) / 30
        elif rsi < self.oversold:
            direction = (self.oversold - rsi) / 30
            confidence = (self.oversold - rsi) / 30
        else:
            return None  # No signal in neutral zone

        return Signal(
            symbol=kwargs.get('symbol', ''),
            direction=round(np.clip(direction, -1.0, 1.0), 4),
            confidence=round(min(confidence, 1.0), 4),
            source=self.name,
            metadata={'rsi': round(rsi, 2)},
        )

class StrategyEngine:
    """
    Combines multiple signals into order intents.
    Weighted average of signal directions, filtered by confidence.
    """

    def __init__(self, portfolio_value: float):
        self.signals: List[BaseSignal] = [
            EMAMomentumSignal(fast=9, slow=21, weight=1.0),
            RSIMeanReversionSignal(period=14, weight=0.7),
        ]
        self.portfolio_value = portfolio_value
        self.base_position_pct = 0.10  # 10% portfolio per trade

    def generate(self, symbol: str, closes, volumes) -> Optional[OrderIntent]:
        raw_signals = []
        for s in self.signals:
            sig = s.compute(closes, volumes, symbol=symbol)
            if sig and sig.confidence >= 0.3:  # Minimum confidence filter
                raw_signals.append((sig, s.weight))

        if not raw_signals: return None

        # Weighted combination
        total_weight = sum(w for _, w in raw_signals)
        combined = sum(s.direction * s.confidence * w for s, w in raw_signals)
        combined /= total_weight

        if abs(combined) < 0.15: return None  # Below minimum conviction

        size_usd = self.portfolio_value * self.base_position_pct * abs(combined)

        return OrderIntent(
            symbol=symbol,
            side='buy' if combined > 0 else 'sell',
            size_usd=round(size_usd, 2),
            order_type='limit' if abs(combined) < 0.7 else 'market',
            limit_price=None,
            urgency='high' if abs(combined) > 0.8 else 'normal',
            signal_sources=[s.source for s, _ in raw_signals],
        )