Article

Trend Reversals and Momentum Confirmation Rolling Backtest of a Parabolic SAR Strategy with RSI Filter

This article introduces a trend-following strategy, ParabolicSARStrategy, that utilizes the Parabolic Stop and Reverse (SAR) indicator for identifying trend direction and potential reversals. To enhance the robustness of its signals and avoid whipsaws, the strategy incorporates the Relative Strength Index (RSI) as a momentum filter and implements a fixed percentage stop-loss for disciplined risk management. The strategy’s effectiveness is then evaluated using a comprehensive rolling backtesting framework.

1. The Parabolic SAR Strategy Concept

The ParabolicSARStrategy is designed to capture trends by following price action with the Parabolic SAR, which provides a trailing stop-loss-like mechanism. It aims to enter positions when a new trend is indicated by SAR and confirm this with momentum.

Key Components:

Parabolic SAR (PSAR): This indicator trails price action and plots dots below the price during an uptrend and above the price during a downtrend. A key signal is generated when the price crosses the SAR:
- Bullish Reversal: Price crosses above SAR, indicating a potential uptrend initiation.
- Bearish Reversal: Price crosses below SAR, indicating a potential downtrend initiation.
- The af (acceleration factor) and afmax (maximum acceleration factor) parameters control the sensitivity and speed of the SAR.
Relative Strength Index (RSI): A momentum oscillator that measures the speed and change of price movements.
- Used here as a filter:
  - For long entries, the strategy requires RSI to be below rsi_overbought (e.g., 70) to avoid entering positions that are already overextended.
  - For short entries, the strategy requires RSI to be above rsi_oversold (e.g., 30) to avoid entering positions that are already oversold.
Fixed Percentage Stop Loss: A simple, yet crucial, risk management tool. Immediately upon entering a trade, a stop-loss order is placed at a fixed percentage (stop_loss_pct) away from the entry price. This limits potential losses if the trend quickly reverses or a false signal occurs.

Entry Logic:

Long Entry:
- SAR signal turns bullish (price crosses above SAR).
- AND RSI is below the rsi_overbought threshold.
Short Entry:
- SAR signal turns bearish (price crosses below SAR).
- AND RSI is above the rsi_oversold threshold.

Exit Logic:

SAR Reversal: When the Parabolic SAR indicator flips, it signals a potential trend reversal, prompting an exit from the current position.
Fixed Stop Loss: If the price hits the predefined stop-loss level, the position is automatically closed.

2. The `ParabolicSARStrategy` Implementation

Here’s the core backtrader strategy code:

import backtrader as bt
import yfinance as yf
import pandas as pd
import matplotlib.pyplot as plt

