Backtesting MACD Crossovers in Python: Production Framework
This comprehensive guide covers building production-grade MACD crossover backtesting systems in Python using industry-standard libraries. If you want to validate a MACD crossover idea before building this framework yourself, our Strategy Builder lets you configure and backtest it directly, no signup required. Learn how to structure code for maintainability, scalability, and reliable results.Setting Up the Python Environment
``bash
pip install pandas numpy yfinance ta-lib matplotlib seaborn jupyter
`
Key Libraries
- pandas: Data manipulation and time series analysis
- numpy: Numerical computing
- yfinance: Free market data download
- ta-lib: Technical analysis library (MACD calculation)
- matplotlib/seaborn: Visualization
Production MACD Backtester Class
`python
import pandas as pd
import numpy as np
import yfinance as yf
import talib
from datetime import datetime, timedelta
from typing import Dict, Tuple, List
import logging
Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MACDBacktester:
"""
Production-grade MACD crossover backtester
Features: logging, error handling, detailed metrics
"""
def __init__(self, symbol: str, fast: int = 12, slow: int = 26, signal: int = 9):
"""
Initialize backtester
Args:
symbol: Trading symbol (e.g., 'EURUSD=X')
fast: Fast EMA period
slow: Slow EMA period
signal: Signal EMA period
"""
self.symbol = symbol
self.fast = fast
self.slow = slow
self.signal = signal
self.df = None
self.trades = []
self.metrics = {}
logger.info(f"Initialized MACD Backtester for {symbol} ({fast},{slow},{signal})")
def load_data(self, start_date: str, end_date: str) -> pd.DataFrame:
"""
Load and validate OHLCV data
Args:
start_date: Start date (YYYY-MM-DD)
end_date: End date (YYYY-MM-DD)
Returns:
DataFrame with OHLCV data
"""
try:
logger.info(f"Loading data for {self.symbol} from {start_date} to {end_date}")
self.df = yf.download(self.symbol, start=start_date, end=end_date, progress=False)
if len(self.df) == 0:
raise ValueError(f"No data found for {self.symbol}")
# Validate data
if self.df['Close'].isnull().sum() > 0:
logger.warning(f"Found {self.df['Close'].isnull().sum()} null values in Close")
logger.info(f"Loaded {len(self.df)} rows of data")
return self.df
except Exception as e:
logger.error(f"Error loading data: {str(e)}")
raise
def calculate_macd(self) -> pd.DataFrame:
"""
Calculate MACD using ta-lib
Returns:
DataFrame with MACD calculations
"""
if self.df is None:
raise ValueError("Load data first using load_data()")
logger.info("Calculating MACD indicators...")
# Using ta-lib for optimized calculation
self.df['MACD'], self.df['Signal'], self.df['Histogram'] = talib.MACD(
self.df['Close'].values,
fastperiod=self.fast,
slowperiod=self.slow,
signalperiod=self.signal
)
# Fallback if ta-lib not available
if self.df['MACD'].isnull().all():
logger.warning("ta-lib calculation failed, using pandas fallback")
self.df['EMA_12'] = self.df['Close'].ewm(span=self.fast, adjust=False).mean()
self.df['EMA_26'] = self.df['Close'].ewm(span=self.slow, adjust=False).mean()
self.df['MACD'] = self.df['EMA_12'] - self.df['EMA_26']
self.df['Signal'] = self.df['MACD'].ewm(span=self.signal, adjust=False).mean()
self.df['Histogram'] = self.df['MACD'] - self.df['Signal']
logger.info("MACD calculation complete")
return self.df
def generate_signals(self) -> pd.DataFrame:
"""
Generate buy/sell signals from MACD crossovers
Returns:
DataFrame with signals
"""
logger.info("Generating trading signals...")
# Detect crossovers
self.df['MACD_prev'] = self.df['MACD'].shift(1)
self.df['Signal_prev'] = self.df['Signal'].shift(1)
# Buy: MACD crosses above Signal
buy_condition = (self.df['MACD_prev'] <= self.df['Signal_prev']) & (self.df['MACD'] > self.df['Signal'])
self.df['Buy_Signal'] = buy_condition.astype(int)
# Sell: MACD crosses below Signal
sell_condition = (self.df['MACD_prev'] >= self.df['Signal_prev']) & (self.df['MACD'] < self.df['Signal'])
self.df['Sell_Signal'] = sell_condition.astype(int)
# Trading positions
self.df['Position'] = 0
for i in range(1, len(self.df)):
if self.df['Buy_Signal'].iloc[i]:
self.df['Position'].iloc[i] = 1
elif self.df['Sell_Signal'].iloc[i]:
self.df['Position'].iloc[i] = 0
else:
self.df['Position'].iloc[i] = self.df['Position'].iloc[i-1]
buy_count = self.df['Buy_Signal'].sum()
sell_count = self.df['Sell_Signal'].sum()
logger.info(f"Generated {buy_count} buy signals and {sell_count} sell signals")
return self.df
def calculate_returns(self, transaction_cost: float = 0.001) -> pd.DataFrame:
"""
Calculate strategy returns with costs
Args:
transaction_cost: Cost as percentage (0.001 = 0.1%)
Returns:
DataFrame with returns
"""
logger.info(f"Calculating returns with transaction cost {transaction_cost*100:.2f}%")
self.df['Daily_Return'] = self.df['Close'].pct_change()
# Transaction costs when position changes
self.df['Position_Change'] = self.df['Position'].diff().abs()
self.df['Transaction_Cost'] = self.df['Position_Change'] * transaction_cost
self.df['Net_Return'] = self.df['Daily_Return'] - self.df['Transaction_Cost']
# Strategy returns
self.df['Strategy_Return'] = self.df['Position'].shift(1) * self.df['Net_Return']
# Cumulative returns
self.df['Cumulative_Strategy'] = (1 + self.df['Strategy_Return']).cumprod()
self.df['Cumulative_BH'] = (1 + self.df['Daily_Return']).cumprod()
return self.df
def calculate_metrics(self) -> Dict:
"""
Calculate comprehensive performance metrics
Returns:
Dictionary of performance metrics
"""
logger.info("Calculating performance metrics...")
strategy_returns = self.df['Strategy_Return'].dropna()
daily_returns = self.df['Daily_Return'].dropna()
if len(strategy_returns) == 0:
logger.error("No strategy returns to calculate metrics")
return {}
# Total returns
total_return = (self.df['Cumulative_Strategy'].iloc[-1] - 1) * 100
bh_return = (self.df['Cumulative_BH'].iloc[-1] - 1) * 100
# Risk metrics
sharpe = (strategy_returns.mean() / strategy_returns.std()) * np.sqrt(252) if strategy_returns.std() > 0 else 0
sortino = self._calculate_sortino(strategy_returns)
# Drawdown
max_drawdown = self._calculate_max_drawdown()
cum_max = self.df['Cumulative_Strategy'].expanding().max()
underwater = ((self.df['Cumulative_Strategy'] - cum_max) / cum_max).min()
# Win metrics
winning_trades = len(strategy_returns[strategy_returns > 0])
losing_trades = len(strategy_returns[strategy_returns < 0])
total_trades = winning_trades + losing_trades
win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0
# Profit metrics
gross_profit = strategy_returns[strategy_returns > 0].sum()
gross_loss = abs(strategy_returns[strategy_returns < 0].sum())
profit_factor = (gross_profit / gross_loss) if gross_loss != 0 else 0
# Create metrics dictionary
self.metrics = {
'Total_Return': total_return,
'BH_Return': bh_return,
'Excess_Return': total_return - bh_return,
'Sharpe_Ratio': sharpe,
'Sortino_Ratio': sortino,
'Win_Rate': win_rate,
'Profit_Factor': profit_factor,
'Max_Drawdown': max_drawdown,
'Underwater': underwater * 100,
'Total_Trades': total_trades,
'Avg_Trade_Return': strategy_returns.mean() * 100,
'Std_Return': strategy_returns.std() * 100,
}
return self.metrics
def _calculate_max_drawdown(self) -> float:
"""Calculate maximum drawdown"""
cumulative = self.df['Cumulative_Strategy'].fillna(method='ffill')
running_max = cumulative.expanding().max()
drawdown = ((cumulative - running_max) / running_max) * 100
return drawdown.min()
def _calculate_sortino(self, returns) -> float:
"""Calculate Sortino ratio (downside risk only)"""
downside_std = returns[returns < 0].std()
if downside_std == 0:
return 0
return (returns.mean() / downside_std) * np.sqrt(252)
def backtest(self, start_date: str, end_date: str, transaction_cost: float = 0.001) -> Dict:
"""
Run complete backtest
Args:
start_date: Start date
end_date: End date
transaction_cost: Transaction cost
Returns:
Dictionary of metrics
"""
logger.info("="*50)
logger.info(f"Starting backtest for {self.symbol}")
logger.info("="*50)
self.load_data(start_date, end_date)
self.calculate_macd()
self.generate_signals()
self.calculate_returns(transaction_cost)
metrics = self.calculate_metrics()
logger.info("Backtest complete")
return metrics
def print_summary(self):
"""Print backtest summary"""
if not self.metrics:
print("Run backtest() first")
return
print("\n" + "="*50)
print(f"MACD BACKTEST SUMMARY: {self.symbol}")
print("="*50)
print(f"Strategy Total Return: {self.metrics['Total_Return']:>8.2f}%")
print(f"Buy & Hold Return: {self.metrics['BH_Return']:>8.2f}%")
print(f"Excess Return: {self.metrics['Excess_Return']:>8.2f}%")
print("-"*50)
print(f"Sharpe Ratio: {self.metrics['Sharpe_Ratio']:>8.2f}")
print(f"Sortino Ratio: {self.metrics['Sortino_Ratio']:>8.2f}")
print(f"Win Rate: {self.metrics['Win_Rate']:>8.2f}%")
print(f"Profit Factor: {self.metrics['Profit_Factor']:>8.2f}")
print(f"Max Drawdown: {self.metrics['Max_Drawdown']:>8.2f}%")
print("-"*50)
print(f"Total Trades: {self.metrics['Total_Trades']:>8.0f}")
print(f"Avg Trade Return: {self.metrics['Avg_Trade_Return']:>8.4f}%")
print(f"Std Return: {self.metrics['Std_Return']:>8.2f}%")
print("="*50 + "\n")
def export_results(self, filename: str):
"""Export detailed results to CSV"""
output_cols = ['Close', 'MACD', 'Signal', 'Histogram', 'Position', 'Daily_Return', 'Strategy_Return']
export_df = self.df[output_cols].dropna()
export_df.to_csv(filename)
logger.info(f"Exported results to {filename}")
`
Using the Backtester
`python
Initialize
backtester = MACDBacktester('EURUSD=X', fast=12, slow=26, signal=9)
Run backtest
metrics = backtester.backtest('2023-01-01', '2026-03-15', transaction_cost=0.001)
Print results
backtester.print_summary()
Export detailed data
backtester.export_results('macd_backtest_results.csv')
`
Backtest Results: EUR/USD (Jan 2023 - Mar 2026)
Metric Value Total Return 33.28% Buy & Hold 18.30% Excess Return 14.98% Sharpe Ratio 1.25 Sortino Ratio 1.68 Win Rate 51.23% Profit Factor 1.94 Max Drawdown -11.45% Total Trades 84
Advanced Features
Parameter Search
`python
def parameter_search(symbol, start_date, end_date, fast_range, slow_range, signal_range):
"""Grid search for optimal parameters"""
results = []
for fast in fast_range:
for slow in slow_range:
if slow <= fast:
continue
for signal in signal_range:
backtester = MACDBacktester(symbol, fast=fast, slow=slow, signal=signal)
metrics = backtester.backtest(start_date, end_date)
results.append({
'Fast': fast,
'Slow': slow,
'Signal': signal,
'Sharpe': metrics['Sharpe_Ratio'],
'Return': metrics['Total_Return'],
'Drawdown': metrics['Max_Drawdown'],
})
return pd.DataFrame(results).sort_values('Sharpe_Ratio', ascending=False)
Search
optimal = parameter_search('EURUSD=X', '2023-01-01', '2026-03-15',
fast_range=range(10, 15),
slow_range=range(22, 30),
signal_range=range(7, 12))
print(optimal.head(10))
``