import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import ccxt
import talib
import logging
logger = logging.getLogger(__name__)
class CryptoMACDBacktester:
    """Backtest MACD on cryptocurrencies with 24/7 market handling"""
def __init__(self, exchange='binance', symbol='BTC/USDT', timeframe='1h',
                 fast=8, slow=17, signal=9):
        """
        Initialize crypto MACD backtester
Args:
            exchange: CCXT exchange name
            symbol: Trading pair (BTC/USDT, ETH/USDT, etc.)
            timeframe: Candle timeframe (1h, 4h, 1d)
            fast: Fast EMA (8-12 for crypto)
            slow: Slow EMA (17-26 for crypto)
            signal: Signal EMA (9)
        """
        self.exchange = ccxt.__dict__[exchange]()
        self.symbol = symbol
        self.timeframe = timeframe
        self.fast = fast
        self.slow = slow
        self.signal = signal
        self.df = None
        self.metrics = {}
def fetch_crypto_data(self, start_date, end_date):
        """
        Fetch OHLCV data from crypto exchange
CCXT provides unified API across exchanges
        """
        logger.info(f"Fetching {self.symbol} on {self.timeframe} timeframe")
# Convert dates to milliseconds
        since = int(pd.to_datetime(start_date).timestamp() * 1000)
        limit = 1000
all_candles = []
        current_since = since
# CCXT returns max 1000 candles per request
        while True:
            try:
                candles = self.exchange.fetch_ohlcv(self.symbol, self.timeframe, since=current_since, limit=limit)
if not candles:
                    break
all_candles.extend(candles)
# Stop if we've reached end date
                last_candle_time = candles[-1][0]
                if last_candle_time > int(pd.to_datetime(end_date).timestamp() * 1000):
                    break
current_since = last_candle_time + 1
except Exception as e:
                logger.error(f"Error fetching data: {str(e)}")
                break
# Convert to DataFrame
        self.df = pd.DataFrame(
            all_candles,
            columns=['Timestamp', 'Open', 'High', 'Low', 'Close', 'Volume']
        )
self.df['Timestamp'] = pd.to_datetime(self.df['Timestamp'], unit='ms')
        self.df = self.df.set_index('Timestamp')
logger.info(f"Fetched {len(self.df)} candles")
        return self.df
def calculate_crypto_macd(self):
        """Calculate MACD with crypto-specific parameters"""
        logger.info(f"Calculating MACD ({self.fast},{self.slow},{self.signal})")
# Calculate MACD
        macd, signal_line, histogram = talib.MACD(
            self.df['Close'].values,
            fastperiod=self.fast,
            slowperiod=self.slow,
            signalperiod=self.signal
        )
self.df['MACD'] = macd
        self.df['Signal'] = signal_line
        self.df['Histogram'] = histogram
# Additional crypto-specific indicators
        self.df['RSI'] = talib.RSI(self.df['Close'].values, timeperiod=14)
        self.df['ATR'] = talib.ATR(self.df['High'].values, self.df['Low'].values,
                                   self.df['Close'].values, timeperiod=14)
return self.df
def generate_crypto_signals(self):
        """Generate signals with crypto-specific filters"""
        logger.info("Generating trading signals")
self.df['MACD_prev'] = self.df['MACD'].shift(1)
        self.df['Signal_prev'] = self.df['Signal'].shift(1)
# Basic crossover
        buy_signal = (self.df['MACD_prev'] <= self.df['Signal_prev']) & (self.df['MACD'] > self.df['Signal'])
        sell_signal = (self.df['MACD_prev'] >= self.df['Signal_prev']) & (self.df['MACD'] < self.df['Signal'])
# Crypto filter 1: RSI confirmation (avoid overbought/oversold extremes)
        buy_valid = buy_signal & (self.df['RSI'] < 70)
        sell_valid = sell_signal & (self.df['RSI'] > 30)
