Volatility-Adaptive Trend Strategy Using Dynamic Moving Average Crossover and Trailing Stop Protection
Description:
This strategy trades BTC using a custom moving average that reacts
faster when volatility rises and slower when volatility falls.
The core indicator is a volatility-adjusted EMA:
\[VAMA_t=\alpha_t Close_t+(1-\alpha_t)VAMA_{t-1}\]
Its speed changes with volatility:
\[\alpha_t=\frac{2}{N+1}\times\frac{Vol_t}{AvgVol_t}\]
When volatility expands, VAMA reacts faster.
When volatility contracts, VAMA becomes smoother.
Trading logic:
VAMA crosses above SMA → Enter long
VAMA crosses below SMA → Exit
Risk control:
\[Stop_t=Peak_t\times(1-0.05)\]
Python Code:
import backtrader as bt
import yfinance as yf
import pandas as pd
import math
import matplotlib.pyplot as plt
class VolatilityAdjustedMovingAverage(bt.Indicator):
lines = ("vama",)
params = (
("period", 30),
("vol_period", 7),
("min_alpha_ratio", 0.1),
("max_alpha_ratio", 2.0),
)
def __init__(self):
self.base_alpha = 2.0 / (self.p.period + 1.0)
self.vol = bt.indicators.StandardDeviation(
self.data.close,
period=self.p.vol_period,
)
self.avg_vol = bt.indicators.SimpleMovingAverage(
self.vol,
period=self.p.period,
)
self.addminperiod(self.p.vol_period + self.p.period)
def next(self):
close = self.data.close[0]
current_vol = self.vol[0]
avg_vol = self.avg_vol[0]
if avg_vol != 0 and not math.isnan(avg_vol):
vol_ratio = current_vol / avg_vol
else:
vol_ratio = 1.0
vol_ratio = max(
self.p.min_alpha_ratio,
min(vol_ratio, self.p.max_alpha_ratio),
)
alpha = self.base_alpha * vol_ratio
alpha = max(1e-9, min(alpha, 1.0))
if len(self) > 1 and not math.isnan(self.vama[-1]):
self.vama[0] = alpha * close + (1 - alpha) * self.vama[-1]
else:
self.vama[0] = close
class VamaStrategy(bt.Strategy):
params = (
("vama_period", 30),
("vama_vol_period", 7),
("sma_period", 90),
("trail_percent", 0.10),
("printlog", False),
)
def __init__(self):
self.vama = VolatilityAdjustedMovingAverage(
self.data,
period=self.p.vama_period,
vol_period=self.p.vama_vol_period,
)
self.sma = bt.indicators.SimpleMovingAverage(
self.data.close,
period=self.p.sma_period,
)
self.crossover = bt.indicators.CrossOver(self.vama, self.sma)
self.order = None
self.trail_order = None
self.equity_dates = []
self.equity_values = []
def log(self, txt):
if self.p.printlog:
print(f"{self.datas[0].datetime.date(0)} - {txt}")
def cancel_trail(self):
if self.trail_order:
self.cancel(self.trail_order)
self.trail_order = None
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status == order.Completed:
if order == self.order:
if order.isbuy():
self.log(f"BUY EXECUTED: {order.executed.price:.2f}")
self.trail_order = self.sell(
exectype=bt.Order.StopTrail,
trailpercent=self.p.trail_percent,
)
self.order = None
elif order == self.trail_order:
self.log(f"TRAIL STOP EXECUTED: {order.executed.price:.2f}")
self.trail_order = None
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
if order == self.order:
self.order = None
if order == self.trail_order:
self.trail_order = None
def notify_trade(self, trade):
if trade.isclosed:
self.log(f"TRADE CLOSED | Gross: {trade.pnl:.2f} | Net: {trade.pnlcomm:.2f}")
def next(self):
self.equity_dates.append(self.datas[0].datetime.date(0))
self.equity_values.append(self.broker.getvalue())
if self.order:
return
if len(self.data) < max(self.p.vama_period + self.p.vama_vol_period, self.p.sma_period) + 5:
return
if not self.position:
if self.crossover[0] > 0:
self.order = self.buy()
else:
if self.crossover[0] < 0:
self.cancel_trail()
self.order = self.close()
start_date = "2025-01-01"
end_date = None
cash = 100000.0
symbol = "BTC-USD"
data = yf.download(
symbol,
start=start_date,
end=end_date,
progress=False,
auto_adjust=False,
).droplevel(1, 1)
data.columns = data.columns.str.lower()
data.index = pd.to_datetime(data.index)
data = data.dropna()
cerebro = bt.Cerebro()
cerebro.addstrategy(
VamaStrategy,
vama_period=30,
vama_vol_period=10,
sma_period=30,
trail_percent=0.05,
printlog=False,
)
feed = bt.feeds.PandasData(dataname=data)
cerebro.adddata(feed)
cerebro.broker.setcash(cash)
cerebro.broker.setcommission(commission=0.001)
cerebro.addsizer(bt.sizers.PercentSizer, percents=95)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe", timeframe=bt.TimeFrame.Days)
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")
print("Starting Value:", cerebro.broker.getvalue())
results = cerebro.run()
strategy = results[0]
print("Final Value:", cerebro.broker.getvalue())
print("Sharpe:", strategy.analyzers.sharpe.get_analysis())
print("Drawdown:", strategy.analyzers.drawdown.get_analysis())
print("Returns:", strategy.analyzers.returns.get_analysis())
print("Trades:", strategy.analyzers.trades.get_analysis())
equity = pd.Series(strategy.equity_values, index=pd.to_datetime(strategy.equity_dates))
equity = equity[~equity.index.duplicated(keep="first")]
buy_hold = cash * data["close"] / data["close"].iloc[0]
buy_hold = buy_hold.reindex(equity.index).ffill()
print("Buy & Hold Final Value:", buy_hold.iloc[-1])
print("Strategy Excess Return:", cerebro.broker.getvalue() - buy_hold.iloc[-1])
plt.figure(figsize=(12, 6))
plt.plot(equity.index, equity.values, label="VAMA Strategy with Trailing Stop")
plt.plot(buy_hold.index, buy_hold.values, label="Buy and Hold")
plt.title(f"{symbol} VAMA Crossover Strategy vs Buy and Hold")
plt.xlabel("Date")
plt.ylabel("Portfolio Value")
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()