Crypto Trading Bot Development

Crypto Trading Bot Development

A crypto bot is not magic and not guaranteed profit. It's an automated system for executing a trading strategy. Good strategy + bad implementation = money lost. Bad strategy + good implementation = money lost slowly. Let's cover how to build a production-ready bot that doesn't lose funds for technical reasons.

Bot Architecture

Any bot consists of several independent layers:

Data layer — get market data. REST for historical, WebSocket for real-time. Normalize data from different exchanges into unified format.

Strategy layer — compute signals. Takes candles/orderbook/data, returns: BUY / SELL / HOLD with volume.

Execution layer — place orders. Gets signal, places via exchange API, tracks execution.

Risk management layer — system-level limits: maximum drawdown, maximum position, portfolio stop-loss.

Persistence layer — save state, trades, P&L.

interface TradingBot {
  dataFeed: DataFeed;
  strategy: Strategy;
  executor: OrderExecutor;
  riskManager: RiskManager;
  state: BotState;
}

// Main cycle
async function runBotCycle(bot: TradingBot) {
  const candles = await bot.dataFeed.getLatestCandles(bot.config.pair, bot.config.timeframe);
  const signal = bot.strategy.calculate(candles);
  
  if (!bot.riskManager.allowsSignal(signal, bot.state)) {
    return; // risk control blocked
  }
  
  if (signal.action !== 'HOLD') {
    await bot.executor.execute(signal, bot.state);
  }
}

Exchange Connection via CCXT

CCXT (CryptoCurrency eXchange Trading Library) — standard library connecting to 100+ exchanges with unified API. Python, JavaScript, PHP.

import ccxt
import asyncio

exchange = ccxt.binance({
    'apiKey': API_KEY,
    'secret': API_SECRET,
    'options': {
        'defaultType': 'spot',  # spot / future / margin
    },
    'enableRateLimit': True,  # built-in rate limiter
})

async def fetch_ohlcv(symbol: str, timeframe: str, limit: int = 200):
    ohlcv = await exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    # [[timestamp, open, high, low, close, volume], ...]
    return pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

CCXT handles rate limits, normalizes response formats, maps error codes. For multi-exchange strategies — switch exchanges with one line.

Strategy Types

Trend Following

Follow the trend: buy on uptrend, sell on downtrend. Examples: EMA crossover, ADX + DI.

import pandas_ta as ta

def ema_crossover_signal(df: pd.DataFrame, fast: int = 9, slow: int = 21) -> str:
    df['ema_fast'] = ta.ema(df['close'], length=fast)
    df['ema_slow'] = ta.ema(df['close'], length=slow)
    
    prev_diff = df['ema_fast'].iloc[-2] - df['ema_slow'].iloc[-2]
    curr_diff = df['ema_fast'].iloc[-1] - df['ema_slow'].iloc[-1]
    
    if prev_diff < 0 and curr_diff > 0:
        return 'BUY'   # golden cross
    elif prev_diff > 0 and curr_diff < 0:
        return 'SELL'  # death cross
    return 'HOLD'

Mean Reversion

Assumes price returns to average: sell on overbought, buy on oversold. RSI, Bollinger Bands. Works in sideways, loses on trends.

Statistical Arbitrage

Find cointegrated pairs (BTC/ETH historically move together). On spread divergence — long laggard, short leader. Requires constant cointegration monitoring (unstable).

Risk Management

More important than the strategy. Without it, any strategy eventually blows up account.

class RiskManager:
    def __init__(self, config: RiskConfig):
        self.max_position_pct = config.max_position_pct  # % of deposit
        self.max_daily_loss_pct = config.max_daily_loss_pct
        self.max_drawdown_pct = config.max_drawdown_pct
        self.daily_pnl = 0
        self.peak_balance = None
    
    def calculate_position_size(self, balance: float, price: float, stop_price: float) -> float:
        # Kelly criterion / fixed fractional
        risk_per_trade = balance * (self.max_position_pct / 100)
        price_risk = abs(price - stop_price) / price
        
        if price_risk == 0:
            return 0
        
        position_value = risk_per_trade / price_risk
        return min(position_value, balance * 0.3)  # max 30% in one position
    
    def check_circuit_breaker(self, current_balance: float) -> bool:
        if self.peak_balance is None:
            self.peak_balance = current_balance
        
        drawdown = (self.peak_balance - current_balance) / self.peak_balance * 100
        daily_loss = self.daily_pnl / self.peak_balance * 100
        
        if drawdown > self.max_drawdown_pct:
            return False  # stop trading
        if daily_loss < -self.max_daily_loss_pct:
            return False  # stop trading
        
        return True

Mandatory risk manager parameters:

• Max position: % of deposit per trade
• Daily loss limit: bot stops at X% daily loss
• Max drawdown: global stop at peak decline
• Stop-loss: per trade

Backtesting

Bot without backtesting is gambling. Backtrader (Python) or backtesting.py for simple start, Vectorbt for fast vectorized testing.

from backtesting import Backtest, Strategy
from backtesting.lib import crossover
import backtesting.lib as lib

class EMAStrategy(Strategy):
    fast = 9
    slow = 21
    
    def init(self):
        close = self.data.Close
        self.ema_fast = self.I(lib.EMA, close, self.fast)
        self.ema_slow = self.I(lib.EMA, close, self.slow)
    
    def next(self):
        if crossover(self.ema_fast, self.ema_slow):
            self.buy(size=0.95)
        elif crossover(self.ema_slow, self.ema_fast):
            self.position.close()

bt = Backtest(data, EMAStrategy, cash=10000, commission=0.001)
stats = bt.run()
print(stats[['Return [%]', 'Sharpe Ratio', 'Max. Drawdown [%]', 'Win Rate [%]']])

Key metrics:

• Sharpe Ratio > 1.5 — acceptable
• Max Drawdown — how much bot could lose from peak
• Win Rate — % profitable trades (not main metric)
• Profit Factor = gross profit / gross loss (> 1.5 good)

Warning on overfitting: strategy with 20+ parameters optimized on 2 years data likely overfitted. Test on out-of-sample data.

Deployment and Monitoring

Production bot must:

• Run 24/7 on VPS/cloud (minimum 2 CPU, 4 GB RAM, SSD)
• Auto-restart on crash (systemd, Docker + restart policy)
• Log all actions with timestamp
• Send Telegram alerts on errors or anomalies
• Have web dashboard with status, P&L, open positions

# Telegram alert on critical error
async def send_alert(message: str, level: str = 'INFO'):
    bot = telegram.Bot(token=TELEGRAM_TOKEN)
    prefix = {'INFO': 'ℹ', 'WARNING': '⚠️', 'ERROR': '🔴', 'CRITICAL': '🚨'}
    await bot.send_message(
        chat_id=CHAT_ID,
        text=f"{prefix.get(level, '')} {level}\n{message}\n\nBot: {BOT_NAME}\nTime: {datetime.utcnow()}"
    )

Developing a bot for specific strategy: 3–6 weeks including backtesting, exchange integration, and monitoring setup.