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

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