Uncovering Trends and Reversals A Vortex Indicator Strategy with Mean Reversion Exits

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

plt.rcParams['figure.figsize'] = (12, 8)

class VortexReversionExitStrategy(bt.Strategy):
    """
    Vortex Strategy with Trailing Stop and Mean Reversion Exit.
    1. Vortex Indicator for entry signals.
    2. Long-term MA for macro trend filtering.
    3. Volatility Filter for stable markets.
    4. Mean Reversion Exit: Close position if price re-enters a defined "reversion zone" around MA.
    """
    params = (
        ('vortex_period', 30),
        ('long_term_ma_period', 30),
        ('atr_period', 14),  # Changed from 7 to 14 for more standard ATR
        ('atr_threshold', 0.05),  # Reduced from 0.05 for more realistic filtering
        ('atr_stop_multiplier', 3.),  # Reduced from 3.0 for tighter risk control
        ('reversion_zone_pct', 0.01),  # 1.5% reversion zone (slightly larger than 1%)
        ('min_vortex_diff', 0.01),  # Minimum difference between VI+ and VI- for signal
        ('min_distance_from_ma', 0.01),  # Minimum distance from MA required for entry (2%)
    )

    def __init__(self):
        self.order = None
        
        # Indicators
        self.vortex = bt.indicators.Vortex(self.data, period=self.p.vortex_period)
        self.long_term_ma = bt.indicators.SimpleMovingAverage(
            self.data.close, period=self.p.long_term_ma_period
        )
        self.atr = bt.indicators.AverageTrueRange(self.data, period=self.p.atr_period)
        self.vortex_cross = bt.indicators.CrossOver(
            self.vortex.lines.vi_plus,  
            self.vortex.lines.vi_minus
        )
        
        # Trade management variables
        self.stop_price = None
        self.highest_price_since_entry = None
        self.lowest_price_since_entry = None
        self.entry_price = None
        self.entry_trend_direction = None  # Track original trend direction at entry
        self.trade_count = 0

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

    def _reset_trade_variables(self):
        """Reset all trade tracking variables"""
        self.stop_price = None
        self.highest_price_since_entry = None
        self.lowest_price_since_entry = None
        self.entry_price = None
        self.entry_trend_direction = None

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return
            
        if order.status == order.Completed:
            if order.isbuy():
                if self.position.size > 0:  # Long entry completed
                    self.entry_price = order.executed.price
                    self.highest_price_since_entry = self.data.high[0]
                    self.stop_price = self.entry_price - (self.atr[0] * self.p.atr_stop_multiplier)
                    self.entry_trend_direction = 'up'  # Remember we entered on uptrend
                    self.log(f'LONG EXECUTED - Entry: {self.entry_price:.2f}, Initial Stop: {self.stop_price:.2f}')
                else:  # Closing short position
                    self.log(f'SHORT CLOSED - Exit: {order.executed.price:.2f}')
                    self._reset_trade_variables()
                    
            elif order.issell():
                if self.position.size < 0:  # Short entry completed
                    self.entry_price = order.executed.price
                    self.lowest_price_since_entry = self.data.low[0]
                    self.stop_price = self.entry_price + (self.atr[0] * self.p.atr_stop_multiplier)
                    self.entry_trend_direction = 'down'  # Remember we entered on downtrend
                    self.log(f'SHORT EXECUTED - Entry: {self.entry_price:.2f}, Initial Stop: {self.stop_price:.2f}')
                else:  # Closing long position
                    self.log(f'LONG CLOSED - Exit: {order.executed.price:.2f}')
                    self._reset_trade_variables()
        elif order.status in [order.Canceled, order.Rejected]:
            self.log(f'ORDER CANCELED/REJECTED - Status: {order.getstatusname()}')
            
        self.order = None

    def notify_trade(self, trade):
        if trade.isclosed:
            profit_pct = (trade.pnl / abs(trade.value)) * 100 if trade.value != 0 else 0
            self.log(f'TRADE CLOSED - PnL: ${trade.pnl:.2f} ({profit_pct:.2f}%)')

    def next(self):
        # Skip if we have pending orders
        if self.order:
            return

        # Ensure enough data for all indicators
        min_periods = max(self.p.vortex_period, self.p.long_term_ma_period, self.p.atr_period)
        if len(self) < min_periods:
            return

        current_price = self.data.close[0]
        ma_price = self.long_term_ma[0]
        
        # Market condition filters
        atr_ratio = self.atr[0] / current_price
        is_stable_vol = atr_ratio < self.p.atr_threshold
        
        # Distance from MA calculations
        distance_from_ma_pct = abs(current_price - ma_price) / ma_price
        
        # Trend conditions
        is_macro_uptrend = current_price > ma_price
        is_macro_downtrend = current_price < ma_price
        
        # Ensure sufficient distance from MA for entry (avoid entering near reversion zone)
        sufficient_distance_for_long = (current_price > ma_price and 
                                        distance_from_ma_pct > self.p.min_distance_from_ma)
        sufficient_distance_for_short = (current_price < ma_price and 
                                         distance_from_ma_pct > self.p.min_distance_from_ma)
        
        # Vortex signals with strength filter
        vi_plus = self.vortex.lines.vi_plus[0]
        vi_minus = self.vortex.lines.vi_minus[0]
        vortex_diff = abs(vi_plus - vi_minus)
        
        is_buy_signal = (self.vortex_cross[0] > 0 and 
                         vi_plus > vi_minus and 
                         vortex_diff > self.p.min_vortex_diff)
        is_sell_signal = (self.vortex_cross[0] < 0 and 
                          vi_minus > vi_plus and 
                          vortex_diff > self.p.min_vortex_diff)

        # Position management
        if self.position:
            position_size = self.position.size
            
            # Check for mean reversion exit first (before trailing stop updates)
            in_reversion_zone = distance_from_ma_pct < self.p.reversion_zone_pct
            
            if in_reversion_zone:
                # Long position: exit if price falls back towards or below MA
                if (position_size > 0 and self.entry_trend_direction == 'up' and 
                    current_price <= ma_price * (1 + self.p.reversion_zone_pct)):
                    
                    self.order = self.close()
                    self.log(f'MEAN REVERSION EXIT - Long closed. Price: {current_price:.2f}, MA: {ma_price:.2f}, Distance: {distance_from_ma_pct:.3f}')
                    return # Exit next() immediately after closing
                    
                # Short position: exit if price rises back towards or above MA
                elif (position_size < 0 and self.entry_trend_direction == 'down' and 
                      current_price >= ma_price * (1 - self.p.reversion_zone_pct)):
                    
                    self.order = self.close()
                    self.log(f'MEAN REVERSION EXIT - Short closed. Price: {current_price:.2f}, MA: {ma_price:.2f}, Distance: {distance_from_ma_pct:.3f}')
                    return # Exit next() immediately after closing
            
            # Manual ATR Trailing Stop Logic (primary exit mechanism)
            if position_size > 0:  # Long position
                # Update highest price
                if self.data.high[0] > self.highest_price_since_entry:
                    self.highest_price_since_entry = self.data.high[0]
                
                # Calculate new trailing stop
                new_stop = self.highest_price_since_entry - (self.atr[0] * self.p.atr_stop_multiplier)
                
                # Only move stop up (in favorable direction)
                if new_stop > self.stop_price:
                    self.stop_price = new_stop
                    self.log(f'TRAILING STOP UPDATED - New stop: {self.stop_price:.2f}')
                
                # Check for stop loss hit
                if current_price <= self.stop_price:
                    self.order = self.close()
                    self.log(f'STOP LOSS HIT - Long closed at {current_price:.2f}')
                    return # Exit next() immediately after closing
                    
            elif position_size < 0:  # Short position
                # Update lowest price
                if self.data.low[0] < self.lowest_price_since_entry:
                    self.lowest_price_since_entry = self.data.low[0]
                
                # Calculate new trailing stop
                new_stop = self.lowest_price_since_entry + (self.atr[0] * self.p.atr_stop_multiplier)
                
                # Only move stop down (in favorable direction)
                if new_stop < self.stop_price:
                    self.stop_price = new_stop
                    self.log(f'TRAILING STOP UPDATED - New stop: {self.stop_price:.2f}')
                
                # Check for stop loss hit
                if current_price >= self.stop_price:
                    self.order = self.close()
                    self.log(f'STOP LOSS HIT - Short closed at {current_price:.2f}')
                    return

        # Entry logic - only when no position
        else:
            # Long entry conditions
            if (is_stable_vol and sufficient_distance_for_long and 
                is_buy_signal and is_macro_uptrend):
                
                self.order = self.buy()
                self.trade_count += 1
                self.log(f'LONG SIGNAL - Price: {current_price:.2f}, MA: {ma_price:.2f}, Distance: {distance_from_ma_pct:.3f}, VI+: {vi_plus:.3f}, VI-: {vi_minus:.3f}')
            
            # Short entry conditions
            elif (is_stable_vol and sufficient_distance_for_short and 
                  is_sell_signal and is_macro_downtrend):
                
                self.order = self.sell()
                self.trade_count += 1
                self.log(f'SHORT SIGNAL - Price: {current_price:.2f}, MA: {ma_price:.2f}, Distance: {distance_from_ma_pct:.3f}, VI+: {vi_plus:.3f}, VI-: {vi_minus:.3f}')

    def stop(self):
        print(f'\n=== VORTEX MEAN REVERSION EXIT STRATEGY RESULTS ===')
        print(f'Total Trades: {self.trade_count}')
        print(f'Strategy: Vortex with trailing stops and mean reversion exits')
        print(f'Parameters:')
        print(f'  - Vortex Period: {self.p.vortex_period}')
        print(f'  - Long MA Period: {self.p.long_term_ma_period}')
        print(f'  - ATR Period: {self.p.atr_period}')
        print(f'  - ATR Threshold: {self.p.atr_threshold:.3f}')
        print(f'  - ATR Stop Multiplier: {self.p.atr_stop_multiplier}')
        print(f'  - Reversion Zone: {self.p.reversion_zone_pct:.3f} ({self.p.reversion_zone_pct*100:.1f}%)')
        print(f'  - Min Distance from MA: {self.p.min_distance_from_ma:.3f} ({self.p.min_distance_from_ma*100:.1f}%)')
        print(f'  - Min Vortex Diff: {self.p.min_vortex_diff}')