Article

AO Twin Peaks + Market Structure Break + ATR Trailing Stop with VecortBT

This article presents a complete, publish-ready implementation of a rule-based strategy that combines the Awesome Oscillator (AO) “twin peaks” divergence pattern, a Market Structure Break (MSB) entry trigger using confirmed swing levels, and an ATR-based trailing stop for exits. It also shows how to run a practical parameter optimization workflow in VectorBT, including best-parameter reporting, equity curves for top configurations, and heatmap visualizations.

Strategy overview

The system is built from three components:

AO twin peaks setup: a divergence pattern that “arms” a bullish or bearish bias when momentum fails to confirm price extremes.
MSB entry trigger: a trade is entered only when price breaks a confirmed swing level in the direction implied by the armed setup.
ATR trailing stop: positions are exited using a volatility-adjusted trailing stop derived from Wilder’s ATR.

The combination aims to avoid early divergence entries, require confirmation through structure, and manage risk with adaptive exits.

Indicators and definitions

Awesome Oscillator (AO):

Median price = (High + Low) / 2
AO = SMA(median, fast) − SMA(median, slow)

Twin peaks divergence logic:

Bearish divergence arms when price forms a higher high while AO forms a lower high.
Bullish divergence arms when price forms a lower low while AO forms a higher low.
Local AO peaks/troughs are detected using a 3-point pattern (i−2, i−1, i), where the middle point is a local extremum.

Confirmed swing levels:

Swing highs and lows are identified using a centered rolling window of length (2*swing_period + 1).
A swing is considered “confirmed” only after swing_period bars; the detected swing level is shifted forward and forward-filled to represent the most recently confirmed structure level.

ATR (Wilder):

True range is computed from (High−Low), |High−PrevClose|, |Low−PrevClose|
ATR is Wilder-smoothed via an exponential moving average with alpha = 1/atr_period.

Entry and exit rules

Entry conditions (flat only):

Short entry: bearish divergence armed AND Close breaks below the last confirmed swing low.
Long entry: bullish divergence armed AND Close breaks above the last confirmed swing high.

Exit conditions (positioned):

Long: trail stop at max(previous_stop, highest_high_since_entry − ATR * multiplier). Exit when Close < stop.
Short: trail stop at min(previous_stop, lowest_low_since_entry + ATR * multiplier). Exit when Close > stop.

Full implementation in VectorBT

import numpy as np
import pandas as pd
import vectorbt as vbt

# Data
df = vbt.YFData.download("ETH-USD", interval="1d", start="2025-01-01").get().dropna()

