Crypto Quantitative Trading Part 2: Advanced Strategies and Backtesting Framework

 1# Example: Overfitted vs Robust Strategy
 2
 3# ❌ Overfitted (curve-fit to historical data)
 4def overfitted_strategy():
 5    """
 6    Buy when:
 7    - RSI crosses 31.7 (not 30 or 32)
 8    - MACD histogram > 42.3
 9    - Hour of day is 14 or 23
10    - Day of month is prime number
11
12    This likely worked great on historical data
13    but will fail miserably going forward.
14    """
15    pass
16
17# ✅ Robust (based on sound principles)
18def robust_strategy():
19    """
20    Buy when:
21    - Oversold (RSI < 30)
22    - Positive momentum (MACD positive)
23    - Simple, interpretable rules
24    - Works across multiple timeframes
25
26    This strategy has economic rationale
27    and is likely to generalize to new data.
28    """
29    pass

  1from dataclasses import dataclass
  2from typing import List, Dict, Optional
  3from enum import Enum
  4import pandas as pd
  5import numpy as np
  6from datetime import datetime
  7
  8class OrderSide(Enum):
  9    BUY = "buy"
 10    SELL = "sell"
 11
 12class OrderType(Enum):
 13    MARKET = "market"
 14    LIMIT = "limit"
 15
 16@dataclass
 17class Order:
 18    """Represents a trading order"""
 19    timestamp: datetime
 20    symbol: str
 21    side: OrderSide
 22    order_type: OrderType
 23    quantity: float
 24    price: Optional[float] = None  # For limit orders
 25    filled_price: Optional[float] = None
 26    filled_quantity: float = 0.0
 27    status: str = "pending"  # pending, filled, cancelled
 28    commission: float = 0.0
 29
 30@dataclass
 31class Position:
 32    """Represents a trading position"""
 33    symbol: str
 34    quantity: float
 35    entry_price: float
 36    entry_time: datetime
 37    current_price: float = 0.0
 38    unrealized_pnl: float = 0.0
 39    realized_pnl: float = 0.0
 40
 41@dataclass
 42class Trade:
 43    """Represents a completed trade"""
 44    entry_time: datetime
 45    exit_time: datetime
 46    symbol: str
 47    side: OrderSide
 48    entry_price: float
 49    exit_price: float
 50    quantity: float
 51    pnl: float
 52    pnl_percent: float
 53    commission: float
 54    duration: float  # hours
 55
 56class BacktestEngine:
 57    """
 58    Core backtesting engine
 59    """
 60
 61    def __init__(self, initial_capital: float = 100000,
 62                 commission_rate: float = 0.001,
 63                 slippage: float = 0.0005):
 64        """
 65        Initialize backtesting engine
 66
 67        initial_capital: Starting capital in USD
 68        commission_rate: Commission per trade (0.001 = 0.1%)
 69        slippage: Estimated slippage (0.0005 = 0.05%)
 70        """
 71        self.initial_capital = initial_capital
 72        self.capital = initial_capital
 73        self.commission_rate = commission_rate
 74        self.slippage = slippage
 75
 76        self.positions: Dict[str, Position] = {}
 77        self.orders: List[Order] = []
 78        self.trades: List[Trade] = []
 79        self.equity_curve: List[Dict] = []
 80
 81        self.current_time = None
 82        self.current_prices: Dict[str, float] = {}
 83
 84    def update_time(self, timestamp: datetime, prices: Dict[str, float]):
 85        """Update current time and prices"""
 86        self.current_time = timestamp
 87        self.current_prices = prices
 88
 89        # Update position values
 90        for symbol, position in self.positions.items():
 91            if symbol in prices:
 92                position.current_price = prices[symbol]
 93                position.unrealized_pnl = (
 94                    (position.current_price - position.entry_price) *
 95                    position.quantity
 96                )
 97
 98        # Record equity
 99        total_equity = self.calculate_total_equity()