class ParabolicSARStrategy(bt.Strategy):
    params = (
        ('af', 0.02),            # Acceleration factor for PSAR
        ('afmax', 0.1),          # Maximum acceleration factor for PSAR
        ('rsi_period', 14),      # RSI period for momentum
        ('rsi_overbought', 70),  # RSI overbought level for filtering long entries
        ('rsi_oversold', 30),    # RSI oversold level for filtering short entries
        ('stop_loss_pct', 0.02), # Fixed percentage stop loss (e.g., 0.02 for 2%)
    )
    
    def __init__(self):
        # Parabolic SAR indicator
        self.psar = bt.indicators.ParabolicSAR(
            af=self.params.af, 
            afmax=self.params.afmax
        )
        
        # Momentum confirmation with RSI
        self.rsi = bt.indicators.RSI(period=self.params.rsi_period)
        
        # Define signals based on Price vs. SAR (for readability)
        # Price above SAR (self.data.close > self.psar) means SAR is below price, bullish setup
        # Price below SAR (self.data.close < self.psar) means SAR is above price, bearish setup
        self.sar_long_position = self.data.close > self.psar 
        self.sar_short_position = self.data.close < self.psar
        
        # Detect actual SAR directional changes (crossovers)
        # self.sar_signal > 0 when SAR turns bullish (price crosses above SAR)
        # self.sar_signal < 0 when SAR turns bearish (price crosses below SAR)
        self.sar_signal = bt.indicators.CrossOver(self.sar_long_position, 0.5) 
        
        # Track orders to prevent multiple simultaneous orders
        self.order = None
        self.stop_order = None # For the fixed stop-loss order

    def log(self, txt, dt=None):
        ''' Logging function for strategy actions '''
        dt = dt or self.datas[0].datetime.date(0)
        print(f'{dt.isoformat()} - {txt}')

    def notify_order(self, order):
        # Ignore submitted/accepted orders, wait for completion or failure
        if order.status in [order.Submitted, order.Accepted]:
            return
        
        # Handle completed orders
        if order.status in [order.Completed]:
            if order.isbuy(): # If a buy order (entry or cover short) completed
                self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Size: {order.executed.size:.2f}')
                # If we are now in a long position, set a fixed stop-loss
                if self.position.size > 0:
                    stop_price = order.executed.price * (1 - self.params.stop_loss_pct)
                    # Place a Stop Sell order at the calculated stop_price
                    self.stop_order = self.sell(exectype=bt.Order.Stop, price=stop_price, size=self.position.size)
                    self.log(f'Long Stop Loss set at {stop_price:.2f}')
            
            elif order.issell(): # If a sell order (entry or exit long) completed
                self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Size: {order.executed.size:.2f}')
                # If we are now in a short position, set a fixed stop-loss
                if self.position.size < 0:
                    stop_price = order.executed.price * (1 + self.params.stop_loss_pct)
                    # Place a Stop Buy order at the calculated stop_price
                    self.stop_order = self.buy(exectype=bt.Order.Stop, price=stop_price, size=abs(self.position.size))
                    self.log(f'Short Stop Loss set at {stop_price:.2f}')
            
            # Clear the main order reference after any order completion
            self.order = None 
            # If the completed order was our stop-loss order, clear its reference
            if self.stop_order and order.ref == self.stop_order.ref:
                self.stop_order = None
        
        # Handle canceled/rejected orders
        elif order.status in [order.Canceled, order.Rejected]:
            self.log(f'Order Canceled/Rejected: Status {order.getstatusname()}')
            if self.order and order.ref == self.order.ref: # If entry order failed
                self.order = None
            if self.stop_order and order.ref == self.stop_order.ref: # If stop order failed
                self.stop_order = None # IMPORTANT: If stop fails, the position is unprotected!

    def notify_trade(self, trade):
        ''' Reports profit/loss when a trade is closed '''
        if not trade.isclosed:
            return
        self.log(f'TRADE P/L: GROSS {trade.pnl:.2f}, NET {trade.pnlcomm:.2f}')

    def next(self):
        # Prevent new entry/exit orders if one is already pending
        if self.order is not None:
            return
        
        # Ensure indicators have warmed up enough. PSAR needs a few bars, RSI needs its period.
        min_warmup = max(self.params.afmax / self.params.af * 10, self.params.rsi_period) # Rough estimate for PSAR warmup
        if len(self.data) < min_warmup:
             return # Not enough data for reliable indicator calculations

        # --- Trading Logic based on SAR signals and RSI confirmation ---
        
        # If SAR turns bullish (price crosses above SAR)
        if self.sar_signal[0] > 0: # This indicates a bullish crossover
            # Confirm with RSI not overbought (i.e., has room to move up)
            if self.rsi[0] < self.params.rsi_overbought:
                if self.position.size < 0: # If currently short, close short position
                    self.log(f'SAR BULLISH signal. Close SHORT. Price: {self.data.close[0]:.2f}')
                    if self.stop_order is not None: self.cancel(self.stop_order) # Cancel old stop order
                    self.order = self.close()
                elif not self.position: # If flat, go long
                    self.log(f'SAR BULLISH signal. BUY. Price: {self.data.close[0]:.2f}')
                    self.order = self.buy() # Stop loss will be set in notify_order
                    
        # If SAR turns bearish (price crosses below SAR)
        elif self.sar_signal[0] < 0: # This indicates a bearish crossover
            # Confirm with RSI not oversold (i.e., has room to move down)
            if self.rsi[0] > self.params.rsi_oversold:
                if self.position.size > 0: # If currently long, close long position
                    self.log(f'SAR BEARISH signal. Close LONG. Price: {self.data.close[0]:.2f}')
                    if self.stop_order is not None: self.cancel(self.stop_order) # Cancel old stop order
                    self.order = self.close()
                elif not self.position: # If flat, go short
                    self.log(f'SAR BEARISH signal. SELL (Short). Price: {self.data.close[0]:.2f}')
                    self.order = self.sell() # Stop loss will be set in notify_order

    def stop(self):
        self.log(f'Final Portfolio Value: {self.broker.getvalue():.2f}')