def run_ao_twinpeaks_msb_atrtrail(
    df,
    ao_fast=7,
    ao_slow=30,
    swing_period=7,
    atr_period=7,
    atr_stop_mult=3.0,
):
    o, h, l, c = (df["Open"], df["High"], df["Low"], df["Close"])

    # Awesome Oscillator
    median = (h + l) / 2.0
    ao = median.rolling(ao_fast).mean() - median.rolling(ao_slow).mean()

    # Confirmed swing points
    ws = 2 * swing_period + 1
    swing_high_raw = h.where(h == h.rolling(ws, center=True).max())
    swing_low_raw  = l.where(l == l.rolling(ws, center=True).min())
    swing_high = swing_high_raw.shift(swing_period).ffill()
    swing_low  = swing_low_raw.shift(swing_period).ffill()

    # ATR (Wilder)
    prev_c = c.shift(1)
    tr = pd.concat([(h - l), (h - prev_c).abs(), (l - prev_c).abs()], axis=1).max(axis=1)
    atr = tr.ewm(alpha=1/atr_period, adjust=False).mean()

    n = len(df)
    entries = np.zeros(n, dtype=bool)
    exits   = np.zeros(n, dtype=bool)
    short_entries = np.zeros(n, dtype=bool)
    short_exits   = np.zeros(n, dtype=bool)

    bear_armed = False
    bull_armed = False
    ao_peaks, px_peaks = [], []
    ao_troughs, px_troughs = [], []

    pos = 0  # 1 long, -1 short, 0 flat
    stop = np.nan
    hi_since = -np.inf
    lo_since = np.inf

    ao_v = ao.to_numpy()
    h_v, l_v, c_v = h.to_numpy(), l.to_numpy(), c.to_numpy()
    sh_v, sl_v = swing_high.to_numpy(), swing_low.to_numpy()
    atr_v = atr.to_numpy()

    for i in range(2, n):
        # Detect local AO peak/trough at i-1
        if np.isfinite(ao_v[i-2]) and np.isfinite(ao_v[i-1]) and np.isfinite(ao_v[i]):
            if ao_v[i-2] < ao_v[i-1] > ao_v[i]:
                ao_peaks.append(ao_v[i-1]); px_peaks.append(h_v[i-1])
                ao_peaks, px_peaks = ao_peaks[-2:], px_peaks[-2:]
            if ao_v[i-2] > ao_v[i-1] < ao_v[i]:
                ao_troughs.append(ao_v[i-1]); px_troughs.append(l_v[i-1])
                ao_troughs, px_troughs = ao_troughs[-2:], px_troughs[-2:]

        # Arm on divergence
        if len(ao_peaks) == 2 and px_peaks[1] > px_peaks[0] and ao_peaks[1] < ao_peaks[0]:
            bear_armed, bull_armed = True, False
            ao_peaks.clear(); px_peaks.clear()

        if len(ao_troughs) == 2 and px_troughs[1] < px_troughs[0] and ao_troughs[1] > ao_troughs[0]:
            bull_armed, bear_armed = True, False
            ao_troughs.clear(); px_troughs.clear()

        # Entries on MSB
        if pos == 0:
            if bear_armed and np.isfinite(sl_v[i-1]) and c_v[i] < sl_v[i-1]:
                short_entries[i] = True
                pos = -1
                bear_armed = False
                lo_since = l_v[i]
                stop = lo_since + atr_v[i] * atr_stop_mult if np.isfinite(atr_v[i]) else np.nan

            elif bull_armed and np.isfinite(sh_v[i-1]) and c_v[i] > sh_v[i-1]:
                entries[i] = True
                pos = 1
                bull_armed = False
                hi_since = h_v[i]
                stop = hi_since - atr_v[i] * atr_stop_mult if np.isfinite(atr_v[i]) else np.nan

        # Trailing ATR stop exits
        else:
            if pos == 1:
                hi_since = max(hi_since, h_v[i])
                if np.isfinite(atr_v[i]):
                    stop = max(stop, hi_since - atr_v[i] * atr_stop_mult) if np.isfinite(stop) else (hi_since - atr_v[i]*atr_stop_mult)
                if np.isfinite(stop) and c_v[i] < stop:
                    exits[i] = True
                    pos = 0
                    stop = np.nan
                    bear_armed = bull_armed = False

            elif pos == -1:
                lo_since = min(lo_since, l_v[i])
                if np.isfinite(atr_v[i]):
                    stop = min(stop, lo_since + atr_v[i] * atr_stop_mult) if np.isfinite(stop) else (lo_since + atr_v[i]*atr_stop_mult)
                if np.isfinite(stop) and c_v[i] > stop:
                    short_exits[i] = True
                    pos = 0
                    stop = np.nan
                    bear_armed = bull_armed = False

    pf = vbt.Portfolio.from_signals(
        c,
        entries=entries,
        exits=exits,
        short_entries=short_entries,
        short_exits=short_exits,
        freq="1D",
        init_cash=10_000,
        fees=0.001,
        slippage=0.0005,
    )
    return pf

Baseline backtest

pf = run_ao_twinpeaks_msb_atrtrail(df)
print(pf.stats())
pf.plot().show()

Parameter optimization with reporting, heatmaps, and equity curves

The following grid search explores AO windows and the ATR stop multiplier. The heatmap displays the best Sharpe for each (ao_fast, ao_slow) pair after selecting the best atr_stop_mult in that cell.

import numpy as np
import pandas as pd
import vectorbt as vbt

df = vbt.YFData.download("ETH-USD", interval="1d", start="2025-01-01").get().dropna()

ao_fast_range = np.arange(5, 21, 1)
ao_slow_range = np.arange(20, 61, 2)
atr_stop_mult_range = np.array([2.0, 2.5, 3.0, 3.5, 4.0])

swing_period = 7
atr_period = 7

rows = []
for af in ao_fast_range:
    for aslow in ao_slow_range:
        if af >= aslow:
            continue
        for sm in atr_stop_mult_range:
            pf = run_ao_twinpeaks_msb_atrtrail(
                df,
                ao_fast=int(af),
                ao_slow=int(aslow),
                swing_period=int(swing_period),
                atr_period=int(atr_period),
                atr_stop_mult=float(sm),
            )
            rows.append({
                "ao_fast": int(af),
                "ao_slow": int(aslow),
                "atr_stop_mult": float(sm),
                "Sharpe": float(pf.sharpe_ratio()),
                "CAGR": float(pf.annualized_return()),
                "TotalReturn": float(pf.total_return()),
                "MaxDD": float(pf.max_drawdown()),
                "Trades": int(pf.trades.count()),
            })

res = pd.DataFrame(rows).dropna()

# Best parameter sets
best_sharpe = res.loc[res["Sharpe"].idxmax()]
best_cagr = res.loc[res["CAGR"].idxmax()]

print("\nBest by Sharpe")
print(best_sharpe[["ao_fast","ao_slow","atr_stop_mult","Sharpe","CAGR","TotalReturn","MaxDD","Trades"]])

print("\nBest by CAGR")
print(best_cagr[["ao_fast","ao_slow","atr_stop_mult","Sharpe","CAGR","TotalReturn","MaxDD","Trades"]])

TOP_N = 20
print("\nTop Sharpe combinations")
print(res.sort_values("Sharpe", ascending=False).head(TOP_N)[
    ["ao_fast","ao_slow","atr_stop_mult","Sharpe","CAGR","TotalReturn","MaxDD","Trades"]
])