100        self.equity_curve.append({
101            'timestamp': timestamp,
102            'capital': self.capital,
103            'position_value': total_equity - self.capital,
104            'total_equity': total_equity,
105        })
106
107    def calculate_total_equity(self) -> float:
108        """Calculate total portfolio equity"""
109        position_value = sum(
110            pos.quantity * pos.current_price
111            for pos in self.positions.values()
112        )
113        return self.capital + position_value
114
115    def place_order(self, symbol: str, side: OrderSide,
116                    quantity: float, order_type: OrderType = OrderType.MARKET,
117                    limit_price: Optional[float] = None) -> Order:
118        """
119        Place an order
120
121        Returns: Order object
122        """
123        order = Order(
124            timestamp=self.current_time,
125            symbol=symbol,
126            side=side,
127            order_type=order_type,
128            quantity=quantity,
129            price=limit_price,
130        )
131
132        # Execute market orders immediately
133        if order_type == OrderType.MARKET:
134            self._execute_order(order)
135
136        self.orders.append(order)
137        return order
138
139    def _execute_order(self, order: Order):
140        """Execute an order"""
141        if order.symbol not in self.current_prices:
142            order.status = "rejected"
143            return
144
145        # Calculate execution price with slippage
146        current_price = self.current_prices[order.symbol]
147        if order.side == OrderSide.BUY:
148            execution_price = current_price * (1 + self.slippage)
149        else:
150            execution_price = current_price * (1 - self.slippage)
151
152        # Calculate costs
153        order_value = order.quantity * execution_price
154        commission = order_value * self.commission_rate
155
156        # Check if we have enough capital (for buy orders)
157        if order.side == OrderSide.BUY:
158            total_cost = order_value + commission
159            if total_cost > self.capital:
160                order.status = "rejected"
161                return
162
163        # Execute the order
164        order.filled_price = execution_price
165        order.filled_quantity = order.quantity
166        order.commission = commission
167        order.status = "filled"
168
169        # Update positions
170        if order.side == OrderSide.BUY:
171            self._open_position(order)
172        else:
173            self._close_position(order)
174
175    def _open_position(self, order: Order):
176        """Open or add to a position"""
177        if order.symbol in self.positions:
178            # Add to existing position (average entry price)
179            pos = self.positions[order.symbol]
180            total_quantity = pos.quantity + order.filled_quantity
181            total_cost = (
182                pos.entry_price * pos.quantity +
183                order.filled_price * order.filled_quantity
184            )
185            pos.entry_price = total_cost / total_quantity
186            pos.quantity = total_quantity
187        else:
188            # Create new position
189            self.positions[order.symbol] = Position(
190                symbol=order.symbol,
191                quantity=order.filled_quantity,
192                entry_price=order.filled_price,
193                entry_time=order.timestamp,
194                current_price=order.filled_price,
195            )
196
197        # Update capital
198        self.capital -= (order.filled_quantity * order.filled_price + order.commission)
199
200    def _close_position(self, order: Order):
201        """Close or reduce a position"""
202        if order.symbol not in self.positions:
203            order.status = "rejected"
204            return
205
206        pos = self.positions[order.symbol]
207
208        # Calculate P&L
209        pnl = (order.filled_price - pos.entry_price) * order.filled_quantity
210        pnl -= order.commission
211
212        # Create trade record
213        trade = Trade(
214            entry_time=pos.entry_time,
215            exit_time=order.timestamp,
216            symbol=order.symbol,
217            side=OrderSide.BUY,  # Position side
218            entry_price=pos.entry_price,
219            exit_price=order.filled_price,
220            quantity=order.filled_quantity,
221            pnl=pnl,
222            pnl_percent=(order.filled_price / pos.entry_price - 1) * 100,
223            commission=order.commission,
224            duration=(order.timestamp - pos.entry_time).total_seconds() / 3600,
225        )
226        self.trades.append(trade)
227
228        # Update capital
229        self.capital += (order.filled_quantity * order.filled_price - order.commission)
230
231        # Update or remove position
232        pos.quantity -= order.filled_quantity
233        pos.realized_pnl += pnl
234
235        if pos.quantity <= 0:
236            del self.positions[order.symbol]
237
238    def get_position(self, symbol: str) -> Optional[Position]:
239        """Get current position for a symbol"""
240        return self.positions.get(symbol)
241
242    def get_equity_curve(self) -> pd.DataFrame:
243        """Get equity curve as DataFrame"""
244        return pd.DataFrame(self.equity_curve).set_index('timestamp')
245
246    def get_trades_df(self) -> pd.DataFrame:
247        """Get all trades as DataFrame"""
248        if not self.trades:
249            return pd.DataFrame()
250
251        return pd.DataFrame([vars(t) for t in self.trades])

 1from abc import ABC, abstractmethod
 2
 3class TradingStrategy(ABC):
 4    """
 5    Base class for trading strategies
 6    """
 7
 8    def __init__(self, name: str):
 9        self.name = name
