Weekly Regime Breakout Crypto Rotation Engine
Description:
This strategy rotates across major crypto assets by looking for
weekly upside breakouts only when the market shows evidence of a
trending regime.
It first defines a 30-day price range:
\[Support=\min(Close_{t-30})\]
\[Resistance=\max(Close_{t-30})\]
A trade becomes eligible when price breaks above resistance:
\[Close_t>Resistance\]
But the breakout must also pass three filters:
KPSS p-value < 0.05 → market is not
range-stationary
ADX > 25 → trend strength is high
+DI > -DI → bullish directional pressure dominates
Capital is then distributed across active breakout assets using inverse volatility:
\[w_i=\frac{1/\sigma_i}{\sum_{j=1}^{n}1/\sigma_j}\]
This gives larger allocations to smoother assets and smaller allocations to more volatile ones.
Each position is protected by a 5% trailing stop:
\[Stop_t=Peak_t\times(1-0.05)\]
Python Script
import warnings
warnings.filterwarnings("ignore")
import yfinance as yf
import pandas as pd
import backtrader as bt
import numpy as np
import matplotlib.pyplot as plt
from datetime import timedelta
from statsmodels.tsa.stattools import kpss
class WeeklyBreakoutInvVolTrail(bt.Strategy):
params = dict(
lookback=30,
adx_min=20,
vol_lookback=30,
rebalance_days=10,
trail_percent=0.05,
max_gross_exposure=0.95,
)
def __init__(self):
self.adx = {}
self.di_plus = {}
self.di_minus = {}
self.next_eval_date = {}
self.week_end = {}
self.resistance = {}
self.support = {}
self.valid_signal = {}
self.directions = {}
self.weights = {}
self.order_refs = {}
self.stop_orders = {}
self.open_trade_index = {}
self.last_rebalance = -999
self.trades_log = []
self.date_history = []
self.value_history = []
self.asset_value_history = {d._name: [] for d in self.datas}
for d in self.datas:
self.adx[d] = bt.ind.ADX(d, period=14)
self.di_plus[d] = bt.ind.PlusDI(d, period=14)
self.di_minus[d] = bt.ind.MinusDI(d, period=14)
self.next_eval_date[d] = None
self.week_end[d] = None
self.resistance[d] = None
self.support[d] = None
self.valid_signal[d] = False
self.directions[d] = 0
self.weights[d] = 0.0
self.open_trade_index[d] = None
def next(self):
need_rebalance = False
for d in self.datas:
date = d.datetime.date(0)
if d in self.order_refs:
continue
if len(d) < self.p.lookback + 20:
continue
if self.next_eval_date[d] is None:
self.next_eval_date[d] = date
if date >= self.next_eval_date[d]:
closes = list(d.close.get(size=self.p.lookback + 1))[:-1]
if len(closes) < self.p.lookback:
continue
try:
kpss_p = kpss(closes, regression="c", nlags="auto")[1]
except Exception:
kpss_p = 1
self.support[d] = min(closes)
self.resistance[d] = max(closes)
self.week_end[d] = date + timedelta(days=6)
self.next_eval_date[d] = date + timedelta(days=7)
self.valid_signal[d] = (
kpss_p < 0.05
and self.adx[d][0] > self.p.adx_min
and self.di_plus[d][0] > self.di_minus[d][0]
)
pos = self.getposition(d)
if pos.size == 0 and self.valid_signal[d] and d.close[0] > self.resistance[d]:
self.directions[d] = 1
need_rebalance = True
self.trades_log.append({
"symbol": d._name,
"entry_date": date,
"entry_price": d.close[0],
"resistance": self.resistance[d],
"support": self.support[d],
"adx": self.adx[d][0],
"di_plus": self.di_plus[d][0],
"di_minus": self.di_minus[d][0],
})
self.open_trade_index[d] = len(self.trades_log) - 1
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] == 1]
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
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,
)
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(max(0.0, 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
idx = self.open_trade_index[d]
if idx is not None:
self.trades_log[idx]["exit_date"] = d.datetime.date(0)
self.trades_log[idx]["exit_price"] = d.close[0]
self.trades_log[idx]["pnl"] = d.close[0] / self.trades_log[idx]["entry_price"] - 1
self.open_trade_index[d] = None
symbols = [
"BTC-USD", "ETH-USD", "BNB-USD", "SOL-USD",
"XRP-USD", "ADA-USD", "DOGE-USD", "AVAX-USD",
]
START_DATE = "2025-01-01"
END_DATE = None
INIT_CASH = 10_000
cerebro = bt.Cerebro()
raw_data = {}
for symbol in symbols:
df = yf.download(
symbol,
start=START_DATE,
end=END_DATE,
auto_adjust=False,
progress=False,
).droplevel(1, 1)
df = df.dropna()
if df.empty:
print(f"Skipping {symbol}: no data.")
continue
raw_data[symbol] = df.copy()
feed = bt.feeds.PandasData(dataname=df, name=symbol)
cerebro.adddata(feed)
cerebro.broker.setcash(INIT_CASH)
cerebro.broker.setcommission(commission=0.001)
cerebro.broker.set_coc(True)
cerebro.addstrategy(
WeeklyBreakoutInvVolTrail,
lookback=30,
adx_min=25,
vol_lookback=30,
rebalance_days=7,
trail_percent=0.05,
max_gross_exposure=0.95,
)
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")
start_value = cerebro.broker.getvalue()
print("Starting Value:", round(start_value, 2))
results = cerebro.run()
strategy = results[0]
end_value = cerebro.broker.getvalue()
print("Final Value:", round(end_value, 2))
print("Return:", round((end_value / start_value - 1) * 100, 2), "%")
trades_df = pd.DataFrame(strategy.trades_log)
if len(trades_df) > 0:
if "pnl" in trades_df.columns:
trades_df["pnl"] = trades_df["pnl"].round(4)
print("\n=== TRADES ===")
print(trades_df)
if "pnl" in trades_df.columns:
print("\n=== SUMMARY BY SYMBOL ===")
print(trades_df.groupby("symbol")["pnl"].agg(["count", "sum", "mean"]))
else:
print("\nNo trades found.")
print("\n=== RETURNS ===")
print(strategy.analyzers.returns.get_analysis())
print("\n=== DRAWDOWN ===")
print(strategy.analyzers.drawdown.get_analysis())
print("\n=== TRADES ANALYZER ===")
print(strategy.analyzers.trades.get_analysis())
eq = pd.Series(strategy.value_history, index=pd.to_datetime(strategy.date_history), name="Strategy")
asset_values = pd.DataFrame(strategy.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="Weekly Breakout 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")
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()