Explanation of ParabolicSARStrategy:

params: Defines configurable parameters for the PSAR indicator (af, afmax), RSI (rsi_period, rsi_overbought, rsi_oversold), and the stop_loss_pct.
__init__(self):
- Initializes the bt.indicators.ParabolicSAR with the specified acceleration factors.
- Initializes the bt.indicators.RSI for momentum filtering.
- Defines self.sar_long_position and self.sar_short_position as boolean lines indicating if the price is currently above/below SAR, respectively.
- Uses bt.indicators.CrossOver on self.sar_long_position to detect actual flips of the SAR, which are the primary signals for entry/exit. A positive crossover (>0) means SAR turned bullish, a negative one (<0) means SAR turned bearish.
- Initializes self.order and self.stop_order for managing trades.
log(self, txt, dt=None): A simple logging utility for strategy actions.
notify_order(self, order): This critical method handles order execution and placement of the fixed stop-loss.
- When an entry order (buy or sell/short) Completed, it calculates the stop_price based on stop_loss_pct and immediately places a bt.Order.Stop order to protect the position. This is a fixed stop, not a trailing stop (the strategy uses PSAR itself as a kind of trailing exit).
- It clears self.order after completion to allow new orders.
- It also clears self.stop_order when the stop order itself is completed (meaning the position was stopped out) or if the stop order is cancelled/rejected.
notify_trade(self, trade): Logs the profit/loss when a trade is fully closed.
next(self): This is the core logic executed on each new bar.
- Order Pending Check: Prevents new entry/exit orders if one is already pending.
- Indicator Warm-up: Ensures sufficient historical data for PSAR and RSI to be calculated reliably.
- Trading Logic:
  - SAR Bullish Signal (self.sar_signal[0] > 0): If SAR indicates an uptrend (price crosses above SAR), and self.rsi[0] is below rsi_overbought (i.e., not overextended), the strategy takes action:
    - If currently short, it self.close()s the short position.
    - If currently flat, it self.buy()s to go long.
  - SAR Bearish Signal (self.sar_signal[0] < 0): If SAR indicates a downtrend (price crosses below SAR), and self.rsi[0] is above rsi_oversold (i.e., not oversold), the strategy takes action:
    - If currently long, it self.close()s the long position.
    - If currently flat, it self.sell()s to go short.
  - For any entry, the fixed stop loss will be placed in notify_order. When self.close() is called for an exit, it automatically cancels any existing stop order associated with that position.
stop(self): Logs the final portfolio value at the end of the backtest.

3. Backtesting and Analysis

The provided script includes two primary ways to test this strategy:

Single Backtest (for initial testing and plotting):