10        self.params = {}
11
12    @abstractmethod
13    def generate_signals(self, df: pd.DataFrame) -> pd.Series:
14        """
15        Generate trading signals
16
17        Returns: Series with values -1 (sell), 0 (hold), 1 (buy)
18        """
19        pass
20
21    @abstractmethod
22    def calculate_position_size(self, signal: float, capital: float,
23                                current_price: float) -> float:
24        """
25        Calculate position size for a signal
26
27        Returns: Position size (quantity)
28        """
29        pass
30
31    def set_params(self, **kwargs):
32        """Update strategy parameters"""
33        self.params.update(kwargs)
34
35    def get_description(self) -> str:
36        """Get strategy description"""
37        return f"{self.name} with params: {self.params}"

  1class MovingAverageCrossover(TradingStrategy):
  2    """
  3    Classic trend following strategy using moving average crossovers
  4
  5    Buy: Fast MA crosses above Slow MA
  6    Sell: Fast MA crosses below Slow MA
  7    """
  8
  9    def __init__(self, fast_period=20, slow_period=50, atr_period=14):
 10        super().__init__("MA Crossover")
 11        self.params = {
 12            'fast_period': fast_period,
 13            'slow_period': slow_period,
 14            'atr_period': atr_period,
 15        }
 16
 17    def generate_signals(self, df: pd.DataFrame) -> pd.Series:
 18        """Generate signals based on MA crossover"""
 19        fast = df['close'].ewm(span=self.params['fast_period']).mean()
 20        slow = df['close'].ewm(span=self.params['slow_period']).mean()
 21
 22        # Generate crossover signals
 23        signals = pd.Series(0, index=df.index)
 24
 25        # Buy signal: Fast crosses above Slow
 26        signals[(fast > slow) & (fast.shift() <= slow.shift())] = 1
 27
 28        # Sell signal: Fast crosses below Slow
 29        signals[(fast < slow) & (fast.shift() >= slow.shift())] = -1
 30
 31        return signals
 32
 33    def calculate_position_size(self, signal: float, capital: float,
 34                                current_price: float) -> float:
 35        """
 36        Calculate position size using fixed fractional method
 37
 38        Risk 2% of capital per trade
 39        """
 40        if signal == 0:
 41            return 0.0
 42
 43        risk_per_trade = capital * 0.02
 44        position_value = capital * 0.95  # Use 95% of capital max
 45
 46        # Calculate quantity
 47        quantity = min(
 48            position_value / current_price,
 49            risk_per_trade / (current_price * 0.02)  # 2% stop loss
 50        )
 51
 52        return quantity
 53
 54# Example usage
 55def backtest_ma_crossover(df: pd.DataFrame, initial_capital=100000):
 56    """
 57    Backtest MA Crossover strategy
 58    """
 59    # Initialize
 60    engine = BacktestEngine(initial_capital=initial_capital)
 61    strategy = MovingAverageCrossover(fast_period=20, slow_period=50)
 62
 63    # Generate signals
 64    signals = strategy.generate_signals(df)
 65
 66    # Run backtest
 67    position = None
 68
 69    for timestamp, row in df.iterrows():
 70        # Update engine
 71        engine.update_time(timestamp, {'BTC/USDT': row['close']})
 72
 73        signal = signals.loc[timestamp]
 74
 75        # Get current position
 76        current_pos = engine.get_position('BTC/USDT')
 77
 78        # Execute signals
 79        if signal == 1 and current_pos is None:
 80            # Buy signal and no position
 81            quantity = strategy.calculate_position_size(
 82                signal,
 83                engine.capital,
 84                row['close']
 85            )
 86            engine.place_order('BTC/USDT', OrderSide.BUY, quantity)
 87
 88        elif signal == -1 and current_pos is not None:
 89            # Sell signal and have position
 90            engine.place_order(
 91                'BTC/USDT',
 92                OrderSide.SELL,
 93                current_pos.quantity
 94            )
 95
 96    return engine
 97
 98# Run backtest
 99from crypto_quantitative_trading_part1_fundamentals import CryptoDataCollector
100
101collector = CryptoDataCollector('binance')
102df = collector.fetch_historical_data('BTC/USDT', '1h', '2023-01-01', '2024-01-01')
103
104# Add indicators
105from crypto_quantitative_trading_part1_fundamentals import TrendIndicators
106df['sma_20'] = TrendIndicators.sma(df['close'], 20)
107df['sma_50'] = TrendIndicators.sma(df['close'], 50)
108
109# Backtest
110result = backtest_ma_crossover(df)
111equity = result.get_equity_curve()
112trades = result.get_trades_df()
113
114print(f"\nBacktest Results:")
115print(f"Initial Capital: ${result.initial_capital:,.2f}")
116print(f"Final Equity: ${result.calculate_total_equity():,.2f}")
117print(f"Total Return: {(result.calculate_total_equity() / result.initial_capital - 1) * 100:.2f}%")
118print(f"Number of Trades: {len(trades)}")

 1class ADXTrendFollowing(TradingStrategy):
 2    """
 3    Enhanced trend following with ADX filter
 4
 5    Only trade when trend is strong (ADX > threshold)
 6    """
 7
 8    def __init__(self, fast_period=12, slow_period=26, adx_period=14, adx_threshold=25):
 9        super().__init__("ADX Trend Following")
