The VolatilityRegimeStrategy
is an adaptive
trading system that changes its trading approach based on
current market volatility, using different sub-strategies for trending
and ranging environments.
Market behavior is not constant; it frequently shifts between periods
of high volatility (often associated with strong trends
or uncertainty) and low volatility (often associated
with sideways, ranging movement). A truly robust trading strategy should
adapt to these changing market conditions. The
VolatilityRegimeStrategy
is designed to do exactly this,
dynamically switching between a trend-following
sub-strategy in medium volatility and a mean-reversion
sub-strategy in low volatility, while exiting all positions
during periods of high volatility.
This strategy classifies the market into three regimes based on the percentile rank of historical volatility:
All positions are managed with a dynamic ATR-based trailing stop.
PercentRank
IndicatorThe core of regime detection relies on the custom
PercentRank
indicator, which determines how the current
volatility compares to historical volatility.
import backtrader as bt
import numpy as np
# The custom PercentRank indicator is still required
class PercentRank(bt.Indicator):
= ('pctrank',); params = (('period', 100),)
lines def __init__(self): self.addminperiod(self.p.period)
def next(self):
= np.asarray(self.data.get(size=self.p.period))
data_window self.lines.pctrank[0] = (np.sum(data_window < self.data[0]) / self.p.period) * 100.0
The PercentRank
indicator calculates the percentile rank
of the current data point within a specified historical
period
. For example, if the period
is 100, it
tells you what percentage of the last 100 values are less than the
current value. This is used to gauge how “high” or “low” the current
volatility is relative to its recent history.
VolatilityRegimeStrategy
Implementationimport backtrader as bt
import numpy as np
# The custom PercentRank indicator is still required
class PercentRank(bt.Indicator):
= ('pctrank',); params = (('period', 100),)
lines def __init__(self): self.addminperiod(self.p.period)
def next(self):
= np.asarray(self.data.get(size=self.p.period))
data_window self.lines.pctrank[0] = (np.sum(data_window < self.data[0]) / self.p.period) * 100.0
class VolatilityRegimeStrategy(bt.Strategy):
"""
An enhanced adaptive system that uses more robust sub-strategies for each
volatility regime: Elder Impulse for trends and Volatility Reversal for ranges.
"""
= (
params # Regime Detection
'vol_period', 7), ('vol_lookback', 30),
('high_vol_thresh', 75.0), ('low_vol_thresh', 25.0),
(# Upgraded Trend-Following Sub-Strategy (Elder Impulse)
'impulse_ema', 7), ('macd_fast', 7), ('macd_slow', 30), ('macd_signal', 7),
(# Upgraded Mean-Reversion Sub-Strategy (Volatility Reversal)
'bb_period', 7), ('bb_devfactor', 2.0),
('rsi_period', 14), ('rsi_ob', 70), ('rsi_os', 30),
(# Risk Management
'atr_period', 7), ('atr_stop_multiplier', 3.),
(
)
def __init__(self):
self.order = None
# --- Regime Indicators ---
self.hist_vol = bt.indicators.StandardDeviation(self.data.close, period=self.p.vol_period)
self.vol_rank = PercentRank(self.hist_vol, period=self.p.vol_lookback)
# --- Trend Sub-Strategy Indicators ---
self.impulse_ema = bt.indicators.ExponentialMovingAverage(self.data, period=self.p.impulse_ema)
self.macd_histo = bt.indicators.MACDHistogram(self.data, period_me1=self.p.macd_fast, period_me2=self.p.macd_slow, period_signal=self.p.macd_signal)
# --- Mean-Reversion Sub-Strategy Indicators ---
self.bband = bt.indicators.BollingerBands(self.data, period=self.p.bb_period, devfactor=self.p.bb_devfactor)
self.rsi = bt.indicators.RSI(self.data, period=self.p.rsi_period)
self.atr = bt.indicators.AverageTrueRange(self.data, period=self.p.atr_period)
# --- State and Stop ---
self.bullish_climax_armed = False; self.bearish_climax_armed = False
self.stop_price = None; self.highest_price_since_entry = None; self.lowest_price_since_entry = None
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]: return
if order.status in [order.Completed]:
if self.position and self.stop_price is None:
if order.isbuy(): self.highest_price_since_entry = self.data.high[0]; self.stop_price = self.highest_price_since_entry - (self.atr[0] * self.p.atr_stop_multiplier)
elif order.issell(): self.lowest_price_since_entry = self.data.low[0]; self.stop_price = self.lowest_price_since_entry + (self.atr[0] * self.p.atr_stop_multiplier)
elif not self.position: self.stop_price = None; self.highest_price_since_entry = None; self.lowest_price_since_entry = None
self.order = None
def next(self):
if self.order: return
= self.vol_rank.pctrank[0]
current_vol_rank
# --- HIGH-VOLATILITY / RISK-OFF REGIME ---
if current_vol_rank > self.p.high_vol_thresh:
if self.position: self.order = self.close() # Close any open positions
# Reset any armed signals from other regimes
self.bullish_climax_armed = False; self.bearish_climax_armed = False
return # Do not trade in high volatility
# --- LOW-VOLATILITY / MEAN-REVERSION REGIME ---
if current_vol_rank < self.p.low_vol_thresh:
if not self.position:
# Trigger entry only after a climax has been armed and price reverses from BBand extreme
if self.bullish_climax_armed:
self.bullish_climax_armed = False # Disarm
if self.data.close[0] > self.bband.bot[0]: # Price moves back inside BB from below
self.order = self.buy()
return # Exit after attempting an entry
if self.bearish_climax_armed:
self.bearish_climax_armed = False # Disarm
if self.data.close[0] < self.bband.top[0]: # Price moves back inside BB from above
self.order = self.sell()
return # Exit after attempting an entry
# Arm a climax signal if price breaks BB and RSI is overbought/oversold
= self.rsi[0] < self.p.rsi_os
is_oversold = self.rsi[0] > self.p.rsi_ob
is_overbought if self.data.close[0] < self.bband.bot[0] and is_oversold:
self.bullish_climax_armed = True
elif self.data.close[0] > self.bband.top[0] and is_overbought:
self.bearish_climax_armed = True
# --- MEDIUM-VOLATILITY / TREND-FOLLOWING REGIME ---
else: # This 'else' covers all cases where vol_rank is NOT high AND NOT low, thus medium
if not self.position:
# Elder Impulse System logic for trend-following
= self.impulse_ema[0] > self.impulse_ema[-1]
ema_is_rising = self.macd_histo[0] > self.macd_histo[-1]
histo_is_rising
if ema_is_rising and histo_is_rising: # Strong bullish impulse
self.order = self.buy()
elif not ema_is_rising and not histo_is_rising: # Strong bearish impulse
self.order = self.sell()
# --- POSITION MANAGEMENT (ATR Trailing Stop) ---
if self.position:
if self.position.size > 0: # Long position
self.highest_price_since_entry = max(self.highest_price_since_entry, self.data.high[0])
= self.highest_price_since_entry - (self.atr[0] * self.p.atr_stop_multiplier)
new_stop self.stop_price = max(self.stop_price, new_stop) # Only move stop up
if self.data.close[0] < self.stop_price:
self.order = self.close()
elif self.position.size < 0: # Short position
self.lowest_price_since_entry = min(self.lowest_price_since_entry, self.data.low[0])
= self.lowest_price_since_entry + (self.atr[0] * self.p.atr_stop_multiplier)
new_stop self.stop_price = min(self.stop_price, new_stop) # Only move stop down
if self.data.close[0] > self.stop_price:
self.order = self.close()
params
)The strategy’s behavior is highly configurable:
vol_period
: Period for calculating historical
volatility (Standard Deviation).vol_lookback
: Period for calculating the percentile
rank of volatility.high_vol_thresh
: Percentile threshold for identifying a
high-volatility regime.low_vol_thresh
: Percentile threshold for identifying a
low-volatility regime.impulse_ema
: EMA period for the Elder Impulse
System.macd_fast
, macd_slow
,
macd_signal
: Parameters for the MACD Histogram
component.bb_period
, bb_devfactor
: Period and
standard deviation factor for Bollinger Bands.rsi_period
, rsi_ob
, rsi_os
:
Period, overbought, and oversold levels for the RSI indicator.atr_period
: Period for the Average True Range.atr_stop_multiplier
: Multiplier for ATR to set the
trailing stop distance.__init__
)In the __init__
method, all necessary indicators for
regime detection, trend-following, and mean-reversion sub-strategies are
instantiated:
self.hist_vol
(Standard Deviation of close price) and self.vol_rank
(our
custom PercentRank
of hist_vol
).self.impulse_ema
(Exponential Moving Average) and
self.macd_histo
(MACD Histogram) for the Elder Impulse
System.self.bband
(Bollinger Bands) and self.rsi
(Relative Strength Index).self.atr
: Average True Range for the trailing
stop.bullish_climax_armed
,
bearish_climax_armed
(flags for mean-reversion entry), and
stop_price
, highest_price_since_entry
,
lowest_price_since_entry
(for the ATR trailing stop).notify_order
)This method is used to manage the dynamic ATR trailing stop.
stop_price
based on the entry price and the
current ATR. It also records the highest_price_since_entry
or lowest_price_since_entry
.None
, preparing for the next trade.next
)The next
method is the heart of the adaptive strategy,
executing its logic based on the current volatility regime:
Regime Detection: current_vol_rank
is calculated.
High-Volatility / Risk-Off Regime:
if current_vol_rank > self.p.high_vol_thresh:
: If
volatility is high, any existing position is immediately closed
(self.order = self.close()
).climax_armed
signals from the mean-reversion
sub-strategy are reset, and the method returns
, preventing
any trading in this regime.Low-Volatility / Mean-Reversion Regime:
if current_vol_rank < self.p.low_vol_thresh:
:self.bullish_climax_armed
(meaning price broke below BB and
RSI was oversold) and the current close[0]
moves back
above the bband.bot[0]
, a buy
order is
placed. A similar logic applies for bearish entries when
self.bearish_climax_armed
and price moves back
below bband.top[0]
. The return
statement
after attempted entry ensures that only one action is taken per bar in
this sub-strategy.close[0]
is below
bband.bot[0]
AND rsi[0]
is
oversold
. self.bullish_climax_armed
is set to
True
.close[0]
is above
bband.top[0]
AND rsi[0]
is
overbought
. self.bearish_climax_armed
is set
to True
. This “arming” sets up the next bar for a potential
mean-reversion entry if price begins to reverse.Medium-Volatility / Trend-Following Regime:
else:
: This block executes if the
current_vol_rank
is not in the high or low thresholds,
indicating a medium-volatility (trending) environment.impulse_ema
is rising AND macd_histo
is
rising (bullish impulse), a buy
order is placed.impulse_ema
is falling AND macd_histo
is falling (bearish impulse), a sell
order is placed.Position Management (ATR Trailing Stop):
if self.position:
: Regardless of the regime, if a
position is open, the manual ATR trailing stop is continuously
updated.stop_price
always
trails the highest_price_since_entry
, moving up but never
down. If close[0]
falls below stop_price
, the
position is closed.stop_price
always
trails the lowest_price_since_entry
, moving down but never
up. If close[0]
rises above stop_price
, the
position is closed.To evaluate the strategy’s performance, we’ll employ a rolling backtest. This method assesses the strategy over multiple, successive time windows, providing a more robust view of its consistency compared to a single, fixed backtest.
from collections import deque
import backtrader as bt
import numpy as np
import pandas as pd
import yfinance as yf
import dateutil.relativedelta as rd
# Assuming PercentRank and VolatilityRegimeStrategy classes are defined above this section.
def run_rolling_backtest(
ticker,
start,
end,
window_months,
strategy_class,=None
strategy_params
):= strategy_params or {}
strategy_params = []
all_results = pd.to_datetime(start)
start_dt = pd.to_datetime(end)
end_dt = start_dt
current_start
while True:
= current_start + rd.relativedelta(months=window_months)
current_end if current_end > end_dt:
= end_dt
current_end if current_start >= current_end:
break
print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")
= yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
data
if data.empty or len(data) < 90:
print("Not enough data for this period. Skipping.")
+= rd.relativedelta(months=window_months)
current_start continue
if isinstance(data.columns, pd.MultiIndex):
= data.droplevel(1, 1)
data
= data['Close'].iloc[0]
start_price = data['Close'].iloc[-1]
end_price = (end_price - start_price) / start_price * 100
benchmark_ret
= bt.feeds.PandasData(dataname=data)
feed = bt.Cerebro()
cerebro
**strategy_params)
cerebro.addstrategy(strategy_class,
cerebro.adddata(feed)100000)
cerebro.broker.setcash(=0.001)
cerebro.broker.setcommission(commission=95)
cerebro.addsizer(bt.sizers.PercentSizer, percents
= cerebro.broker.getvalue()
start_val try:
cerebro.run()except Exception as e:
print(f"Error running backtest for {current_start.date()} to {current_end.date()}: {e}")
+= rd.relativedelta(months=window_months)
current_start continue
= cerebro.broker.getvalue()
final_val = (final_val - start_val) / start_val * 100
strategy_ret
all_results.append({'start': current_start.date(),
'end': current_end.date(),
'strategy_return_pct': strategy_ret,
'benchmark_return_pct': benchmark_ret,
'final_value': final_val,
})
print(f"Strategy Return: {strategy_ret:.2f}% | Buy & Hold Return: {benchmark_ret:.2f}%")
+= rd.relativedelta(months=window_months)
current_start
if current_start > end_dt:
break
return pd.DataFrame(all_results)
ticker
, start
, end
: The asset
symbol and the overall historical period for the backtest.window_months
: The duration of each individual
backtesting window in months.strategy_class
: The backtrader.Strategy
class to be tested (e.g., VolatilityRegimeStrategy
).strategy_params
: A dictionary to pass specific
parameters to the chosen strategy for each run.yfinance
with
auto_adjust=False
and droplevel
applied.backtrader.Cerebro
instance.strategy_class
and its
parameters.Pandas DataFrame
containing the
comprehensive results for each rolling window.
The VolatilityRegimeStrategy
represents a significant
step towards building truly adaptive trading systems.
By dynamically adjusting its approach based on the current market
volatility environment, it seeks to optimize performance by employing
the most suitable sub-strategy (trend-following, mean-reversion, or
risk-off) for the prevailing conditions. This adaptive framework,
combined with robust risk management through an ATR trailing stop,
offers a comprehensive and intelligent approach to navigating diverse
market dynamics. The use of a rolling backtest is
essential for thoroughly evaluating the consistency and robustness of
such an adaptive strategy across various historical periods.