if __name__=='__main__':
    # Download data and run backtest (Example for BTC-USD)
    data = yf.download('BTC-USD', '2021-01-01', '2024-01-01', auto_adjust=False) # Use auto_adjust=False as per preference
    data.columns = data.columns.droplevel(1) if isinstance(data.columns, pd.MultiIndex) else data.columns # Droplevel if MultiIndex
    data_feed = bt.feeds.PandasData(dataname=data)

    cerebro = bt.Cerebro()
    cerebro.addstrategy(ParabolicSARStrategy) # Add your strategy
    cerebro.adddata(data_feed)
    cerebro.addsizer(bt.sizers.PercentSizer, percents=95)
    cerebro.broker.setcash(100000)
    cerebro.broker.setcommission(commission=0.001)

    print(f'Start: ${cerebro.broker.getvalue():,.2f}')
    results = cerebro.run()
    print(f'End: ${cerebro.broker.getvalue():,.2f}')
    print(f'Return: {((cerebro.broker.getvalue() / 100000) - 1) * 100:.2f}%')

    # Fix matplotlib plotting issues (already in your code)
    plt.rcParams['figure.max_open_warning'] = 0
    plt.rcParams['agg.path.chunksize'] = 10000

    try:
        cerebro.plot(iplot=False, style='candlestick', volume=True) # Added candlestick and volume
        plt.show()
    except Exception as e:
        print(f"Plotting error: {e}")
        print("Strategy completed successfully - plotting skipped")

This block allows you to quickly run the strategy over a single period and visualize its trades using cerebro.plot().

Rolling Backtest (for robustness evaluation): This method is crucial for understanding a strategy’s consistency across various market conditions, preventing curve-fitting to a single historical period.

import dateutil.relativedelta as rd # Added import
import seaborn as sns # Added import
from datetime import datetime # Added import for current date

# Define the strategy for the rolling backtest
strategy = ParabolicSARStrategy

def run_rolling_backtest(
    ticker="ETH-USD",
    start="2018-01-01",
    # Updated end date to the current date for a more live test
    end=datetime.now().date(), # Get current date for the end of the backtest
    window_months=3,
    strategy_params=None
):
    strategy_params = strategy_params or {}
    all_results = []
    start_dt = pd.to_datetime(start)
    end_dt = pd.to_datetime(end)
    current_start = start_dt

    while True:
        current_end = current_start + rd.relativedelta(months=window_months)
        if current_end > end_dt:
            current_end = end_dt 
            if current_start >= current_end:
                break

        print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")

        # Data download using yfinance, respecting user's preference
        # Using the saved preference: yfinance download with auto_adjust=False and droplevel(axis=1, level=1)
        data = yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)

        # Apply droplevel if data is a MultiIndex, as per user's preference
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, axis=1)

        # Check for sufficient data after droplevel for strategy warm-up
        # PSAR needs a few bars, RSI needs its period.
        # Longest period is RSI.
        rsi_period = strategy_params.get('rsi_period', ParabolicSARStrategy.params.rsi_period)
        min_bars_needed = max(rsi_period, 50) # Rough estimate for PSAR warmup to be safe, or just RSI period

        if data.empty or len(data) < min_bars_needed:
            print(f"Not enough data for period {current_start.date()} to {current_end.date()} (requires at least {min_bars_needed} bars). Skipping.")
            if current_end == end_dt:
                break
            current_start = current_end
            continue

        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(strategy, **strategy_params)
        cerebro.adddata(feed)
        cerebro.broker.setcash(100000)
        cerebro.broker.setcommission(commission=0.001)
        cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

        start_val = cerebro.broker.getvalue()
        cerebro.run()
        final_val = cerebro.broker.getvalue()
        ret = (final_val - start_val) / start_val * 100

        all_results.append({
            'start': current_start.date(),
            'end': current_end.date(),
            'return_pct': ret,
            'final_value': final_val,
        })

        print(f"Return: {ret:.2f}% | Final Value: {final_val:.2f}")

        if current_end == end_dt:
            break
        current_start = current_end

    return pd.DataFrame(all_results)


def report_stats(df):
    returns = df['return_pct']
    stats = {
        'Mean Return %': np.mean(returns),
        'Median Return %': np.median(returns),
        'Std Dev %': np.std(returns),
        'Min Return %': np.min(returns),
        'Max Return %': np.max(returns),
        'Sharpe Ratio': np.mean(returns) / np.std(returns) if np.std(returns) > 0 else np.nan
    }
    print("\n=== ROLLING BACKTEST STATISTICS ===")
    for k, v in stats.items():
        print(f"{k}: {v:.2f}")
    return stats