10        self.params = {
11            'fast_period': fast_period,
12            'slow_period': slow_period,
13            'adx_period': adx_period,
14            'adx_threshold': adx_threshold,
15        }
16
17    def generate_signals(self, df: pd.DataFrame) -> pd.Series:
18        """Generate signals with ADX filter"""
19        from crypto_quantitative_trading_part1_fundamentals import TrendIndicators
20
21        # Calculate indicators
22        fast = df['close'].ewm(span=self.params['fast_period']).mean()
23        slow = df['close'].ewm(span=self.params['slow_period']).mean()
24        adx, plus_di, minus_di = TrendIndicators.adx(
25            df['high'], df['low'], df['close'], self.params['adx_period']
26        )
27
28        # Generate signals
29        signals = pd.Series(0, index=df.index)
30
31        # Only trade when ADX indicates strong trend
32        strong_trend = adx > self.params['adx_threshold']
33
34        # Buy: Fast > Slow AND ADX > threshold AND +DI > -DI
35        buy_condition = (
36            (fast > slow) &
37            (fast.shift() <= slow.shift()) &
38            strong_trend &
39            (plus_di > minus_di)
40        )
41        signals[buy_condition] = 1
42
43        # Sell: Fast < Slow OR ADX weakening
44        sell_condition = (
45            (fast < slow) &
46            (fast.shift() >= slow.shift())
47        ) | (adx < self.params['adx_threshold'] * 0.7)
48
49        signals[sell_condition] = -1
50
51        return signals
52
53    def calculate_position_size(self, signal: float, capital: float,
54                                current_price: float) -> float:
55        """Kelly Criterion based position sizing"""
56        if signal == 0:
57            return 0.0
58
59        # Simplified Kelly: f = (p * b - q) / b
60        # Assuming 55% win rate, 1.5:1 reward:risk
61        win_rate = 0.55
62        reward_risk = 1.5
63
64        kelly_fraction = (win_rate * reward_risk - (1 - win_rate)) / reward_risk
65        kelly_fraction = max(0, min(kelly_fraction, 0.25))  # Cap at 25%
66
67        position_value = capital * kelly_fraction
68        quantity = position_value / current_price
69
70        return quantity

 1class BollingerMeanReversion(TradingStrategy):
 2    """
 3    Mean reversion strategy using Bollinger Bands
 4
 5    Buy: Price touches lower band (oversold)
 6    Sell: Price reaches middle band or upper band
 7    """
 8
 9    def __init__(self, bb_period=20, bb_std=2.0, rsi_period=14):
10        super().__init__("Bollinger Mean Reversion")
11        self.params = {
12            'bb_period': bb_period,
13            'bb_std': bb_std,
14            'rsi_period': rsi_period,
15        }
16
17    def generate_signals(self, df: pd.DataFrame) -> pd.Series:
18        """Generate mean reversion signals"""
19        from crypto_quantitative_trading_part1_fundamentals import (
20            VolatilityIndicators, MomentumIndicators
21        )
22
23        # Calculate indicators
24        bb_middle, bb_upper, bb_lower = VolatilityIndicators.bollinger_bands(
25            df['close'],
26            self.params['bb_period'],
27            self.params['bb_std']
28        )
29        rsi = MomentumIndicators.rsi(df['close'], self.params['rsi_period'])
30
31        # Calculate Bollinger Band position
32        bb_position = (df['close'] - bb_lower) / (bb_upper - bb_lower)
33
34        signals = pd.Series(0, index=df.index)
35
36        # Buy: Price at lower band AND RSI oversold
37        buy_condition = (
38            (bb_position < 0.1) &  # Near lower band
39            (rsi < 30) &  # Oversold
40            (df['close'] > df['close'].shift())  # Starting to bounce
41        )
42        signals[buy_condition] = 1
43
44        # Sell: Price at middle band or upper band
45        sell_condition = (
46            (bb_position > 0.5) |  # Above middle
47            (rsi > 70)  # Overbought
48        )
49        signals[sell_condition] = -1
50
51        return signals
52
53    def calculate_position_size(self, signal: float, capital: float,
54                                current_price: float) -> float:
55        """
56        Fixed fractional position sizing
57        Risk 1% per trade (mean reversion is riskier)
58        """
59        if signal == 0:
60            return 0.0
61
62        risk_per_trade = capital * 0.01
63        position_value = capital * 0.5  # Max 50% of capital
64
65        quantity = min(
66            position_value / current_price,
67            risk_per_trade / (current_price * 0.03)  # 3% stop loss
68        )
69
70        return quantity

 1class RSIVolumeMeanReversion(TradingStrategy):
 2    """
 3    Mean reversion using RSI with volume confirmation
 4
 5    Buy: RSI oversold + volume spike (capitulation)
 6    Sell: RSI overbought + volume spike (exhaustion)
 7    """
 8
 9    def __init__(self, rsi_period=14, volume_period=20, volume_threshold=1.5):
