Hilbert-ADX Portfolio Trading Strategy
Description:
This strategy rotates across major crypto assets using cycle
timing + trend confirmation.
The Hilbert Transform SineWave detects short-term cycle turns:
Sine crosses above LeadSine → bullish timing signal
Sine crosses below LeadSine → bearish timing signal
Trend regime is confirmed with ADX:
ADX > 20 = tradable trend
Direction is confirmed by DI dominance:
+DI > -DI → long only
-DI > +DI → short only
Active long and short signals are sized by inverse volatility:
\[w_i=\frac{1/\sigma_i}{\sum_{j=1}^{n}1/\sigma_j}\]
Risk is controlled with a 5% trailing stop:
\[Stop_t=Peak_t\times(1-0.05)\]
Python Bakctesting Code:
import backtrader as bt
import yfinance as yf
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
class MultiAssetHilbertTrendInvVolTrail(bt.Strategy):
params = dict(
adx_period=30,
adx_threshold=20,
trail_percent=0.05,
vol_lookback=30,
rebalance_days=10,
max_gross_exposure=0.95,
)
def __init__(self):
self.ind = {}
self.weights = {}
self.directions = {}
self.order_refs = {}
self.stop_orders = {}
self.last_rebalance = -999
self.date_history = []
self.value_history = []
self.asset_value_history = {d._name: [] for d in self.datas}
for d in self.datas:
ht = bt.talib.HT_SINE(d.close)
sine_cross = bt.ind.CrossOver(ht.sine, ht.leadsine)
adx = bt.ind.ADX(d, period=self.p.adx_period)
plus_di = bt.ind.PlusDI(d, period=self.p.adx_period)
minus_di = bt.ind.MinusDI(d, period=self.p.adx_period)
self.ind[d] = dict(
sine_cross=sine_cross,
adx=adx,
plus_di=plus_di,
minus_di=minus_di,
)
self.weights[d] = 0.0
self.directions[d] = 0
def next(self):
need_rebalance = False
for d in self.datas:
if d in self.order_refs:
continue
if len(d) < max(self.p.adx_period, self.p.vol_lookback) + 50:
continue
pos = self.getposition(d)
ind = self.ind[d]
trending = ind["adx"][0] > self.p.adx_threshold
uptrend = trending and ind["plus_di"][0] > ind["minus_di"][0]
downtrend = trending and ind["minus_di"][0] > ind["plus_di"][0]
cross_up = ind["sine_cross"][0] > 0
cross_down = ind["sine_cross"][0] < 0
if pos.size == 0:
if uptrend and cross_up:
self.directions[d] = 1
need_rebalance = True
elif downtrend and cross_down:
self.directions[d] = -1
need_rebalance = True
if len(self) - self.last_rebalance >= self.p.rebalance_days:
need_rebalance = True
if need_rebalance:
self.last_rebalance = len(self)
active = [d for d in self.datas if self.directions[d] != 0]
vols = {}
for d in active:
rets = []
for i in range(1, self.p.vol_lookback + 1):
if d.close[-i] == 0:
continue
r = d.close[-i + 1] / d.close[-i] - 1
rets.append(r)
if len(rets) >= 10:
vol = np.std(rets)
if vol > 0 and np.isfinite(vol):
vols[d] = vol
for d in self.datas:
self.weights[d] = 0.0
inv_sum = sum(1 / v for v in vols.values())
if inv_sum > 0:
for d, vol in vols.items():
self.weights[d] = (1 / vol) / inv_sum
equity = self.broker.getvalue()
for d in self.datas:
if d in self.order_refs:
continue
price = d.close[0]
if price <= 0:
continue
target_value = equity * self.weights[d] * self.p.max_gross_exposure * self.directions[d]
target_size = target_value / price
current_size = self.getposition(d).size
delta = target_size - current_size
if abs(delta * price) < 10:
continue
if delta > 0:
self.order_refs[d] = self.buy(data=d, size=delta)
elif delta < 0:
self.order_refs[d] = self.sell(data=d, size=abs(delta))
for d in self.datas:
pos = self.getposition(d)
if pos.size > 0 and d not in self.stop_orders:
self.stop_orders[d] = self.sell(
data=d,
size=pos.size,
exectype=bt.Order.StopTrail,
trailpercent=self.p.trail_percent,
)
elif pos.size < 0 and d not in self.stop_orders:
self.stop_orders[d] = self.buy(
data=d,
size=abs(pos.size),
exectype=bt.Order.StopTrail,
trailpercent=self.p.trail_percent,
)
self.date_history.append(self.datas[0].datetime.datetime(0))
self.value_history.append(self.broker.getvalue())
for d in self.datas:
pos = self.getposition(d)
self.asset_value_history[d._name].append(abs(pos.size * d.close[0]))
def notify_order(self, order):
d = order.data
if order.status in [order.Submitted, order.Accepted]:
return
if d in self.order_refs and order.ref == self.order_refs[d].ref:
del self.order_refs[d]
if d in self.stop_orders and order.ref == self.stop_orders[d].ref:
del self.stop_orders[d]
if order.status == order.Completed:
self.directions[d] = 0
self.weights[d] = 0.0
cerebro = bt.Cerebro()
assets = [
"BTC-USD", "ETH-USD", "BNB-USD", "SOL-USD",
"XRP-USD", "ADA-USD", "DOGE-USD", "AVAX-USD",
"LINK-USD", "DOT-USD", "LTC-USD", "BCH-USD",
]
START_DATE = "2025-01-01"
END_DATE = None
INIT_CASH = 100_000
raw_data = {}
for t in assets:
df = yf.download(
t,
start=START_DATE,
end=END_DATE,
interval="1d",
progress=False,
auto_adjust=False,
).droplevel(1, 1)
df.dropna(inplace=True)
if df.empty:
print(f"Skipping {t}: no data.")
continue
raw_data[t] = df.copy()
data = bt.feeds.PandasData(dataname=df, name=t)
cerebro.adddata(data)
cerebro.addstrategy(MultiAssetHilbertTrendInvVolTrail)
cerebro.broker.setcash(INIT_CASH)
cerebro.broker.setcommission(commission=0.001)
cerebro.broker.set_shortcash(True)
cerebro.broker.set_slippage_perc(perc=0.0005)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe", timeframe=bt.TimeFrame.Days)
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="dd")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")
start_value = cerebro.broker.getvalue()
results = cerebro.run()
strat = results[0]
end_value = cerebro.broker.getvalue()
print(f"Start Value: {start_value:,.2f}")
print(f"End Value: {end_value:,.2f}")
print(f"Return: {(end_value / start_value - 1) * 100:.2f}%")
print(f"Sharpe: {strat.analyzers.sharpe.get_analysis()}")
print(f"DrawDown: {strat.analyzers.dd.get_analysis()}")
print(f"Trades: {strat.analyzers.trades.get_analysis()}")
eq = pd.Series(strat.value_history, index=pd.to_datetime(strat.date_history), name="Strategy")
asset_values = pd.DataFrame(strat.asset_value_history, index=eq.index)
asset_values = asset_values.clip(lower=0)
cash_plot = (eq - asset_values.sum(axis=1)).clip(lower=0)
bench_close = pd.DataFrame(index=eq.index)
for t, df in raw_data.items():
bench_close[t] = df["Close"].reindex(eq.index, method="ffill")
bench_ret = bench_close.pct_change().fillna(0.0)
equal_bh = INIT_CASH * (1 + bench_ret.mean(axis=1)).cumprod()
btc_close = bench_close["BTC-USD"].dropna()
btc_bh = INIT_CASH * (btc_close / btc_close.iloc[0])
btc_bh = btc_bh.reindex(eq.index, method="ffill")
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 9), sharex=True)
ax1.plot(eq.index, eq.values, color="black", linewidth=2, label="Hilbert ADX InvVol Trail")
ax1.plot(equal_bh.index, equal_bh.values, linestyle="--", label="Equal-Weight Buy & Hold")
ax1.plot(btc_bh.index, btc_bh.values, linestyle="--", label="BTC Buy & Hold")
pf_ret = eq.iloc[-1] / eq.iloc[0] - 1
ew_ret = equal_bh.iloc[-1] / equal_bh.iloc[0] - 1
btc_ret = btc_bh.iloc[-1] / btc_bh.iloc[0] - 1
x_last = eq.index[-1]
ax1.annotate(f"PF: {pf_ret * 100:.2f}%", xy=(x_last, eq.iloc[-1]), xytext=(8, 0),
textcoords="offset points", va="center", ha="left", color="black", clip_on=False)
ax1.annotate(f"EW: {ew_ret * 100:.2f}%", xy=(x_last, equal_bh.iloc[-1]), xytext=(8, 0),
textcoords="offset points", va="center", ha="left", clip_on=False)
ax1.annotate(f"BTC: {btc_ret * 100:.2f}%", xy=(x_last, btc_bh.iloc[-1]), xytext=(8, 0),
textcoords="offset points", va="center", ha="left", clip_on=False)
ax1.grid(True, alpha=0.3)
ax1.legend()
labels = ["Cash"] + list(asset_values.columns)
colors = ["#d3d3d3"] + list(plt.cm.tab10.colors)[:len(asset_values.columns)]
ax2.stackplot(
eq.index,
cash_plot,
*[asset_values[col] for col in asset_values.columns],
labels=labels,
colors=colors,
alpha=0.8,
)
ax2.set_title("Portfolio Allocation Breakdown")
ax2.set_ylabel("Value ($)")
ax2.grid(True, alpha=0.3)
ax2.legend(loc="upper left", bbox_to_anchor=(1.0, 1), frameon=True)
plt.tight_layout()
plt.show()