def plot_four_charts(df, rolling_sharpe_window=4):
    """
    Generates four analytical plots for rolling backtest results.
    """
    fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(12, 8)) # Adjusted figsize for clarity

    periods = list(range(len(df)))
    returns = df['return_pct']

    # 1. Period Returns (Top Left)
    colors = ['green' if r >= 0 else 'red' for r in returns]
    ax1.bar(periods, returns, color=colors, alpha=0.7)
    ax1.set_title('Period Returns', fontsize=14, fontweight='bold')
    ax1.set_xlabel('Period')
    ax1.set_ylabel('Return %')
    ax1.axhline(y=0, color='black', linestyle='-', alpha=0.3)
    ax1.grid(True, alpha=0.3)

    # 2. Cumulative Returns (Top Right)
    cumulative_returns = (1 + returns / 100).cumprod() * 100 - 100
    ax2.plot(periods, cumulative_returns, marker='o', linewidth=2, markersize=4, color='blue')
    ax2.set_title('Cumulative Returns', fontsize=14, fontweight='bold')
    ax2.set_xlabel('Period')
    ax2.set_ylabel('Cumulative Return %')
    ax2.grid(True, alpha=0.3)

    # 3. Rolling Sharpe Ratio (Bottom Left)
    rolling_sharpe = returns.rolling(window=rolling_sharpe_window).apply(
        lambda x: x.mean() / x.std() if x.std() > 0 else np.nan, raw=False
    )
    valid_mask = ~rolling_sharpe.isna()
    valid_periods = [i for i, valid in enumerate(valid_mask) if valid]
    valid_sharpe = rolling_sharpe[valid_mask]

    ax3.plot(valid_periods, valid_sharpe, marker='o', linewidth=2, markersize=4, color='orange')
    ax3.axhline(y=0, color='red', linestyle='--', alpha=0.5)
    ax3.set_title(f'Rolling Sharpe Ratio ({rolling_sharpe_window}-period)', fontsize=14, fontweight='bold')
    ax3.set_xlabel('Period')
    ax3.set_ylabel('Sharpe Ratio')
    ax3.grid(True, alpha=0.3)

    # 4. Return Distribution (Bottom Right)
    bins = min(15, max(5, len(returns)//2))
    ax4.hist(returns, bins=bins, alpha=0.7, color='steelblue', edgecolor='black')
    mean_return = returns.mean()
    ax4.axvline(mean_return, color='red', linestyle='--', linewidth=2, 
                label=f'Mean: {mean_return:.2f}%')
    ax4.set_title('Return Distribution', fontsize=14, fontweight='bold')
    ax4.set_xlabel('Return %')
    ax4.set_ylabel('Frequency')
    ax4.legend()
    ax4.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

if __name__ == '__main__':
    # The end date is set to the current date for a more "live" simulation
    current_date = datetime.now().date() 

    df = run_rolling_backtest(
        ticker="ETH-USD", # Default ticker for article's example
        start="2018-01-01",
        end=current_date, # Use the current date
        window_months=3,

    )

    print("\n=== ROLLING BACKTEST RESULTS ===")
    print(df)

    stats = report_stats(df)
    plot_four_charts(df)

4. Conclusion

The ParabolicSARStrategy provides a straightforward yet effective approach to trend following by utilizing the dynamic nature of the Parabolic SAR indicator. The integration of RSI acts as a crucial momentum filter, helping to avoid trades in overextended conditions and potentially reducing whipsaws. The fixed percentage stop-loss is an essential component for disciplined risk management, ensuring that potential losses are controlled. The rigorous rolling backtesting framework is invaluable for assessing the strategy’s consistency and resilience across diverse market environments, offering a more reliable evaluation of its long-term viability. Further fine-tuning of the SAR acceleration factors, RSI thresholds, and stop-loss percentage could lead to optimized performance for specific assets or market conditions.

1. The Parabolic SAR Strategy Concept

2. The ParabolicSARStrategy Implementation

3. Backtesting and Analysis

4. Conclusion

2. The `ParabolicSARStrategy` Implementation