10        super().__init__("RSI Volume Mean Reversion")
11        self.params = {
12            'rsi_period': rsi_period,
13            'volume_period': volume_period,
14            'volume_threshold': volume_threshold,
15        }
16
17    def generate_signals(self, df: pd.DataFrame) -> pd.Series:
18        """Generate RSI + Volume signals"""
19        from crypto_quantitative_trading_part1_fundamentals import MomentumIndicators
20
21        # Calculate RSI
22        rsi = MomentumIndicators.rsi(df['close'], self.params['rsi_period'])
23
24        # Calculate volume ratio
25        avg_volume = df['volume'].rolling(self.params['volume_period']).mean()
26        volume_ratio = df['volume'] / avg_volume
27
28        signals = pd.Series(0, index=df.index)
29
30        # Buy: RSI < 30 + Volume spike (panic selling)
31        buy_condition = (
32            (rsi < 30) &
33            (volume_ratio > self.params['volume_threshold'])
34        )
35        signals[buy_condition] = 1
36
37        # Sell: RSI > 70 + Volume spike (FOMO buying)
38        sell_condition = (
39            (rsi > 70) &
40            (volume_ratio > self.params['volume_threshold'])
41        )
42        signals[sell_condition] = -1
43
44        return signals
45
46    def calculate_position_size(self, signal: float, capital: float,
47                                current_price: float) -> float:
48        """Scale position by signal strength"""
49        if signal == 0:
50            return 0.0
51
52        base_size = capital * 0.02 / current_price
53        return base_size

 1class BTCETHPairsTrad(TradingStrategy):
 2    """
 3    Statistical arbitrage between BTC and ETH
 4
 5    Trade the spread between BTC/ETH ratio and its mean
 6    """
 7
 8    def __init__(self, lookback_period=30, entry_z_score=2.0, exit_z_score=0.5):
 9        super().__init__("BTC-ETH Pairs Trading")
