How to Build a Perpetual Futures Trading Bot with Bybit API (2026)

Why Bybit for Futures Trading Bots?

Bybit has become one of the top platforms for algo traders:

• Maker rebates: Get paid 0.01% for providing liquidity (limit orders)
• High leverage: Up to 100x on major pairs (though 3-5x is recommended)
• Excellent API: WebSocket, REST, low rate limits
• Testnet: Full-featured testnet for bot development and testing
• Python SDK: Official pybit library maintained by Bybit

Setting Up Bybit Python SDK

pip install pybit websocket-client

from pybit.unified_trading import HTTP, WebSocket
import os

# ALWAYS test on testnet first!
session = HTTP(
    testnet=True,  # Change to False for live trading
    api_key=os.environ['BYBIT_TEST_KEY'],
    api_secret=os.environ['BYBIT_TEST_SECRET'],
)

# Check account
account_info = session.get_wallet_balance(accountType="UNIFIED")
print(f"USDT Balance: {account_info['result']['list'][0]['coin'][0]['walletBalance']}")

Getting Market Data

def get_klines(symbol: str, interval: str = '60', limit: int = 200) -> list:
    """Fetch OHLCV data from Bybit"""
    response = session.get_kline(
        category="linear",  # USDT perpetuals
        symbol=symbol,
        interval=interval,  # '1', '3', '5', '15', '30', '60', '120', '240', 'D', 'W'
        limit=limit
    )
    
    klines = response['result']['list']
    # Returns: [timestamp, open, high, low, close, volume, turnover]
    return [[float(x) for x in k] for k in reversed(klines)]

def get_orderbook(symbol: str) -> dict:
    """Get level 2 orderbook"""
    response = session.get_orderbook(
        category="linear",
        symbol=symbol,
        limit=25
    )
    return response['result']

# Get BTC perp data
btc_klines = get_klines('BTCUSDT', interval='60', limit=200)
print(f"Latest BTC close: ${btc_klines[-1][4]:,.2f}")

Strategy 1: Trend Following with EMA Crossover

import numpy as np
from typing import Optional

def calculate_ema(prices: list, period: int) -> float:
    """Calculate Exponential Moving Average"""
    k = 2 / (period + 1)
    ema = prices[0]
    for price in prices[1:]:
        ema = price * k + ema * (1 - k)
    return ema

def calculate_atr(klines: list, period: int = 14) -> float:
    """Average True Range for position sizing"""
    trs = []
    for i in range(1, len(klines)):
        high = klines[i][2]
        low = klines[i][3]
        prev_close = klines[i-1][4]
        tr = max(high - low, abs(high - prev_close), abs(low - prev_close))
        trs.append(tr)
    return sum(trs[-period:]) / period

class EMACrossStrategy:
    def __init__(self, fast=12, slow=26, signal_period=9):
        self.fast = fast
        self.slow = slow
        self.signal_period = signal_period
    
    def get_signal(self, klines: list) -> Optional[str]:
        """Return 'BUY', 'SELL', or None"""
        closes = [k[4] for k in klines]
        
        # Calculate EMAs
        ema_fast = [calculate_ema(closes[:i+self.fast], self.fast) 
                   for i in range(len(closes) - self.fast + 1)]
        ema_slow = [calculate_ema(closes[:i+self.slow], self.slow) 
                   for i in range(len(closes) - self.slow + 1)]
        
        # Align lengths
        min_len = min(len(ema_fast), len(ema_slow))
        ema_fast = ema_fast[-min_len:]
        ema_slow = ema_slow[-min_len:]
        
        # Check for crossover
        if len(ema_fast) < 2:
            return None
        
        was_above = ema_fast[-2] > ema_slow[-2]
        is_above = ema_fast[-1] > ema_slow[-1]
        
        if not was_above and is_above:
            return 'BUY'  # Bullish crossover
        elif was_above and not is_above:
            return 'SELL'  # Bearish crossover
        
        return None

Placing Orders on Bybit

def get_position(symbol: str) -> Optional[dict]:
    """Get current position for a symbol"""
    response = session.get_positions(
        category="linear",
        symbol=symbol
    )
    positions = response['result']['list']
    if positions and float(positions[0]['size']) > 0:
        return positions[0]
    return None