print("\nTop CAGR combinations")
print(res.sort_values("CAGR", ascending=False).head(TOP_N)[
    ["ao_fast","ao_slow","atr_stop_mult","Sharpe","CAGR","TotalReturn","MaxDD","Trades"]
])

# Heatmap: best Sharpe per (ao_fast, ao_slow) over atr_stop_mult
best_per_cell = res.sort_values("Sharpe", ascending=False).drop_duplicates(["ao_fast","ao_slow"])
sharpe_hm = best_per_cell.pivot(index="ao_fast", columns="ao_slow", values="Sharpe").sort_index().sort_index(axis=1)

sharpe_hm.vbt.heatmap(
    xaxis_title="ao_slow",
    yaxis_title="ao_fast",
    trace_kwargs=dict(colorbar_title="Best Sharpe (over atr_stop_mult)")
).show()

# Heatmap: which atr_stop_mult wins per cell
mult_hm = best_per_cell.pivot(index="ao_fast", columns="ao_slow", values="atr_stop_mult").sort_index().sort_index(axis=1)
mult_hm.vbt.heatmap(
    xaxis_title="ao_slow",
    yaxis_title="ao_fast",
    trace_kwargs=dict(colorbar_title="Best atr_stop_mult")
).show()

# Equity curves for best Sharpe and best CAGR
pf_best_sharpe = run_ao_twinpeaks_msb_atrtrail(
    df,
    ao_fast=int(best_sharpe["ao_fast"]),
    ao_slow=int(best_sharpe["ao_slow"]),
    swing_period=swing_period,
    atr_period=atr_period,
    atr_stop_mult=float(best_sharpe["atr_stop_mult"]),
)

pf_best_cagr = run_ao_twinpeaks_msb_atrtrail(
    df,
    ao_fast=int(best_cagr["ao_fast"]),
    ao_slow=int(best_cagr["ao_slow"]),
    swing_period=swing_period,
    atr_period=atr_period,
    atr_stop_mult=float(best_cagr["atr_stop_mult"]),
)

pf_best_sharpe.value().vbt.plot(
    title=f"Equity (Best Sharpe) af={int(best_sharpe['ao_fast'])} as={int(best_sharpe['ao_slow'])} mult={best_sharpe['atr_stop_mult']}"
).show()

pf_best_cagr.value().vbt.plot(
    title=f"Equity (Best CAGR) af={int(best_cagr['ao_fast'])} as={int(best_cagr['ao_slow'])} mult={best_cagr['atr_stop_mult']}"
).show()

# Optional: equity curves for top K Sharpe configs
TOP_K = 5
topk = res.sort_values("Sharpe", ascending=False).head(TOP_K)

equity_curves = []
curve_cols = []
for _, r in topk.iterrows():
    pf_k = run_ao_twinpeaks_msb_atrtrail(
        df,
        ao_fast=int(r["ao_fast"]),
        ao_slow=int(r["ao_slow"]),
        swing_period=swing_period,
        atr_period=atr_period,
        atr_stop_mult=float(r["atr_stop_mult"]),
    )
    equity_curves.append(pf_k.value())
    curve_cols.append(f"af={int(r['ao_fast'])},as={int(r['ao_slow'])},m={r['atr_stop_mult']}")

equity_df = pd.concat(equity_curves, axis=1)
equity_df.columns = curve_cols

equity_df.vbt.plot(title=f"Equity Curves (Top {TOP_K} Sharpe configs)").show()

Interpreting the outputs

Best-parameter reporting:

“Best by Sharpe” identifies the configuration with the strongest risk-adjusted return.

Best by Sharpe: ao_fast 5.000000 ao_slow 50.000000 atr_stop_mult 3.500000 Sharpe 2.117923 CAGR 1.534235 TotalReturn 1.893219 MaxDD -0.228883

“Best by CAGR” identifies the configuration with the highest annualized growth (often at the cost of higher drawdowns).

Best by CAGR: ao_fast 5.000000 ao_slow 50.000000 atr_stop_mult 3.500000 Sharpe 2.117923 CAGR 1.534235 TotalReturn 1.893219 MaxDD -0.228883

Heatmaps:

Sharpe heatmap shows the best achievable Sharpe at each AO fast/slow pair (after choosing the best ATR multiplier within that cell).

ATR multiplier heatmap shows which stop multiplier produced that best Sharpe for each AO pair.

Equity curves:

The best-by-Sharpe and best-by-CAGR equity lines visualize how each “winner” achieved its outcome (smooth compounding vs. high-variance bursts).

The top-K Sharpe overlay helps assess robustness: a tight cluster of similar curves suggests the parameter region may be more stable than a single outlier.

Reproducibility notes

The backtest uses daily candles, applies fees and slippage, and supports both long and short entries.
The strategy uses confirmed swing levels (shifted by swing_period) to avoid acting on unconfirmed structure points.
Optimization is a grid search because the strategy is stateful and not easily expressed as fully vectorized signal arrays.

Disclaimer

This is research and educational code. Real-world execution, liquidity, gap risk, and exchange-specific constraints can materially change results.