10        self.params = {
11            'lookback_period': lookback_period,
12            'entry_z_score': entry_z_score,
13            'exit_z_score': exit_z_score,
14        }
15
16    def generate_signals(self, btc_df: pd.DataFrame, eth_df: pd.DataFrame) -> Dict:
17        """
18        Generate pairs trading signals
19
20        Returns: Dict with 'btc_signal' and 'eth_signal'
21        """
22        # Calculate BTC/ETH ratio
23        ratio = btc_df['close'] / eth_df['close']
24
25        # Calculate rolling statistics
26        period = self.params['lookback_period']
27        ratio_mean = ratio.rolling(period).mean()
28        ratio_std = ratio.rolling(period).std()
29
30        # Calculate z-score
31        z_score = (ratio - ratio_mean) / ratio_std
32
33        # Generate signals
34        btc_signals = pd.Series(0, index=btc_df.index)
35        eth_signals = pd.Series(0, index=eth_df.index)
36
37        # When z-score is high: BTC overvalued, ETH undervalued
38        # Short BTC, Long ETH
39        overvalued = z_score > self.params['entry_z_score']
40        btc_signals[overvalued] = -1
41        eth_signals[overvalued] = 1
42
43        # When z-score is low: BTC undervalued, ETH overvalued
44        # Long BTC, Short ETH
45        undervalued = z_score < -self.params['entry_z_score']
46        btc_signals[undervalued] = 1
47        eth_signals[undervalued] = -1
48
49        # Exit when z-score returns to mean
50        exit_long = (z_score < self.params['exit_z_score']) & (z_score.shift() >= self.params['exit_z_score'])
51        exit_short = (z_score > -self.params['exit_z_score']) & (z_score.shift() <= -self.params['exit_z_score'])
52
53        btc_signals[exit_long | exit_short] = 0
54        eth_signals[exit_long | exit_short] = 0
55
56        return {
57            'btc_signal': btc_signals,
58            'eth_signal': eth_signals,
59            'z_score': z_score,
60        }
61
62    def calculate_position_size(self, signal: float, capital: float,
63                                current_price: float) -> float:
64        """Equal dollar weighting for pairs"""
65        if signal == 0:
66            return 0.0
67
68        # Use 50% of capital for each leg
69        position_value = capital * 0.5
70        quantity = position_value / current_price
71
72        return quantity

  1class RiskManager:
  2    """
  3    Comprehensive risk management system
  4    """
  5
  6    def __init__(self, max_position_size=0.3, max_daily_loss=0.02,
  7                 stop_loss=0.05, take_profit=0.15):
  8        """
  9        max_position_size: Maximum fraction of capital per position
 10        max_daily_loss: Maximum daily loss (fraction of capital)
 11        stop_loss: Stop loss percentage (0.05 = 5%)
 12        take_profit: Take profit percentage (0.15 = 15%)
 13        """
 14        self.max_position_size = max_position_size
 15        self.max_daily_loss = max_daily_loss
 16        self.stop_loss = stop_loss
 17        self.take_profit = take_profit
 18
 19        self.daily_pnl = 0.0
 20        self.daily_start_capital = 0.0
 21        self.last_reset_date = None
 22
 23    def check_stop_loss(self, position: Position) -> bool:
 24        """Check if position hit stop loss"""
 25        if position.quantity == 0:
 26            return False
 27
 28        pnl_percent = (position.current_price / position.entry_price - 1)
 29        return pnl_percent <= -self.stop_loss
 30
 31    def check_take_profit(self, position: Position) -> bool:
 32        """Check if position hit take profit"""
 33        if position.quantity == 0:
 34            return False
 35
 36        pnl_percent = (position.current_price / position.entry_price - 1)
 37        return pnl_percent >= self.take_profit
 38
 39    def check_daily_loss_limit(self, current_capital: float,
 40                               current_date: datetime) -> bool:
 41        """Check if daily loss limit reached"""
 42        # Reset daily tracking at start of new day
 43        if self.last_reset_date is None or current_date.date() != self.last_reset_date:
 44            self.daily_start_capital = current_capital
 45            self.daily_pnl = 0.0
 46            self.last_reset_date = current_date.date()
 47
 48        # Calculate daily loss
 49        daily_loss = (current_capital - self.daily_start_capital) / self.daily_start_capital
 50
 51        return daily_loss <= -self.max_daily_loss
 52
 53    def validate_position_size(self, quantity: float, price: float,
 54                               total_capital: float) -> float:
 55        """
 56        Validate and adjust position size
 57
 58        Returns: Adjusted quantity
 59        """
 60        position_value = quantity * price
 61        max_value = total_capital * self.max_position_size
 62
 63        if position_value > max_value:
 64            quantity = max_value / price
 65
 66        return quantity
 67
 68    def calculate_optimal_stop_loss(self, entry_price: float,
 69                                    atr: float, atr_multiplier: float = 2.0) -> float:
 70        """
 71        Calculate optimal stop loss based on ATR
 72
 73        atr: Average True Range
 74        atr_multiplier: Number of ATRs for stop
 75        """
 76        stop_distance = atr * atr_multiplier
 77        stop_price = entry_price - stop_distance
 78
 79        return stop_price
 80
 81# Enhanced backtesting with risk management
 82def backtest_with_risk_management(df: pd.DataFrame, strategy: TradingStrategy,
 83                                  initial_capital: float = 100000):
 84    """
 85    Backtest strategy with comprehensive risk management
 86    """
 87    engine = BacktestEngine(initial_capital=initial_capital)
 88    risk_manager = RiskManager(
 89        max_position_size=0.3,
 90        max_daily_loss=0.02,
 91        stop_loss=0.05,
 92        take_profit=0.15
 93    )
 94
 95    # Calculate ATR for stop loss
 96    from crypto_quantitative_trading_part1_fundamentals import VolatilityIndicators
 97    df['atr'] = VolatilityIndicators.atr(df['high'], df['low'], df['close'])
 98
 99    # Generate signals