def open_long(symbol: str, usdt_amount: float, leverage: int = 3) -> dict:
    """Open a long position with proper risk management"""
    
    # Set leverage first
    session.set_leverage(
        category="linear",
        symbol=symbol,
        buyLeverage=str(leverage),
        sellLeverage=str(leverage)
    )
    
    # Get current price
    ticker = session.get_tickers(category="linear", symbol=symbol)
    current_price = float(ticker['result']['list'][0]['lastPrice'])
    
    # Calculate quantity
    quantity = (usdt_amount * leverage) / current_price
    
    # Round to Bybit's required precision
    qty_step = get_qty_step(symbol)  # e.g., 0.001 for BTCUSDT
    quantity = round(quantity / qty_step) * qty_step
    
    # Calculate stop loss (2% below entry)
    stop_loss = current_price * 0.98
    take_profit = current_price * 1.04  # 4% take profit (2:1 R:R)
    
    order = session.place_order(
        category="linear",
        symbol=symbol,
        side="Buy",
        orderType="Market",
        qty=str(quantity),
        stopLoss=str(round(stop_loss, 2)),
        takeProfit=str(round(take_profit, 2)),
        slTriggerBy="MarkPrice",
        tpTriggerBy="MarkPrice",
    )
    
    print(f"✅ Opened {leverage}x LONG {quantity} {symbol} @ ${current_price:,.2f}")
    print(f"   SL: ${stop_loss:,.2f} | TP: ${take_profit:,.2f}")
    
    return order

def close_position(symbol: str) -> dict:
    """Close all positions for a symbol"""
    position = get_position(symbol)
    if not position:
        print(f"No position to close for {symbol}")
        return {}
    
    side = "Sell" if position['side'] == 'Buy' else "Buy"
    
    order = session.place_order(
        category="linear",
        symbol=symbol,
        side=side,
        orderType="Market",
        qty=position['size'],
        reduceOnly=True
    )
    
    print(f"✅ Closed {position['side']} position for {symbol}")
    return order

Real-Time WebSocket for Price Feeds

from pybit.unified_trading import WebSocket

def handle_trade_message(message):
    """Process real-time trade updates"""
    for trade in message['data']:
        price = float(trade['p'])
        size = float(trade['v'])
        side = trade['S']  # 'Buy' or 'Sell'
        
        # Large trade detection
        if size * price > 100_000:  # > $100K trade
            print(f"🐋 Large {side}: {size} BTC @ ${price:,.2f} (${size*price:,.0f})")

# Subscribe to BTC real-time trades
ws = WebSocket(
    testnet=False,
    channel_type="linear",
)

ws.trade_stream(
    symbol="BTCUSDT",
    callback=handle_trade_message
)

Funding Rate Arbitrage Bot

def get_funding_rate(symbol: str) -> dict:
    """Get current and predicted funding rate"""
    response = session.get_tickers(category="linear", symbol=symbol)
    ticker = response['result']['list'][0]
    
    return {
        'funding_rate': float(ticker.get('fundingRate', 0)),
        'next_funding_time': int(ticker.get('nextFundingTime', 0)),
        'predicted_rate': float(ticker.get('predictedDeliveryPrice', 0)),
    }

def execute_funding_arbitrage(symbol: str, capital_usdt: float):
    """
    Funding rate arb: 
    - If funding is positive: short perp, buy spot → collect funding
    - If funding is negative: long perp, sell spot → collect funding
    """
    funding = get_funding_rate(symbol)
    annual_rate = funding['funding_rate'] * 3 * 365 * 100
    
    print(f"Funding rate: {funding['funding_rate']*100:.4f}% per 8h ({annual_rate:.1f}% APY)")
    
    # Only execute if annualized rate > 15%
    if abs(annual_rate) > 15:
        if annual_rate > 0:
            print("Opening delta-neutral: SELL perp, BUY spot")
            # Open short on Bybit
            session.place_order(
                category="linear", symbol=symbol,
                side="Sell", orderType="Market",
                qty=str(capital_usdt / get_price(symbol))
            )
            # Buy spot on Binance (separate implementation)
        else:
            print("Opening delta-neutral: BUY perp, SELL spot")

Testing on Bybit Testnet

Always run on testnet for at least 2 weeks before live trading:

# Set testnet credentials
export BYBIT_TEST_KEY=your_testnet_key
export BYBIT_TEST_SECRET=your_testnet_secret

Bybit testnet at testnet.bybit.com gives you:

• Free testnet USDT
• All the same APIs as mainnet
• No real money at risk

The discipline of running on testnet for 2 weeks has saved countless developers from expensive mistakes.