# Crypto filter 2: Histogram confirmation (momentum alignment)
        buy_valid = buy_valid & (self.df['Histogram'] > 0)
        sell_valid = sell_valid & (self.df['Histogram'] < 0)
self.df['Buy_Signal'] = buy_valid.astype(int)
        self.df['Sell_Signal'] = sell_valid.astype(int)
# Position holding
        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]
return self.df
def calculate_crypto_returns(self, transaction_cost=0.001, slippage=0.002):
        """
        Calculate returns with crypto-specific costs
Args:
            transaction_cost: Exchange fee (0.1% = 0.001)
            slippage: Additional market impact (0.2% = 0.002 typical)
        """
        logger.info(f"Calculating returns (fee={transaction_cost100:.2f}%, slippage={slippage100:.2f}%)")
self.df['Daily_Return'] = self.df['Close'].pct_change()
# Transaction costs (both directions)
        total_cost = transaction_cost + slippage
        self.df['Position_Change'] = self.df['Position'].diff().abs()
        self.df['Transaction_Cost'] = self.df['Position_Change'] * total_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_crypto_metrics(self):
        """Calculate crypto-specific performance metrics"""
        strategy_returns = self.df['Strategy_Return'].dropna()
total_return = (self.df['Cumulative_Strategy'].iloc[-1] - 1) * 100
        bh_return = (self.df['Cumulative_BH'].iloc[-1] - 1) * 100
sharpe = (strategy_returns.mean() / strategy_returns.std()) * np.sqrt(8760) if strategy_returns.std() > 0 else 0
win_rate = len(strategy_returns[strategy_returns > 0]) / len(strategy_returns) * 100
# For crypto, drawdown is critical (market can crash 50%+)
        cum_max = self.df['Cumulative_Strategy'].expanding().max()
        max_drawdown = ((self.df['Cumulative_Strategy'] - cum_max) / cum_max).min() * 100
self.metrics = {
            'Total_Return': total_return,
            'BH_Return': bh_return,
            'Excess_Return': total_return - bh_return,
            'Sharpe_Ratio': sharpe,
            'Win_Rate': win_rate,
            'Max_Drawdown': max_drawdown,
            'Profit_Factor': (strategy_returns[strategy_returns > 0].sum() /
                             abs(strategy_returns[strategy_returns < 0].sum())),
            'Total_Trades': len(strategy_returns[strategy_returns != 0]),
        }
return self.metrics
def backtest(self, start_date, end_date):
        """Run complete backtest"""
        self.fetch_crypto_data(start_date, end_date)
        self.calculate_crypto_macd()
        self.generate_crypto_signals()
        self.calculate_crypto_returns()
        return self.calculate_crypto_metrics()

def optimize_crypto_parameters(exchange, symbol, timeframe, start_date, end_date):
    """Find optimal MACD parameters for crypto asset"""
    results = []
for fast in range(6, 12):
        for slow in range(18, 35):
            if slow <= fast:
                continue
            for signal in range(7, 12):
                backtester = CryptoMACDBacktester(exchange, symbol, timeframe, fast, slow, 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)

def adjust_macd_for_volatility(df, base_fast=10, base_slow=30):
    """Adjust MACD parameters based on current volatility"""
    volatility = df['Close'].pct_change().rolling(20).std()
    volatility_ratio = volatility / volatility.mean()
# Increase periods when volatility is high
    df['Fast'] = (base_fast * volatility_ratio).astype(int)
    df['Slow'] = (base_slow * volatility_ratio).astype(int)
return df

def calculate_crypto_position_size(account_balance, atr, risk_percent=0.02):
    """
    Position sizing for crypto with ATR-based stops
Args:
        account_balance: Total account balance
        atr: Average True Range (volatility)
        risk_percent: Risk per trade (2% = 0.02)
    """
    risk_amount = account_balance * risk_percent
    position_size = risk_amount / atr
return position_size