100    signals = strategy.generate_signals(df)
101
102    # Run backtest
103    for timestamp, row in df.iterrows():
104        # Update engine
105        engine.update_time(timestamp, {'BTC/USDT': row['close']})
106
107        # Check daily loss limit
108        if risk_manager.check_daily_loss_limit(engine.capital, timestamp):
109            # Stop trading for the day
110            continue
111
112        current_pos = engine.get_position('BTC/USDT')
113
114        # Check stop loss / take profit
115        if current_pos is not None:
116            current_pos.current_price = row['close']
117
118            if risk_manager.check_stop_loss(current_pos):
119                # Hit stop loss - close position
120                engine.place_order('BTC/USDT', OrderSide.SELL, current_pos.quantity)
121                continue
122
123            if risk_manager.check_take_profit(current_pos):
124                # Hit take profit - close position
125                engine.place_order('BTC/USDT', OrderSide.SELL, current_pos.quantity)
126                continue
127
128        # Process signals
129        signal = signals.loc[timestamp]
130
131        if signal == 1 and current_pos is None:
132            # Buy signal
133            quantity = strategy.calculate_position_size(
134                signal, engine.capital, row['close']
135            )
136
137            # Validate position size
138            quantity = risk_manager.validate_position_size(
139                quantity, row['close'], engine.capital
140            )
141
142            if quantity > 0:
143                engine.place_order('BTC/USDT', OrderSide.BUY, quantity)
144
145        elif signal == -1 and current_pos is not None:
146            # Sell signal
147            engine.place_order('BTC/USDT', OrderSide.SELL, current_pos.quantity)
148
149    return engine

  1class PerformanceAnalyzer:
  2    """
  3    Calculate comprehensive performance metrics
  4    """
  5
  6    @staticmethod
  7    def calculate_metrics(equity_curve: pd.DataFrame,
  8                         trades_df: pd.DataFrame,
  9                         initial_capital: float,
 10                         risk_free_rate: float = 0.04) -> Dict:
 11        """
 12        Calculate all performance metrics
 13
 14        risk_free_rate: Annual risk-free rate (default 4%)
 15        """
 16        if equity_curve.empty or trades_df.empty:
 17            return {}
 18
 19        # Total return
 20        final_equity = equity_curve['total_equity'].iloc[-1]
 21        total_return = (final_equity / initial_capital - 1) * 100
 22
 23        # Period
 24        days = (equity_curve.index[-1] - equity_curve.index[0]).days
 25        years = days / 365.25
 26
 27        # Annualized return
 28        annualized_return = ((final_equity / initial_capital) ** (1/years) - 1) * 100
 29
 30        # Returns series
 31        returns = equity_curve['total_equity'].pct_change().dropna()
 32
 33        # Volatility
 34        daily_vol = returns.std()
 35        annualized_vol = daily_vol * np.sqrt(365 * 24) * 100  # For hourly data
 36
 37        # Sharpe Ratio
 38        excess_return = annualized_return / 100 - risk_free_rate
 39        sharpe_ratio = excess_return / (annualized_vol / 100) if annualized_vol > 0 else 0
 40
 41        # Maximum Drawdown
 42        cumulative = (1 + returns).cumprod()
 43        running_max = cumulative.expanding().max()
 44        drawdown = (cumulative - running_max) / running_max
 45        max_drawdown = drawdown.min() * 100
 46
 47        # Win Rate
 48        winning_trades = len(trades_df[trades_df['pnl'] > 0])
 49        total_trades = len(trades_df)
 50        win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0
 51
 52        # Average Win/Loss
 53        avg_win = trades_df[trades_df['pnl'] > 0]['pnl'].mean() if winning_trades > 0 else 0
 54        avg_loss = trades_df[trades_df['pnl'] < 0]['pnl'].mean() if total_trades - winning_trades > 0 else 0
 55
 56        # Profit Factor
 57        total_wins = trades_df[trades_df['pnl'] > 0]['pnl'].sum()
 58        total_losses = abs(trades_df[trades_df['pnl'] < 0]['pnl'].sum())
 59        profit_factor = total_wins / total_losses if total_losses > 0 else 0
 60
 61        # Expectancy
 62        expectancy = (win_rate / 100 * avg_win) - ((1 - win_rate / 100) * abs(avg_loss))
 63
 64        # Sortino Ratio (downside deviation)
 65        downside_returns = returns[returns < 0]
 66        downside_std = downside_returns.std() * np.sqrt(365 * 24)
 67        sortino_ratio = excess_return / downside_std if downside_std > 0 else 0
 68
 69        # Calmar Ratio
 70        calmar_ratio = (annualized_return / 100) / abs(max_drawdown / 100) if max_drawdown != 0 else 0
 71
 72        return {
 73            'initial_capital': initial_capital,
 74            'final_equity': final_equity,
 75            'total_return': total_return,
 76            'annualized_return': annualized_return,
 77            'annualized_volatility': annualized_vol,
 78            'sharpe_ratio': sharpe_ratio,
 79            'sortino_ratio': sortino_ratio,
 80            'calmar_ratio': calmar_ratio,
 81            'max_drawdown': max_drawdown,
 82            'total_trades': total_trades,
 83            'winning_trades': winning_trades,
 84            'losing_trades': total_trades - winning_trades,
 85            'win_rate': win_rate,
 86            'avg_win': avg_win,
 87            'avg_loss': avg_loss,
 88            'profit_factor': profit_factor,
 89            'expectancy': expectancy,
 90        }
 91
 92    @staticmethod
 93    def print_metrics(metrics: Dict):
 94        """Print metrics in readable format"""
 95        print("\n" + "="*60)
 96        print("PERFORMANCE METRICS")
 97        print("="*60)
 98
 99        print(f"\nCapital:")
