Mean Reversion Strategy Research: SPY Dip-Buying Analysis

def detect_dip_signals(data: pd.DataFrame, rsi_threshold: int = 30,
                       drawdown_threshold: float = 0.08) -> pd.DataFrame:
    """
    Generate buy signals when extreme oversold conditions are met.
    """
    signals = data.copy()
 
    # Calculate RSI (14-period)
    delta = signals['Close'].diff()
    gain = delta.where(delta > 0, 0).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    signals['RSI'] = 100 - (100 / (1 + rs))
 
    # Calculate drawdown from 20-day high
    signals['rolling_high'] = signals['Close'].rolling(window=20).max()
    signals['drawdown'] = (signals['Close'] - signals['rolling_high']) / signals['rolling_high']
 
    # Generate signals
    signals['signal'] = 0
    signals.loc[signals['RSI'] < rsi_threshold, 'signal'] = 1
    signals.loc[signals['drawdown'] < -drawdown_threshold, 'signal'] = 1
 
    return signals

def trend_following_with_dip_entry(data: pd.DataFrame) -> pd.DataFrame:
    """
    Hybrid approach: trend-following with dip-timed entries.
    Entry: Buy on dip (RSI < 35) when above 200 MA
    Exit: Death cross (50 MA crosses below 200 MA)
    """
    signals = data.copy()
 
    signals['MA_50'] = signals['Close'].rolling(window=50).mean()
    signals['MA_200'] = signals['Close'].rolling(window=200).mean()
 
    delta = signals['Close'].diff()
    gain = delta.where(delta > 0, 0).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    signals['RSI'] = 100 - (100 / (1 + gain / loss))
 
    signals['uptrend'] = signals['MA_50'] > signals['MA_200']
    signals['entry'] = (signals['RSI'] < 35) & signals['uptrend']
    signals['exit'] = signals['MA_50'] < signals['MA_200']
 
    return signals

def multi_signal_exit(data: pd.DataFrame) -> pd.DataFrame:
    """Exit when multiple bearish signals align."""
    signals = data.copy()
 
    signals['MA_50'] = signals['Close'].rolling(50).mean()
    signals['momentum_20d'] = signals['Close'].pct_change(20) * 100
 
    exp12 = signals['Close'].ewm(span=12).mean()
    exp26 = signals['Close'].ewm(span=26).mean()
    signals['MACD'] = exp12 - exp26
    signals['MACD_signal'] = signals['MACD'].ewm(span=9).mean()
 
    delta = signals['Close'].diff()
    gain = delta.where(delta > 0, 0).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    signals['RSI'] = 100 - (100 / (1 + gain / loss))
 
    signals['bearish_count'] = (
        (signals['Close'] < signals['MA_50']).astype(int) +
        (signals['momentum_20d'] < -5).astype(int) +
        (signals['MACD'] < signals['MACD_signal']).astype(int) +
        (signals['RSI'] < 45).astype(int)
    )
 
    signals['exit'] = signals['bearish_count'] >= 3
    return signals

def calculate_expectancy(win_rate: float, avg_win: float, avg_loss: float) -> float:
    """
    Calculate expected return per trade.
    Example: 40% win rate with 6:1 ratio = 0.4 * 6 + 0.6 * (-1) = +1.8
    Example: 90% win rate with 1:10 ratio = 0.9 * 1 + 0.1 * (-10) = -0.1
    """
    return (win_rate * avg_win) + ((1 - win_rate) * avg_loss)

from dataclasses import dataclass
from typing import List, Dict
import pandas as pd
import numpy as np
 
@dataclass
class BacktestResult:
    """Results from a backtest run."""
    total_return: float
    annual_return: float
    max_drawdown: float
    sharpe_ratio: float
    win_rate: float
    profit_factor: float
    num_trades: int
    trades: List[Dict]
 
class Backtester:
    """Backtesting engine for long-only strategies."""
 
    def __init__(self, initial_capital: float = 10000, commission: float = 0.001):
        self.initial_capital = initial_capital
        self.commission = commission
 
    def run(self, data: pd.DataFrame, signals: pd.DataFrame) -> BacktestResult:
        """Execute backtest and return performance metrics."""
        cash = self.initial_capital
        shares = 0
        position = 0
        trades = []
        portfolio_values = []
        entry_price = 0
        entry_date = None
 
        for date, row in signals.iterrows():
            close = row['Close']
            signal = row.get('signal', 0)
 
            if signal == 1 and position == 0:
                shares = (cash * (1 - self.commission)) / close
                cash = 0
                position = 1
                entry_price = close
                entry_date = date
 
            elif signal == -1 and position == 1:
                proceeds = shares * close * (1 - self.commission)
                cash = proceeds
                pnl_pct = (close - entry_price) / entry_price * 100
                trades.append({
                    'entry_date': entry_date,
                    'exit_date': date,
                    'entry_price': entry_price,
                    'exit_price': close,
                    'pnl_pct': pnl_pct
                })
                shares = 0
                position = 0
 
            portfolio_values.append(cash + shares * close)
 
        portfolio = np.array(portfolio_values)
        total_return = (portfolio[-1] - self.initial_capital) / self.initial_capital * 100
 
        wins = [t for t in trades if t['pnl_pct'] > 0]
        losses = [t for t in trades if t['pnl_pct'] <= 0]
 
        win_rate = len(wins) / len(trades) * 100 if trades else 0
        avg_win = np.mean([t['pnl_pct'] for t in wins]) if wins else 0
        avg_loss = abs(np.mean([t['pnl_pct'] for t in losses])) if losses else 0.01
        profit_factor = avg_win / avg_loss if avg_loss > 0 else 999
 
        return BacktestResult(
            total_return=total_return,
            annual_return=total_return / 10,
            max_drawdown=self._calculate_max_drawdown(portfolio),
            sharpe_ratio=self._calculate_sharpe(portfolio),
            win_rate=win_rate,
            profit_factor=profit_factor,
            num_trades=len(trades),
            trades=trades
        )
 
    def _calculate_max_drawdown(self, portfolio: np.ndarray) -> float:
        peak = np.maximum.accumulate(portfolio)
        drawdown = (portfolio - peak) / peak * 100
        return abs(drawdown.min())
 
    def _calculate_sharpe(self, portfolio: np.ndarray, rf: float = 0.02) -> float:
        returns = np.diff(portfolio) / portfolio[:-1]
        excess = returns - rf / 252
        return np.sqrt(252) * excess.mean() / excess.std() if excess.std() > 0 else 0