Imports¶
This section gathers, in one place, the libraries the notebook depends on, so every later cell can rely on them being loaded.
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import talib
from plotly.subplots import make_subplots
Config¶
This section collects the study's fixed choices — the strategy's parameters, the markets and timeframes it runs on, the fee rate, and starting capital — as named constants, set once here and referenced by name throughout.
FEE_PCT = 0.02 / 100 # binance futures taker, per side
INITIAL_CASH = 100
WMA_FAST_PERIOD = 20
WMA_SLOW_PERIOD = 50
# Chandelier trailing stop: highest close since entry, less ATR_MULTIPLIER
# times the Average True Range over ATR_PERIOD bars.
ATR_PERIOD = 22
ATR_MULTIPLIER = 3.0
SYMBOLS = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT", "DOGEUSDT"]
# Synthetic basket symbol — an equal-weight average of every symbol's
# normalized price history, measured like any other market.
BASKET_SYMBOL = "ALL"
# Resampled-history block length — whole stretches of real returns this long
# are drawn to build the resampled run; long enough to keep trends intact.
RESAMPLE_BLOCK_DAYS = 90
TIMEFRAMES = ["30m", "1h", "4h", "1d"]
DATA_DIR = "../data" # cached Binance klines, one CSV per symbol + timeframe
# Bars per year by timeframe — annualizes the Sharpe ratio.
BARS_PER_YEAR = {
"30m": 365 * 24 * 2,
"1h": 365 * 24,
"4h": 365 * 6,
"1d": 365,
}
# One colour per symbol, used for the overlaid lines in every chart.
SYMBOL_COLORS = {
"BTCUSDT": "#f7931a",
"ETHUSDT": "#627eea",
"SOLUSDT": "#14b8a6",
"BNBUSDT": "#f3ba2f",
"DOGEUSDT": "#9b59b6",
BASKET_SYMBOL: "#333333",
}
Strategy¶
This section defines the strategy: the signal it watches, the precise conditions that open and close a position, and the intuition for why it may hold an edge. A short example on simulated prices shows the entry and exit markers in isolation, so the mechanics are clear before any real data or performance is considered.
The strategy tracks two weighted moving averages of the close — a fast one over 20 bars and a slow one over 50. The weighted average leans on recent bars, so it turns sooner than a simple average. A long position opens the moment the fast average crosses above the slow:
$$\text{open at } t \iff \mathrm{WMA}_{20}(t-1) \le \mathrm{WMA}_{50}(t-1) \;\land\; \mathrm{WMA}_{20}(t) > \mathrm{WMA}_{50}(t).$$
The exit is where this variant differs. A plain crossover system waits for the fast average to fall back below the slow one — a slow signal that surrenders much of an open gain before it fires. Here the position also carries a chandelier trailing stop: it remembers the highest close reached since entry and exits if price falls a multiple of the Average True Range below that peak:
$$\text{stop}(t) = \max_{e \le s \le t} p_s \;-\; k \cdot \mathrm{ATR}_{22}(t), \qquad k = 3,$$
with $e$ the entry bar. The trade closes on the first of the two exits — the stop being hit or the downward crossover — so a sharp reversal banks the gain while the slow average is still turning. The stop ratchets up with new highs and never loosens, so it locks in more of each advance as the trend extends. The Average True Range scales the distance to each market's own volatility, keeping the stop wide enough to ride normal noise yet tight enough to cut a real reversal short.
The Average True Range is measured on the close series the strategy already reads — the average absolute close-to-close move over 22 bars — so the same rule applies unchanged to single markets, the synthetic basket, and the resampled history. The strategy is long-only and always fully in or fully out.
def close_atr(prices, period):
true_range = np.abs(np.diff(prices, prepend=prices[0]))
return talib.SMA(true_range, timeperiod=period)
def strategy(prices, fast, slow, atr_period, atr_multiplier):
wma = {
"fast": talib.WMA(prices, timeperiod=fast),
"slow": talib.WMA(prices, timeperiod=slow),
}
atr = close_atr(prices=prices, period=atr_period)
above = wma["fast"] > wma["slow"]
trailing_stop = np.full(len(prices), np.nan)
opens = []
closes = []
entry_index = None
highest_close = None
for i in range(1, len(prices)):
crossed_up = above[i] and not above[i - 1]
crossed_down = above[i - 1] and not above[i]
if entry_index is None:
if crossed_up:
entry_index = i
highest_close = prices[i]
trailing_stop[i] = highest_close - atr_multiplier * atr[i]
continue
highest_close = max(highest_close, prices[i])
stop_level = highest_close - atr_multiplier * atr[i]
trailing_stop[i] = stop_level
if crossed_down or prices[i] <= stop_level:
opens.append(entry_index)
closes.append(i)
entry_index = None
highest_close = None
return wma, np.array(opens, dtype=int), np.array(closes, dtype=int), trailing_stop
An illustrative example on a synthetic price path, drawn as candles around its closes, so the crossover entries (green) and exits (red) can be read against the two averages and the trailing stop that trails each position from above.
rng = np.random.default_rng(19)
example_prices = 100 * np.cumprod(1 + rng.normal(0, 0.01, size=300))
example_wma, example_opens, example_closes, example_trailing_stop = strategy(
prices=example_prices,
fast=WMA_FAST_PERIOD,
slow=WMA_SLOW_PERIOD,
atr_period=ATR_PERIOD,
atr_multiplier=ATR_MULTIPLIER,
)
# Candles are display-only, built around the close path the strategy reads:
# each bar opens at the prior close, with small simulated wicks.
candle_open_prices = np.concatenate(([example_prices[0]], example_prices[:-1]))
candle_high_prices = np.maximum(candle_open_prices, example_prices) * (
1 + rng.uniform(0, 0.004, size=example_prices.size)
)
candle_low_prices = np.minimum(candle_open_prices, example_prices) * (
1 - rng.uniform(0, 0.004, size=example_prices.size)
)
fig = go.Figure()
fig.add_trace(
go.Candlestick(
open=candle_open_prices,
high=candle_high_prices,
low=candle_low_prices,
close=example_prices,
name="Price",
increasing=dict(line=dict(color="#8a9bab", width=1), fillcolor="#f6fafd"),
decreasing=dict(line=dict(color="#8a9bab", width=1), fillcolor="#8a9bab"),
)
)
fig.add_trace(
go.Scatter(
y=example_wma["fast"], mode="lines", name="WMA 20", line=dict(color="#00d4aa", width=1)
)
)
fig.add_trace(
go.Scatter(
y=example_wma["slow"], mode="lines", name="WMA 50", line=dict(color="#1f77b4", width=1)
)
)
fig.add_trace(
go.Scatter(
y=example_trailing_stop,
mode="lines",
name="Trailing Stop",
line=dict(color="#ff9500", width=1, dash="dot"),
)
)
fig.add_trace(
go.Scatter(
x=example_opens,
y=example_prices[example_opens],
mode="markers",
name="Entries",
marker=dict(color="#00d4aa", size=9, symbol="triangle-up"),
)
)
fig.add_trace(
go.Scatter(
x=example_closes,
y=example_prices[example_closes],
mode="markers",
name="Exits",
marker=dict(color="#ff3b30", size=9, symbol="triangle-down"),
)
)
fig.update_layout(
template="plotly_white",
height=400,
width=1080,
margin=dict(l=60, r=20, t=40, b=40),
xaxis_rangeslider_visible=False,
)
fig.show()
Metrics¶
This section turns the strategy's trades into performance metrics, per-trade and cumulative. Each is defined with its formula below; equity is reported in both gross (before fees) and net (after fees) form, so the cost of trading is always visible.
- Symbol — the market this result belongs to (e.g. BTCUSDT); ALL is the equal-weight basket of every symbol, averaged from their normalized price histories over the common window.
- Price Change % — the close price as a percentage change from the start of the window (from the close series $p_0,\dots,p_T$); in the summary, the total change over the period — the buy-and-hold return.
- Moving Averages — the fast and slow weighted moving averages (WMA 20 and WMA 50) from which the entry signal is derived.
- Trailing Stop — the chandelier exit level: the highest close reached since entry, less $k=3$ times the Average True Range over 22 bars; the position closes when the close falls to it or the fast average crosses back below the slow, whichever comes first.
- Entries — bars where a position is opened.
- Exits — bars where a position is closed.
- Trade P&L % — per-trade net return, $\text{pnl}_i = (r_i - 1)\cdot 100\%$ with $r_i = (1 - f)^2\,p^{\text{exit}}_i / p^{\text{entry}}_i$ and $f$ the per-side fee.
- Cumulative Win Rate % — running share of winning trades, $\frac{1}{n}\sum_{i=1}^{n}\mathbf{1}\{\text{pnl}_i > 0\}\cdot 100\%$.
- Cumulative P&L % — the running sum of per-trade P&L.
- Equity — net equity curve, compounded all-in: $E_n = E_0 \prod_{i=1}^{n} r_i$; the gross variant drops the fee term.
- Cumulative Fees — the running total of fees paid, each proportional to capital at trade time.
- Rolling Sharpe — annualized Sharpe of net per-trade returns computed to-date after each trade, $S_n = \frac{\bar x}{\operatorname{std}(x)}\sqrt{T_\text{year}}$ over $x_1,\dots,x_n$, with $x_i = r_i - 1$ and $T_\text{year}$ the observed number of trades per year.
def analytics(symbol, prices, bars_per_year):
wma, opens, closes, trailing_stop = strategy(
prices=prices,
fast=WMA_FAST_PERIOD,
slow=WMA_SLOW_PERIOD,
atr_period=ATR_PERIOD,
atr_multiplier=ATR_MULTIPLIER,
)
trade_count = min(len(opens), len(closes))
opens = opens[:trade_count]
closes = closes[:trade_count]
entry_prices = prices[opens]
exit_prices = prices[closes]
# Per-trade return factor: (1 - fee)^2 covers entry + exit fees,
# (exit / entry) is the raw price move.
factor = (1 - FEE_PCT) ** 2 * (exit_prices / entry_prices)
# Equity compounded with all-in sizing.
equity = INITIAL_CASH * np.cumprod(factor)
equity_no_fee = INITIAL_CASH * np.cumprod(exit_prices / entry_prices)
entry_capital = np.concatenate(([INITIAL_CASH], equity[:-1]))
# Fees in dollars — proportional to capital at trade time.
entry_fees = entry_capital * FEE_PCT
exit_fees = entry_capital * (1 - FEE_PCT) * (exit_prices / entry_prices) * FEE_PCT
fees = entry_fees + exit_fees
cum_fees = np.cumsum(fees)
pnls = equity - entry_capital
pnl_pct = (factor - 1) * 100
total_trades = pnls.size
cum_winrate = np.cumsum(pnls > 0) / np.arange(1, total_trades + 1) * 100
pct_cum_pnl = np.cumsum(pnl_pct)
# Rolling (to-date) Sharpe after each trade: per-trade fractional returns,
# annualized by the observed trade frequency.
returns = factor - 1
with np.errstate(invalid="ignore", divide="ignore"):
cum_mean = np.cumsum(returns) / np.arange(1, total_trades + 1)
cum_var = (np.cumsum(returns**2) / np.arange(1, total_trades + 1)) - cum_mean**2
cum_std = np.sqrt(np.clip(cum_var, 0, None))
cum_sharpe_per_trade = np.where(cum_std > 0, cum_mean / cum_std, 0.0)
cum_sharpe = cum_sharpe_per_trade * np.sqrt(
np.arange(1, total_trades + 1) * bars_per_year / max(len(prices), 1)
)
return {
"symbol": symbol,
"prices": prices,
"wma": wma,
"opens": opens,
"closes": closes,
"pnl_pct": pnl_pct,
"cum_winrate": cum_winrate,
"pct_cum_pnl": pct_cum_pnl,
"equity": equity,
"equity_no_fee": equity_no_fee,
"cum_fees": cum_fees,
"cum_sharpe": cum_sharpe,
}
Visualization¶
This section visualises every metric over time, with one coloured line per symbol so the markets can be compared directly, shown across each timeframe. Each summary table below is also drawn as grouped bars — one panel per metric, one bar per symbol and timeframe — so the final results compare at a glance.
def charts(results):
symbols = list(dict.fromkeys(symbol for symbol, _ in results))
metrics = [
("Price Change %", "pct_prices", None),
("Trade P&L %", "pnl_pct", None),
("Cumulative Win Rate %", "cum_winrate", None),
("Cumulative P&L %", "pct_cum_pnl", None),
("Equity", "equity", "equity_no_fee"),
("Cumulative Fees", "cum_fees", None),
("Rolling Sharpe", "cum_sharpe", None),
]
n_rows = len(metrics)
n_cols = len(TIMEFRAMES)
col_width = 600
row_height = 280
gap_px = 60
total_w = col_width * n_cols + gap_px * max(n_cols - 1, 0)
total_h = row_height * n_rows
h_spacing = gap_px / total_w if n_cols > 1 else 0
fig = make_subplots(
rows=n_rows,
cols=n_cols,
shared_xaxes=True,
vertical_spacing=0.025,
horizontal_spacing=h_spacing,
column_titles=list(TIMEFRAMES),
)
for row_idx, (title, key, gross_key) in enumerate(metrics, start=1):
for col_idx, timeframe in enumerate(TIMEFRAMES, start=1):
max_len = max(len(results[(s, timeframe)]["prices"]) for s in symbols)
for symbol in symbols:
result = results[(symbol, timeframe)]
offset = max_len - len(result["prices"])
if key == "pct_prices":
y_values = (result["prices"] / result["prices"][0] - 1) * 100
x_values = np.arange(offset, offset + len(y_values))
else:
y_values = result[key]
x_values = result["opens"] + offset
fig.add_trace(
go.Scatter(
x=x_values,
y=y_values,
mode="lines",
name=symbol,
legendgroup=symbol,
line=dict(color=SYMBOL_COLORS[symbol], width=1),
showlegend=False,
hovertemplate="%{fullData.name}: %{y}<extra></extra>",
),
row=row_idx,
col=col_idx,
)
if gross_key:
fig.add_trace(
go.Scatter(
x=result["opens"] + offset,
y=result[gross_key],
mode="lines",
name=symbol,
legendgroup=symbol,
line=dict(color=SYMBOL_COLORS[symbol], width=1, dash="dot"),
showlegend=False,
hoverinfo="skip",
),
row=row_idx,
col=col_idx,
)
fig.update_yaxes(title_text=title, title_font=dict(size=13), row=row_idx, col=1)
# Dummy traces with thicker lines so the legend entries appear bold
# while the actual chart lines remain at width=1.
for symbol in symbols:
fig.add_trace(
go.Scatter(
x=[None],
y=[None],
mode="lines",
name=symbol,
legendgroup=symbol,
line=dict(color=SYMBOL_COLORS[symbol], width=3),
showlegend=True,
)
)
# Same pixel gap between the legend and the plot area in every figure —
# paper coordinates scale with figure height, so derive the offset from it.
legend_y = 1 + 75 / (total_h - 110) # 110 = top + bottom margins
fig.update_layout(
template="plotly_white",
height=total_h,
width=total_w,
font=dict(size=11),
hovermode="x unified",
hoverlabel=dict(bgcolor="white"),
legend=dict(
orientation="h",
yanchor="bottom",
y=legend_y,
xanchor="left",
x=0,
),
margin=dict(l=90, r=20, t=70, b=40),
)
fig.update_annotations(font=dict(size=13))
fig.update_xaxes(showgrid=True)
fig.update_yaxes(showgrid=True, zeroline=True)
fig.update_yaxes(range=[-3, 3], row=n_rows) # clamp Rolling Sharpe
fig.show()
def left_aligned_table(df):
def format_value(value):
if not isinstance(value, (int, float)):
return value
if value == int(value):
return f"{int(value):,}"
return f"{value:,.2f}".rstrip("0").rstrip(".")
return (
df.style.format(format_value)
.hide(axis="index")
.set_properties(**{"text-align": "left", "white-space": "nowrap"})
.set_table_styles(
[
{"selector": "th", "props": [("text-align", "left"), ("white-space", "nowrap")]},
{"selector": "", "props": [("min-width", "100%")]},
]
)
)
def summarize(results):
rows = []
for (symbol, timeframe), result in results.items():
equity = result["equity"]
equity_no_fee = result["equity_no_fee"]
prices = result["prices"]
has_trades = len(equity) > 0
rows.append(
{
"Symbol": symbol,
"Timeframe": timeframe,
"Price Change %": round(float(prices[-1] / prices[0] - 1) * 100, 1),
"Cumulative Win Rate %": (
round(float(result["cum_winrate"][-1]), 1) if has_trades else 0.0
),
"Cumulative P&L %": (
round(float(result["pct_cum_pnl"][-1]), 1) if has_trades else 0.0
),
"Equity Net": (round(float(equity[-1]), 2) if has_trades else float(INITIAL_CASH)),
"Equity Gross": (
round(float(equity_no_fee[-1]), 2) if has_trades else float(INITIAL_CASH)
),
"Cumulative Fees": (round(float(result["cum_fees"][-1]), 2) if has_trades else 0.0),
"Rolling Sharpe": (
round(float(result["cum_sharpe"][-1]), 2) if has_trades else 0.0
),
}
)
return pd.DataFrame(rows)
def summary_charts(summary):
symbols = list(dict.fromkeys(summary["Symbol"]))
metrics = [column for column in summary.columns if column not in ("Symbol", "Timeframe")]
n_cols = 4
n_rows = (len(metrics) + n_cols - 1) // n_cols
col_width = 600
row_height = 280
gap_px = 60
total_w = col_width * n_cols + gap_px * max(n_cols - 1, 0)
total_h = row_height * n_rows
h_spacing = gap_px / total_w if n_cols > 1 else 0
v_spacing = gap_px / total_h if n_rows > 1 else 0
fig = make_subplots(
rows=n_rows,
cols=n_cols,
vertical_spacing=v_spacing,
horizontal_spacing=h_spacing,
subplot_titles=metrics,
)
for metric_idx, metric in enumerate(metrics):
row_idx = metric_idx // n_cols + 1
col_idx = metric_idx % n_cols + 1
for symbol in symbols:
symbol_rows = summary[summary["Symbol"] == symbol]
fig.add_trace(
go.Bar(
x=symbol_rows["Timeframe"],
y=symbol_rows[metric],
name=symbol,
legendgroup=symbol,
marker_color=SYMBOL_COLORS[symbol],
showlegend=metric_idx == 0,
),
row=row_idx,
col=col_idx,
)
# Same pixel gap between the legend and the plot area in every figure —
# paper coordinates scale with figure height, so derive the offset from it.
legend_y = 1 + 75 / (total_h - 110) # 110 = top + bottom margins
fig.update_layout(
template="plotly_white",
height=total_h,
width=total_w,
font=dict(size=11),
barmode="group",
hovermode="x unified",
hoverlabel=dict(bgcolor="white"),
legend=dict(
orientation="h",
yanchor="bottom",
y=legend_y,
xanchor="left",
x=0,
),
margin=dict(l=90, r=20, t=70, b=40),
)
fig.update_annotations(font=dict(size=13))
fig.update_xaxes(showgrid=True)
fig.update_yaxes(showgrid=True, zeroline=True)
# Hide the axes of grid slots past the last metric so they stay blank.
for slot_idx in range(len(metrics), n_rows * n_cols):
row_idx = slot_idx // n_cols + 1
col_idx = slot_idx % n_cols + 1
fig.update_xaxes(visible=False, row=row_idx, col=col_idx)
fig.update_yaxes(visible=False, row=row_idx, col=col_idx)
fig.show()
Run on Real Data¶
This section runs the strategy on real market data: a basket of liquid symbols evaluated across several timeframes. Every metric is charted with one coloured line per symbol, so the markets can be compared directly.
The basket symbol ALL is an equal-weight portfolio of the whole set: every symbol's price history is normalized to its starting value over the common window, then averaged. It runs through the same strategy and metrics as any single market.
def basket_prices(symbol_prices):
common_len = min(len(prices) for prices in symbol_prices)
aligned = [prices[-common_len:] for prices in symbol_prices]
normalized = [prices / prices[0] for prices in aligned]
return np.mean(normalized, axis=0)
prices = {
(symbol, timeframe): pd.read_csv(f"{DATA_DIR}/{symbol}_{timeframe}.csv")["close"].to_numpy()
for symbol in SYMBOLS
for timeframe in TIMEFRAMES
}
for timeframe in TIMEFRAMES:
symbol_prices = [prices[(symbol, timeframe)] for symbol in SYMBOLS]
prices[(BASKET_SYMBOL, timeframe)] = basket_prices(symbol_prices=symbol_prices)
results = {
(symbol, timeframe): analytics(
symbol=symbol,
prices=prices[(symbol, timeframe)],
bars_per_year=BARS_PER_YEAR[timeframe],
)
for symbol in [*SYMBOLS, BASKET_SYMBOL]
for timeframe in TIMEFRAMES
}
charts(results=results)