100        print(f"  Initial: ${metrics['initial_capital']:,.2f}")
101        print(f"  Final: ${metrics['final_equity']:,.2f}")
102
103        print(f"\nReturns:")
104        print(f"  Total Return: {metrics['total_return']:.2f}%")
105        print(f"  Annualized Return: {metrics['annualized_return']:.2f}%")
106        print(f"  Annualized Volatility: {metrics['annualized_volatility']:.2f}%")
107
108        print(f"\nRisk-Adjusted Returns:")
109        print(f"  Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
110        print(f"  Sortino Ratio: {metrics['sortino_ratio']:.2f}")
111        print(f"  Calmar Ratio: {metrics['calmar_ratio']:.2f}")
112        print(f"  Maximum Drawdown: {metrics['max_drawdown']:.2f}%")
113
114        print(f"\nTrade Statistics:")
115        print(f"  Total Trades: {metrics['total_trades']}")
116        print(f"  Winning Trades: {metrics['winning_trades']}")
117        print(f"  Losing Trades: {metrics['losing_trades']}")
118        print(f"  Win Rate: {metrics['win_rate']:.2f}%")
119
120        print(f"\nPer-Trade Metrics:")
121        print(f"  Average Win: ${metrics['avg_win']:.2f}")
122        print(f"  Average Loss: ${metrics['avg_loss']:.2f}")
123        print(f"  Profit Factor: {metrics['profit_factor']:.2f}")
124        print(f"  Expectancy: ${metrics['expectancy']:.2f}")
125
126        print("="*60 + "\n")
127
128# Example usage
129result = backtest_with_risk_management(df, MovingAverageCrossover())
130equity = result.get_equity_curve()
131trades = result.get_trades_df()
132
133metrics = PerformanceAnalyzer.calculate_metrics(
134    equity, trades, result.initial_capital
135)
136PerformanceAnalyzer.print_metrics(metrics)

 1def compare_strategies(df: pd.DataFrame, strategies: List[TradingStrategy],
 2                      initial_capital: float = 100000):
 3    """
 4    Compare multiple strategies side-by-side
 5    """
 6    results = {}
 7
 8    for strategy in strategies:
 9        print(f"\nBacktesting {strategy.name}...")
10
11        # Run backtest
12        engine = backtest_with_risk_management(df, strategy, initial_capital)
13
14        # Get results
15        equity = engine.get_equity_curve()
16        trades = engine.get_trades_df()
17
18        # Calculate metrics
19        metrics = PerformanceAnalyzer.calculate_metrics(
20            equity, trades, initial_capital
21        )
22
23        results[strategy.name] = {
24            'engine': engine,
25            'metrics': metrics,
26        }
27
28    # Print comparison
29    print("\n" + "="*80)
30    print("STRATEGY COMPARISON")
31    print("="*80)
32
33    # Create comparison DataFrame
34    comparison = pd.DataFrame({
35        name: result['metrics']
36        for name, result in results.items()
37    }).T
38
39    print(comparison[[
40        'total_return',
41        'annualized_return',
42        'sharpe_ratio',
43        'max_drawdown',
44        'win_rate',
45        'total_trades'
46    ]])
47
48    return results
49
50# Compare strategies
51strategies = [
52    MovingAverageCrossover(fast_period=20, slow_period=50),
53    ADXTrendFollowing(fast_period=12, slow_period=26),
54    BollingerMeanReversion(bb_period=20, bb_std=2.0),
55    RSIVolumeMeanReversion(rsi_period=14),
56]
57
58comparison = compare_strategies(df, strategies)

 1# Exercise 1: Implement your own strategy
 2# Create a strategy combining multiple indicators
 3
 4# Exercise 2: Optimize parameters
 5# Find optimal parameters for MovingAverageCrossover using walk-forward analysis
 6
 7# Exercise 3: Multi-asset strategy
 8# Create a strategy that trades both BTC and ETH
 9
10# Exercise 4: Advanced risk management
11# Implement trailing stop loss and position scaling
12
13# Exercise 5: Strategy ensemble
14# Combine multiple strategies with dynamic weighting