Cryptocurrency Backtesting with CCXT: Complete Tutorial
Category: Algo Trading Date: 2026-03-16Introduction
The cryptocurrency market operates 24/7 with characteristics distinct from traditional equities: extreme volatility, sparse liquidity outside major pairs, variable exchange fees, and rapid market evolution. CCXT (CryptoCurrency eXchange Trading) is the de facto standard library for accessing crypto exchange APIs. This guide covers building a comprehensive crypto backtesting system using CCXT. Crypto backtesting presents unique challenges: handling multiple exchanges with different fee structures, managing extreme volatility, accounting for funding rates (perpetuals), and adapting to rapidly changing market conditions.CCXT Fundamentals and Market Data
``python
import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import asyncio
import aiohttp
from enum import Enum
import warnings
warnings.filterwarnings('ignore')
class ExchangeName(Enum):
"""Supported cryptocurrency exchanges."""
BINANCE = 'binance'
KRAKEN = 'kraken'
BYBIT = 'bybit'
KUCOIN = 'kucoin'
DYDX = 'dydx'
class CryptoMarketData:
"""Fetch and manage cryptocurrency market data."""
def __init__(self, exchange_name: str):
self.exchange_name = exchange_name
self.exchange = self._initialize_exchange(exchange_name)
self.ohlcv_cache: Dict[str, pd.DataFrame] = {}
def _initialize_exchange(self, exchange_name: str):
"""Initialize exchange connection."""
exchange_class = getattr(ccxt, exchange_name)
return exchange_class({
'enableRateLimit': True,
'async': False
})
def get_markets(self) -> Dict:
"""Get all available trading pairs."""
markets = self.exchange.load_markets()
return markets
def fetch_ohlcv(self, symbol: str, timeframe: str = '1h',
limit: int = 1000, start_date: Optional[datetime] = None) -> pd.DataFrame:
"""Fetch OHLCV data for a symbol."""
try:
# Get market data
market = self.exchange.market(symbol)
if not self.exchange.has['fetchOHLCV']:
raise ValueError(f"{self.exchange_name} doesn't support fetchOHLCV")
# Fetch OHLCV
ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
# Convert to DataFrame
df = pd.DataFrame(
ohlcv,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
self.ohlcv_cache[symbol] = df
return df
except Exception as e:
print(f"Error fetching {symbol}: {e}")
return pd.DataFrame()
def fetch_ticker(self, symbol: str) -> Dict:
"""Fetch current ticker data."""
try:
ticker = self.exchange.fetch_ticker(symbol)
return ticker
except Exception as e:
print(f"Error fetching ticker {symbol}: {e}")
return {}
def fetch_order_book(self, symbol: str, limit: int = 20) -> Dict:
"""Fetch current order book."""
try:
orderbook = self.exchange.fetch_order_book(symbol, limit=limit)
return orderbook
except Exception as e:
print(f"Error fetching orderbook {symbol}: {e}")
return {}
def get_trading_fees(self, symbol: str) -> Dict:
"""Get trading fees for symbol."""
try:
fees = self.exchange.fetch_trading_fees(symbol)
return fees
except:
# Return default fees if not available
return {
'trading': {'maker': 0.001, 'taker': 0.001}
}
def fetch_historical_trades(self, symbol: str, limit: int = 1000) -> List[Dict]:
"""Fetch historical trades."""
try:
trades = self.exchange.fetch_trades(symbol, limit=limit)
return trades
except Exception as e:
print(f"Error fetching trades for {symbol}: {e}")
return []
class CryptoBacktester:
"""Comprehensive cryptocurrency backtesting engine."""
def __init__(self, initial_capital: float, exchange_name: str = 'binance'):
self.initial_capital = initial_capital
self.current_cash = initial_capital
self.exchange_name = exchange_name
# Market data
self.market_data = CryptoMarketData(exchange_name)
self.price_data: Dict[str, pd.DataFrame] = {}
# Portfolio management
self.positions: Dict[str, float] = {} # Symbol -> quantity
self.entry_prices: Dict[str, float] = {}
self.trades: List[Dict] = []
# State tracking
self.current_prices: Dict[str, float] = {}
self.current_timestamp = None
self.portfolio_history: List[Dict] = []
# Performance metrics
self.returns_series = []
self.equity_curve = []
def load_data(self, symbols: List[str], timeframe: str = '1h',
limit: int = 1000, start_date: Optional[datetime] = None):
"""Load price data for multiple cryptocurrencies."""
for symbol in symbols:
df = self.market_data.fetch_ohlcv(symbol, timeframe, limit, start_date)
if not df.empty:
self.price_data[symbol] = df
print(f"Loaded {len(df)} candles for {symbol}")
def submit_market_order(self, symbol: str, side: str, amount: float,
timestamp: datetime) -> bool:
"""Submit a market order."""
if symbol not in self.current_prices:
return False
price = self.current_prices[symbol]
fee_rate = 0.001 # Default taker fee
if side == 'buy':
total_cost = amount price (1 + fee_rate)
if total_cost > self.current_cash:
# Adjust amount for available cash
amount = (self.current_cash / price) / (1 + fee_rate)
self.current_cash -= total_cost
self.positions[symbol] = self.positions.get(symbol, 0) + amount
self.entry_prices[symbol] = price
elif side == 'sell':
if symbol not in self.positions or self.positions[symbol] < amount:
return False
proceeds = amount price (1 - fee_rate)
self.current_cash += proceeds
self.positions[symbol] -= amount
if self.positions[symbol] < 1e-10: # Account for floating point
self.positions.pop(symbol, None)
# Record trade
self.trades.append({
'timestamp': timestamp,
'symbol': symbol,
'side': side,
'amount': amount,
'price': price,
'fee_rate': fee_rate
})
return True
def submit_limit_order(self, symbol: str, side: str, amount: float,
price: float, timestamp: datetime) -> bool:
"""Submit a limit order (simplified - executes if price crosses)."""
current_price = self.current_prices.get(symbol)
if current_price is None:
return False
# Execute if conditions are met
if side == 'buy' and price >= current_price:
return self.submit_market_order(symbol, side, amount, timestamp)
elif side == 'sell' and price <= current_price:
return self.submit_market_order(symbol, side, amount, timestamp)
return False
def get_portfolio_value(self) -> float:
"""Calculate total portfolio value."""
position_value = sum(
amount * self.current_prices.get(symbol, 0)
for symbol, amount in self.positions.items()
)
return self.current_cash + position_value
def get_position_stats(self) -> Dict:
"""Get statistics on current positions."""
stats = {
'num_positions': len(self.positions),
'total_position_value': 0.0,
'positions': {}
}
for symbol, amount in self.positions.items():
if amount > 0:
current_price = self.current_prices.get(symbol, 0)
position_value = amount * current_price
pnl = position_value - (amount * self.entry_prices.get(symbol, current_price))
pnl_pct = (pnl / (amount self.entry_prices.get(symbol, 1))) 100
stats['positions'][symbol] = {
'amount': amount,
'entry_price': self.entry_prices.get(symbol),
'current_price': current_price,
'position_value': position_value,
'pnl': pnl,
'pnl_pct': pnl_pct
}
stats['total_position_value'] += position_value
return stats
def backtest(self, strategy_func):
"""Run backtest with strategy function."""
if not self.price_data:
raise ValueError("No price data loaded")
# Get aligned dates across all symbols
all_dates = set()
for symbol, df in self.price_data.items():
all_dates.update(df.index)
all_dates = sorted(list(all_dates))
# Process each timestep
for timestamp in all_dates:
self.current_timestamp = timestamp
# Update current prices
for symbol, df in self.price_data.items():
if timestamp in df.index:
self.current_prices[symbol] = df.loc[timestamp, 'close']
# Execute strategy
strategy_func(self)
# Record portfolio state
portfolio_value = self.get_portfolio_value()
self.equity_curve.append({
'timestamp': timestamp,
'portfolio_value': portfolio_value,
'cash': self.current_cash,
'positions': dict(self.positions)
})
# Calculate return
if len(self.equity_curve) > 1:
prev_value = self.equity_curve[-2]['portfolio_value']
ret = (portfolio_value - prev_value) / prev_value
self.returns_series.append(ret)
def get_performance_metrics(self) -> Dict:
"""Calculate performance metrics."""
if not self.equity_curve:
return {}
values = np.array([e['portfolio_value'] for e in self.equity_curve])
returns = np.array(self.returns_series) if self.returns_series else np.array([])
total_return = (values[-1] - self.initial_capital) / self.initial_capital
annual_return = total_return * 365 / len(returns) if len(returns) > 0 else 0
if len(returns) > 0 and np.std(returns) > 0:
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(365)
else:
sharpe = 0
# Maximum drawdown
peak = np.maximum.accumulate(values)
drawdown = (values - peak) / peak
max_dd = np.min(drawdown)
return {
'total_return_%': total_return * 100,
'annual_return_%': annual_return * 100,
'sharpe_ratio': sharpe,
'max_drawdown_%': max_dd * 100,
'final_value': values[-1],
'total_trades': len(self.trades),
'num_winning_trades': sum(1 for t in self.trades if t['side'] == 'sell'),
'num_losing_trades': sum(1 for t in self.trades if t['side'] == 'buy'),
}
Trading Strategy Examples
def rsi_mean_reversion(backtester: CryptoBacktester):
"""RSI-based mean reversion strategy for BTC/USDT."""
symbol = 'BTC/USDT'
if symbol not in backtester.price_data:
return
df = backtester.price_data[symbol]
# Calculate RSI
prices = df['close'].values
deltas = np.diff(prices)
seed = deltas[:1]
up = seed[seed >= 0].sum() / 14
down = -seed[seed < 0].sum() / 14
rs = up / down
rsi = 100 - (100 / (1 + rs))
current_rsi = rsi if not np.isnan(rsi) else 50
# Strategy logic
if current_rsi < 30: # Oversold
if symbol not in backtester.positions:
backtester.submit_market_order(symbol, 'buy', 0.01, backtester.current_timestamp)
elif current_rsi > 70: # Overbought
if symbol in backtester.positions:
amount = backtester.positions[symbol]
backtester.submit_market_order(symbol, 'sell', amount,
backtester.current_timestamp)
def moving_average_crossover(backtester: CryptoBacktester):
"""Moving average crossover for ETH/USDT."""
symbol = 'ETH/USDT'
if symbol not in backtester.price_data:
return
df = backtester.price_data[symbol]
# Calculate MAs
sma_20 = df['close'].rolling(window=20).mean()
sma_50 = df['close'].rolling(window=50).mean()
if len(sma_20) < 50:
return
current_price = df['close'].iloc[-1]
# Golden cross
if sma_20.iloc[-1] > sma_50.iloc[-1] and sma_20.iloc[-2] <= sma_50.iloc[-2]:
if symbol not in backtester.positions:
backtester.submit_market_order(symbol, 'buy', 0.1,
backtester.current_timestamp)
# Death cross
elif sma_20.iloc[-1] < sma_50.iloc[-1] and sma_20.iloc[-2] >= sma_50.iloc[-2]:
if symbol in backtester.positions:
amount = backtester.positions[symbol]
backtester.submit_market_order(symbol, 'sell', amount,
backtester.current_timestamp)
def volatility_breakout(backtester: CryptoBacktester):
"""Volatility breakout strategy."""
symbol = 'BTC/USDT'
if symbol not in backtester.price_data:
return
df = backtester.price_data[symbol]
# Calculate volatility
returns = df['close'].pct_change()
volatility = returns.rolling(window=20).std().iloc[-1]
# Buy on high volatility
if volatility > volatility.quantile(0.75):
if symbol not in backtester.positions:
backtester.submit_market_order(symbol, 'buy', 0.01,
backtester.current_timestamp)
Example usage
if __name__ == "__main__":
backtester = CryptoBacktester(initial_capital=10000, exchange_name='binance')
# Load data - Note: This will attempt real API calls
# For testing, you might use mock data instead
symbols = ['BTC/USDT', 'ETH/USDT']
# Create sample data for demonstration
dates = pd.date_range('2024-01-01', '2024-12-31', freq='1h')
for symbol in symbols:
prices = 100 + np.cumsum(np.random.randn(len(dates)) * 0.5)
df = pd.DataFrame({
'open': prices (1 + np.random.randn(len(dates)) 0.001),
'high': prices (1 + abs(np.random.randn(len(dates)) 0.005)),
'low': prices (1 - abs(np.random.randn(len(dates)) 0.005)),
'close': prices,
'volume': np.random.randint(1000, 10000, len(dates))
}, index=dates)
backtester.price_data[symbol] = df
# Run backtest
backtester.backtest(moving_average_crossover)
# Print results
metrics = backtester.get_performance_metrics()
print("Crypto Backtest Results:")
for key, value in metrics.items():
if isinstance(value, float):
print(f"{key}: {value:.4f}")
else:
print(f"{key}: {value}")
`
Risk Management in Crypto Trading
`python
class CryptoRiskManager:
"""Risk management specific to cryptocurrency trading."""
@staticmethod
def calculate_position_size(capital: float, risk_per_trade: float,
entry_price: float, stop_loss: float) -> float:
"""Calculate position size based on risk parameters."""
risk_amount = capital * risk_per_trade
risk_per_unit = abs(entry_price - stop_loss)
if risk_per_unit == 0:
return 0
position_size = risk_amount / risk_per_unit
return position_size
@staticmethod
def calculate_funding_costs(position_size: float, funding_rate: float,
holding_period_days: int) -> float:
"""Calculate funding costs for perpetual positions."""
daily_cost = position_size * funding_rate
total_cost = daily_cost * holding_period_days
return total_cost
@staticmethod
def detect_liquidation_risk(position_value: float, entry_price: float,
current_price: float, leverage: float,
liquidation_price: float) -> float:
"""Calculate distance to liquidation."""
distance = abs(current_price - liquidation_price)
distance_pct = (distance / current_price) * 100
return distance_pct
``