What You Are Building
A Python trading bot that uses Bybit’s official MCP server to pull live market data, calculate momentum signals, and execute futures trades on Bybit testnet. Bybit launched their MCP integration on April 22, 2026, and this tutorial is one of the first hands-on builds using it. You will use Claude Code to write the bot, connect it through MCP, and test everything on paper trading before risking real funds.
Why Bybit MCP
Bybit is the second-largest crypto exchange by trading volume. Their MCP release means you can connect AI agents like Claude Code directly to Bybit’s infrastructure through a standardized protocol instead of writing custom API wrappers from scratch.
If you have followed our Binance API setup guide, you know the process: install a Python library, manage API keys, handle rate limits, parse responses. MCP simplifies some of this by giving the AI agent direct access to structured tools for market data, order placement, and position management.
Here is what Bybit’s MCP exposes:
| Module | What You Get |
|---|---|
| Market Data | Real-time tickers, candlestick data, order book snapshots, fee schedules |
| Trading | Spot orders, perpetual futures, conditional orders, stop-loss/take-profit |
| WebSocket Streams | Live price updates, trade execution feeds, position change notifications |
| Account | Balance queries, position summaries, order history |
Prerequisites
- Python 3.10+
- A Bybit account with API keys (testnet keys work for this tutorial)
- Claude Code installed and working
- Node.js 18+ (for the MCP server)
- Basic Python and REST API familiarity
Step 1: Get Bybit API Keys (Testnet)
Start with testnet keys so you do not risk real money while learning:
- Go to testnet.bybit.com
- Create an account or log in
- Navigate to API Management and create a new API key
- Enable Contract Trading and Spot Trading permissions
- Save the API key and secret somewhere safe
Testnet gives you fake funds to trade with. The market data is real, but the orders execute against a simulated order book.
Step 2: Install the Bybit MCP Server
Bybit’s official MCP server is available as an npm package:
npm install -g @bybit/mcp-server
Or clone the repo for more control:
git clone https://github.com/bybit-exchange/mcp-server.git ~/bybit-mcp
cd ~/bybit-mcp
npm install
Step 3: Configure Claude Code for Bybit MCP
Add the Bybit MCP server to your Claude Code configuration:
nano ~/.claude/.mcp.json
{
"mcpServers": {
"bybit": {
"command": "node",
"args": ["/path/to/bybit-mcp/src/index.js"],
"env": {
"BYBIT_API_KEY": "your-testnet-api-key",
"BYBIT_API_SECRET": "your-testnet-secret",
"BYBIT_TESTNET": "true"
}
}
}
}
Set BYBIT_TESTNET to true for paper trading. Change it to false only when you are ready to trade with real funds (and understand the risks).
If you already have other MCP servers configured (like TradingView MCP), add the Bybit entry alongside them in the mcpServers object.
Step 4: Verify the Connection
Restart Claude Code and test:
Check my Bybit connection. What's my testnet USDT balance?
Claude should return your testnet balance through the MCP tools. If it fails, double-check your API keys and make sure the MCP server path is correct.
The Strategy: Simple Momentum
We will build a straightforward momentum strategy for BTCUSDT perpetual futures:
- Fetch the last 50 candles on the 15-minute timeframe
- Calculate a 14-period RSI and a 20-period EMA
- Go long when RSI crosses above 50 and price is above the EMA
- Go short when RSI crosses below 50 and price is below the EMA
- Use a 1% stop-loss and 2% take-profit on every position
- Maximum one open position at a time
This is not a profitable strategy out of the box. It is a teaching example that shows you how to wire up market data, signal generation, and order execution through MCP. Tuning the parameters for profitability is a separate project.
Prompt 1: Core Bot Structure
Here is the prompt I gave Claude Code:
Build a Python crypto trading bot for Bybit BTCUSDT perpetual futures with these specs:
- Use the pybit library to connect to Bybit testnet
- Fetch 50 candles of 15-minute OHLCV data on startup
- Calculate 14-period RSI and 20-period EMA using pandas
- Trading logic: long when RSI crosses above 50 and close > EMA, short when RSI crosses below 50 and close < EMA
- Position sizing: 0.01 BTC per trade
- Set stop-loss at 1% and take-profit at 2% on every order
- Only one open position at a time — check before placing new orders
- Run in a loop, checking every 60 seconds
- Log every signal, order, and fill to both console and a CSV file
- Load API keys from environment variables
Claude Code generated a working first draft. The main issues were:
- It used the old Bybit v3 API endpoints instead of v5
- The RSI calculation had an off-by-one error in the rolling window
- No rate limiting between API calls
Prompt 2: Fix API Version and Add Rate Limiting
Fix these issues in the Bybit bot:
- Switch all API calls to Bybit v5 unified endpoints
- Fix the RSI calculation — use the Wilder smoothing method, not simple rolling
- Add a 0.5 second delay between API calls to stay under rate limits
- Add proper error handling for InsufficientBalance, InvalidOrder, and network timeouts
- When checking for open positions, use the v5 position/list endpoint
This cleaned up most problems. One more round for risk management.
Prompt 3: Risk Controls
Add risk management to the Bybit bot:
- Maximum 3 trades per hour — if exceeded, wait until the next hour
- Daily loss limit: if net realized PnL drops below -$50 for the day, stop trading until next UTC midnight
- Track win/loss ratio and log it every 10 trades
- Add a kill switch: if the file “STOP” exists in the current directory, exit gracefully
The Full Code
Here is the complete bot after three rounds of iteration with manual corrections noted in comments:
import os
import time
import logging
import csv
from datetime import datetime, timezone
from pathlib import Path
from dataclasses import dataclass, field
import pandas as pd
from pybit.unified_trading import HTTP
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler("bybit_bot.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# --- Config ---
API_KEY = os.environ.get("BYBIT_API_KEY", "")
API_SECRET = os.environ.get("BYBIT_API_SECRET", "")
TESTNET = os.environ.get("BYBIT_TESTNET", "true").lower() == "true"
SYMBOL = "BTCUSDT"
INTERVAL = "15"
QTY = "0.01"
SL_PCT = 0.01
TP_PCT = 0.02
RSI_PERIOD = 14
EMA_PERIOD = 20
MAX_TRADES_PER_HOUR = 3
DAILY_LOSS_LIMIT = -50.0
CHECK_INTERVAL = 60
@dataclass
class BotState:
position: str = "none" # "long", "short", "none"
entry_price: float = 0.0
trades_this_hour: int = 0
hour_start: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
daily_pnl: float = 0.0
day_start: str = ""
trade_count: int = 0
wins: int = 0
losses: int = 0
def calc_rsi(closes: pd.Series, period: int = 14) -> pd.Series:
"""Wilder smoothing RSI."""
delta = closes.diff()
gain = delta.where(delta > 0, 0.0)
loss = -delta.where(delta < 0, 0.0)
avg_gain = gain.ewm(alpha=1/period, min_periods=period).mean()
avg_loss = loss.ewm(alpha=1/period, min_periods=period).mean()
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
def calc_ema(closes: pd.Series, period: int = 20) -> pd.Series:
return closes.ewm(span=period, adjust=False).mean()
def fetch_candles(session, symbol: str, interval: str, limit: int = 50):
resp = session.get_kline(
category="linear",
symbol=symbol,
interval=interval,
limit=limit,
)
rows = resp["result"]["list"]
df = pd.DataFrame(rows, columns=["timestamp", "open", "high", "low", "close", "volume", "turnover"])
for col in ["open", "high", "low", "close", "volume"]:
df[col] = df[col].astype(float)
df["timestamp"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms")
df = df.sort_values("timestamp").reset_index(drop=True)
return df
def get_position(session, symbol: str) -> dict:
resp = session.get_positions(category="linear", symbol=symbol)
positions = resp["result"]["list"]
for pos in positions:
size = float(pos.get("size", 0))
if size > 0:
return {
"side": pos["side"],
"size": size,
"entry_price": float(pos["avgPrice"]),
}
return {"side": "None", "size": 0, "entry_price": 0}
def place_order(session, symbol: str, side: str, qty: str,
sl_price: float, tp_price: float):
try:
resp = session.place_order(
category="linear",
symbol=symbol,
side=side,
orderType="Market",
qty=qty,
stopLoss=str(round(sl_price, 2)),
takeProfit=str(round(tp_price, 2)),
timeInForce="GTC",
)
logger.info(f"Order placed: {side} {qty} {symbol} | SL: {sl_price} | TP: {tp_price}")
return resp
except Exception as e:
logger.error(f"Order failed: {e}")
return None
def close_position(session, symbol: str, side: str, qty: str):
close_side = "Sell" if side == "Buy" else "Buy"
try:
resp = session.place_order(
category="linear",
symbol=symbol,
side=close_side,
orderType="Market",
qty=qty,
reduceOnly=True,
timeInForce="GTC",
)
logger.info(f"Position closed: {close_side} {qty} {symbol}")
return resp
except Exception as e:
logger.error(f"Close failed: {e}")
return None
def log_trade(action, price, rsi, ema, pnl=0.0):
with open("trades.csv", "a", newline="") as f:
writer = csv.writer(f)
writer.writerow([
datetime.now(timezone.utc).isoformat(),
action, price, round(rsi, 2), round(ema, 2), round(pnl, 4)
])
def run():
session = HTTP(
testnet=TESTNET,
api_key=API_KEY,
api_secret=API_SECRET,
)
state = BotState()
state.day_start = datetime.now(timezone.utc).strftime("%Y-%m-%d")
# Write CSV header if file doesn't exist
if not Path("trades.csv").exists():
with open("trades.csv", "w", newline="") as f:
writer = csv.writer(f)
writer.writerow(["timestamp", "action", "price", "rsi", "ema", "pnl"])
logger.info(f"Bot started | Symbol: {SYMBOL} | Testnet: {TESTNET}")
logger.info(f"Strategy: RSI({RSI_PERIOD}) + EMA({EMA_PERIOD}) momentum")
while True:
# Kill switch
if Path("STOP").exists():
logger.info("Kill switch activated. Exiting.")
break
# Reset daily PnL at midnight UTC
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
if today != state.day_start:
logger.info(f"New day. Previous day PnL: ${state.daily_pnl:.2f}")
state.daily_pnl = 0.0
state.day_start = today
# Check daily loss limit
if state.daily_pnl <= DAILY_LOSS_LIMIT:
logger.warning(f"Daily loss limit hit (${state.daily_pnl:.2f}). Paused until midnight UTC.")
time.sleep(CHECK_INTERVAL)
continue
# Reset hourly trade counter
now = datetime.now(timezone.utc)
if (now - state.hour_start).total_seconds() >= 3600:
state.trades_this_hour = 0
state.hour_start = now
if state.trades_this_hour >= MAX_TRADES_PER_HOUR:
logger.info(f"Max trades per hour reached ({MAX_TRADES_PER_HOUR}). Waiting.")
time.sleep(CHECK_INTERVAL)
continue
try:
# Fetch candles and calculate indicators
df = fetch_candles(session, SYMBOL, INTERVAL)
time.sleep(0.5) # Rate limiting
df["rsi"] = calc_rsi(df["close"], RSI_PERIOD)
df["ema"] = calc_ema(df["close"], EMA_PERIOD)
current = df.iloc[-1]
previous = df.iloc[-2]
price = current["close"]
rsi = current["rsi"]
ema = current["ema"]
prev_rsi = previous["rsi"]
logger.info(f"Price: {price} | RSI: {rsi:.1f} | EMA: {ema:.1f}")
# Check current position
pos = get_position(session, SYMBOL)
time.sleep(0.5)
has_position = pos["size"] > 0
# Signal detection
long_signal = prev_rsi <= 50 and rsi > 50 and price > ema
short_signal = prev_rsi >= 50 and rsi < 50 and price < ema
if not has_position:
if long_signal:
sl = price * (1 - SL_PCT)
tp = price * (1 + TP_PCT)
result = place_order(session, SYMBOL, "Buy", QTY, sl, tp)
if result:
state.position = "long"
state.entry_price = price
state.trades_this_hour += 1
log_trade("BUY", price, rsi, ema)
elif short_signal:
sl = price * (1 + SL_PCT)
tp = price * (1 - TP_PCT)
result = place_order(session, SYMBOL, "Sell", QTY, sl, tp)
if result:
state.position = "short"
state.entry_price = price
state.trades_this_hour += 1
log_trade("SELL", price, rsi, ema)
# Log win/loss stats every 10 trades
if state.trade_count > 0 and state.trade_count % 10 == 0:
wr = state.wins / state.trade_count * 100 if state.trade_count > 0 else 0
logger.info(f"Stats: {state.trade_count} trades | {wr:.1f}% win rate")
except Exception as e:
logger.error(f"Error in main loop: {e}")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
run()
What Claude Code Got Right
- The pybit v5 API usage was correct after the second prompt. The
category="linear"parameter for futures, theget_klineandget_positionscalls, and the order parameters all matched Bybit’s current API docs. - Error handling was solid. It wrapped each API call individually instead of one big try/except, which makes debugging easier.
- The kill switch file approach is simple and effective. You can stop the bot from another terminal window without sending signals.
What Needed Manual Fixes
- RSI calculation: Claude Code initially used
rolling().mean()instead of exponential weighted mean for Wilder smoothing. This is a common mistake across all AI coding tools. I corrected it in Prompt 2. - Order quantity format: Bybit expects quantity as a string, not a float. Claude Code passed a float on the first attempt, which caused a type error.
- Rate limiting: The first version had no delays between API calls. On testnet this is fine, but on mainnet you will hit rate limits within minutes during volatile markets.
Running the Bot
export BYBIT_API_KEY="your-testnet-key"
export BYBIT_API_SECRET="your-testnet-secret"
export BYBIT_TESTNET="true"
python bybit_bot.py
Monitor the output in your terminal and check trades.csv for a log of all actions. To stop the bot cleanly, create a file named STOP in the same directory:
touch STOP
Moving to Live Trading
Before switching to real funds:
- Run on testnet for at least 2 weeks to understand the strategy’s behavior
- Review the
trades.csvlog for patterns (is it overtrading? Are stops getting hit immediately?) - Start with the minimum position size on mainnet (0.001 BTC)
- Set
BYBIT_TESTNETtofalseand use mainnet API keys - Monitor closely for the first few days
Warning: This bot trades futures with leverage. Losses can exceed your initial margin. Do not use money you cannot afford to lose. The strategy in this tutorial is for educational purposes and has not been backtested for profitability.
What to Build Next
- Add a WebSocket stream for real-time data instead of polling every 60 seconds
- Combine with TradingView MCP for visual chart analysis alongside automated execution
- Implement a proper backtesting framework using historical Bybit data before going live
- Add Telegram or Discord notifications for trade alerts
- Try the DCA bot approach for a less aggressive strategy on Bybit spot markets