pip install pandas numpy yfinance ta-lib matplotlib seaborn jupyter

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}")

# 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')

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))