---
title: "Build a Crypto Trading Bot in Python: Complete Guide"
description: "Create a profitable crypto trading bot in Python using CCXT, backtesting, risk management, and automated order execution. Full tutorial with code."
date: "2026-07-03"
category: "python"
tags: ["python", "crypto", "trading", "ccxt", "bot", "automation", "fintech"]
---
Automating crypto trading removes emotion from decisions and executes strategies 24/7. This guide builds a complete trading bot with exchange integration, backtesting, and risk management.
A fully functional crypto trading bot that connects to major exchanges via CCXT, implements multiple trading strategies, includes backtesting capabilities, and enforces strict risk management rules.
Python dominates algorithmic trading for clear reasons:
trading-bot/
├── bot/
│ ├── __init__.py
│ ├── exchange.py
│ ├── strategies.py
│ ├── risk_manager.py
│ ├── backtester.py
│ └── main.py
├── config.yaml
├── requirements.txt
└── run.py
pip install ccxt pandas numpy pyyaml ta
# bot/exchange.py
import ccxt
import asyncio
from typing import Optional
from dataclasses import dataclass
@dataclass
class OrderResult:
success: bool
order_id: Optional[str] = None
price: Optional[float] = None
amount: Optional[float] = None
error: Optional[str] = None
class Exchange:
def __init__(self, config: dict):
self.exchange = getattr(ccxt, config["exchange"])({
"apiKey": config["api_key"],
"secret": config["secret"],
"password": config.get("passphrase", ""),
"sandbox": config.get("sandbox", True),
"options": {"defaultType": "spot"}
})
async def fetch_ticker(self, symbol: str) -> dict:
return await self.exchange.fetch_ticker(symbol)
async def fetch_ohlcv(self, symbol: str, timeframe: str = "1h", limit: int = 100):
return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
async def create_order(self, symbol: str, side: str, amount: float,
price: Optional[float] = None) -> OrderResult:
try:
order_type = "limit" if price else "market"
order = await self.exchange.create_order(
symbol, order_type, side, amount, price
)
return OrderResult(
success=True,
order_id=order["id"],
price=order.get("average"),
amount=order.get("amount")
)
except Exception as e:
return OrderResult(success=False, error=str(e))
async def fetch_balance(self) -> dict:
return await self.exchange.fetch_balance()
async def cancel_order(self, order_id: str, symbol: str) -> bool:
try:
await self.exchange.cancel_order(order_id, symbol)
return True
except Exception:
return False
# bot/strategies.py
import pandas as pd
import ta
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class Signal:
action: str # "buy", "sell", "hold"
confidence: float
reason: str
class Strategy(ABC):
@abstractmethod
def analyze(self, df: pd.DataFrame) -> Signal:
pass
class RSIMeanReversion(Strategy):
def __init__(self, oversold: int = 30, overbought: int = 70):
self.oversold = oversold
self.overbought = overbought
def analyze(self, df: pd.DataFrame) -> Signal:
rsi = ta.momentum.RSIIndicator(df["close"], window=14).rsi()
current_rsi = rsi.iloc[-1]
if current_rsi < self.oversold:
return Signal("buy", 0.8, f"RSI oversold at {current_rsi:.1f}")
elif current_rsi > self.overbought:
return Signal("sell", 0.8, f"RSI overbought at {current_rsi:.1f}")
return Signal("hold", 0.5, f"RSI neutral at {current_rsi:.1f}")
class MACDCrossover(Strategy):
def analyze(self, df: pd.DataFrame) -> Signal:
macd = ta.trend.MACD(df["close"])
macd_line = macd.macd()
signal_line = macd.macd_signal()
if macd_line.iloc[-1] > signal_line.iloc[-1] and macd_line.iloc[-2] <= signal_line.iloc[-2]:
return Signal("buy", 0.75, "MACD bullish crossover")
elif macd_line.iloc[-1] < signal_line.iloc[-1] and macd_line.iloc[-2] >= signal_line.iloc[-2]:
return Signal("sell", 0.75, "MACD bearish crossover")
return Signal("hold", 0.5, "No crossover")
class BollingerBandBreakout(Strategy):
def analyze(self, df: pd.DataFrame) -> Signal:
bb = ta.volatility.BollingerBands(df["close"], window=20, window_dev=2)
upper = bb.bollinger_hband().iloc[-1]
lower = bb.bollinger_lband().iloc[-1]
current = df["close"].iloc[-1]
if current < lower:
return Signal("buy", 0.7, f"Price below lower band ({lower:.2f})")
elif current > upper:
return Signal("sell", 0.7, f"Price above upper band ({upper:.2f})")
return Signal("hold", 0.5, "Within bands")
# bot/risk_manager.py
from dataclasses import dataclass
from typing import Optional
@dataclass
class RiskConfig:
max_position_size: float = 0.02 # 2% of portfolio per trade
max_daily_loss: float = 0.05 # 5% daily loss limit
max_open_orders: int = 5
stop_loss_pct: float = 0.02 # 2% stop loss
take_profit_pct: float = 0.04 # 4% take profit
class RiskManager:
def __init__(self, config: RiskConfig, initial_balance: float):
self.config = config
self.initial_balance = initial_balance
self.daily_pnl = 0.0
self.open_orders = 0
def calculate_position_size(self, price: float, balance: float) -> float:
max_amount = (balance * self.config.max_position_size) / price
return max_amount
def calculate_stop_loss(self, entry_price: float, side: str) -> float:
if side == "buy":
return entry_price * (1 - self.config.stop_loss_pct)
return entry_price * (1 + self.config.stop_loss_pct)
def calculate_take_profit(self, entry_price: float, side: str) -> float:
if side == "buy":
return entry_price * (1 + self.config.take_profit_pct)
return entry_price * (1 - self.config.take_profit_pct)
def can_trade(self, balance: float) -> tuple[bool, str]:
if self.open_orders >= self.config.max_open_orders:
return False, "Max open orders reached"
if self.daily_pnl < -balance * self.config.max_daily_loss:
return False, "Daily loss limit reached"
return True, "OK"
def record_pnl(self, pnl: float):
self.daily_pnl += pnl
if pnl < 0:
self.open_orders -= 1
else:
self.open_orders -= 1
# bot/backtester.py
import pandas as pd
from typing import Type
from bot.strategies import Strategy, Signal
class Backtester:
def __init__(self, initial_balance: float = 10000, fee_pct: float = 0.001):
self.initial_balance = initial_balance
self.fee_pct = fee_pct
def run(self, strategy: Strategy, df: pd.DataFrame) -> dict:
balance = self.initial_balance
position = 0.0
entry_price = 0.0
trades = []
for i in range(20, len(df)):
window = df.iloc[:i + 1]
signal = strategy.analyze(window)
current_price = df["close"].iloc[i]
if signal.action == "buy" and position == 0:
cost = balance * 0.95
position = (cost * (1 - self.fee_pct)) / current_price
entry_price = current_price
balance -= cost
trades.append({
"type": "buy", "price": current_price,
"amount": position, "balance": balance
})
elif signal.action == "sell" and position > 0:
proceeds = position * current_price * (1 - self.fee_pct)
pnl = proceeds - (position * entry_price)
balance += proceeds
trades.append({
"type": "sell", "price": current_price,
"amount": position, "pnl": pnl, "balance": balance
})
position = 0
final_value = balance + (position * df["close"].iloc[-1] if position else 0)
total_return = (final_value - self.initial_balance) / self.initial_balance
wins = [t for t in trades if t.get("pnl", 0) > 0]
losses = [t for t in trades if t.get("pnl", 0) < 0]
return {
"initial_balance": self.initial_balance,
"final_value": final_value,
"total_return": f"{total_return:.2%}",
"total_trades": len(trades),
"win_rate": f"{len(wins) / max(len(wins) + len(losses), 1):.2%}",
"trades": trades
}
# bot/main.py
import asyncio
import yaml
from bot.exchange import Exchange
from bot.strategies import RSIMeanReversion
from bot.risk_manager import RiskManager, RiskConfig
class TradingBot:
def __init__(self, config_path: str):
with open(config_path) as f:
self.config = yaml.safe_load(f)
self.exchange = Exchange(self.config["exchange"])
self.strategy = RSIMeanReversion()
self.risk_manager = RiskManager(
RiskConfig(**self.config.get("risk", {})),
self.config["initial_balance"]
)
async def run(self, symbol: str, interval: int = 3600):
print(f"Starting bot for {symbol}")
while True:
try:
ohlcv = await self.exchange.fetch_ohlcv(symbol, "1h", 100)
import pandas as pd
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
signal = self.strategy.analyze(df)
balance = await self.exchange.fetch_balance()
total = balance["total"].get("USDT", 0)
can_trade, reason = self.risk_manager.can_trade(total)
if not can_trade:
print(f"Cannot trade: {reason}")
await asyncio.sleep(interval)
continue
if signal.action == "buy" and total > 10:
amount = self.risk_manager.calculate_position_size(
df["close"].iloc[-1], total
)
result = await self.exchange.create_order(symbol, "buy", amount)
if result.success:
self.risk_manager.open_orders += 1
print(f"BUY: {amount} at {result.price}")
elif signal.action == "sell":
# Check existing positions
pass
print(f"Signal: {signal.action} ({signal.confidence:.0%}) - {signal.reason}")
await asyncio.sleep(interval)
except Exception as e:
print(f"Error: {e}")
await asyncio.sleep(60)
if __name__ == "__main__":
bot = TradingBot("config.yaml")
asyncio.run(bot.run("BTC/USDT"))
Ready to use these tools? Browse our collection of tested, production-ready Python scripts:
🔗 Browse Products: [Anna's Digital Products](https://petroleum-board-hawaii-lol.trycloudflare.com)
All products include:
---
Don't want to build it yourself? We have production-ready versions of these tools at [https://petroleum-board-hawaii-lol.trycloudflare.com](https://petroleum-board-hawaii-lol.trycloudflare.com).
What you get:
[Browse the collection →](https://petroleum-board-hawaii-lol.trycloudflare.com)
Browse 120+ Python tools with crypto payments and instant delivery.
Browse Products →