Imports¶
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import talib
from plotly.subplots import make_subplots
Config¶
FEE_PCT = 0.02 / 100 # binance futures taker, per side
INITIAL_CASH = 100
WMA_FAST_PERIOD = 20
WMA_SLOW_PERIOD = 50
SYMBOLS = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "BNBUSDT", "DOGEUSDT"]
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,
}
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, and closes when it crosses back below:
$$\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).$$
It is long-only and always fully in or fully out — a bet that crossovers mark the start of sustained trends worth riding until momentum fades.
def strategy(prices, fast, slow):
wma = {
"fast": talib.WMA(prices, timeperiod=fast),
"slow": talib.WMA(prices, timeperiod=slow),
}
above = (wma["fast"] > wma["slow"]).astype(np.int8)
change = np.diff(above)
opens = np.where(change == 1)[0] + 1
closes = np.where(change == -1)[0] + 1
return wma, opens, closes
Strategy Example¶
An illustrative example on a synthetic price path, so the crossover entries (green) and exits (red) can be read against the two averages.
rng = np.random.default_rng(0)
example_prices = 100 * np.cumprod(1 + rng.normal(0, 0.01, size=300))
example_wma, example_opens, example_closes = strategy(
prices=example_prices,
fast=WMA_FAST_PERIOD,
slow=WMA_SLOW_PERIOD,
)
fig = go.Figure()
fig.add_trace(
go.Scatter(y=example_prices, mode="lines", name="price", line=dict(color="#888", width=1))
)
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(
x=example_opens,
y=example_prices[example_opens],
mode="markers",
name="entry",
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="exit",
marker=dict(color="#ff3b30", size=9, symbol="triangle-down"),
)
)
fig.update_layout(
template="plotly_white",
height=400,
width=900,
margin=dict(l=60, r=20, t=40, b=40),
)
fig.show()
Analytics Metrics¶
This section turns the strategy's trades into performance metrics, per-trade and cumulative. Each is defined with its formula below and 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).prices— the close-price series the run was computed on, $p_0,\dots,p_T$.wma— the fast and slow weighted moving averages from which the signals are derived.opens— bar indices where a position is opened (entry signals).closes— bar indices where a position is closed (exit signals).pnl_pct— 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.cum_winrate— running share of winning trades, $\frac{1}{n}\sum_{i=1}^{n}\mathbf{1}\{\text{pnl}_i > 0\}\cdot 100\%$.pct_cum_pnl— cumulative net P&L, the running sum ofpnl_pct.pct_cum_pnl_no_fee— the same cumulative P&L without fees (gross).equity— net equity curve, compounded all-in: $E_n = E_0 \prod_{i=1}^{n} r_i$.equity_no_fee— the gross equity curve (same compounding, fee term dropped).fees— dollar cost of each trade, proportional to capital at trade time.cum_fees— the running total of fees paid.sharpe— annualized Sharpe of net per-trade returns, $S = \frac{\bar x}{\operatorname{std}(x)}\sqrt{T_\text{year}}$ with $x_i = r_i - 1$.sharpe_no_fee— the same Sharpe computed on gross returns.cum_sharpe— the running (to-date) annualized Sharpe after each trade.
def analytics(symbol, prices, bars_per_year):
wma, opens, closes = strategy(prices=prices, fast=WMA_FAST_PERIOD, slow=WMA_SLOW_PERIOD)
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)
pnl_pct_no_fee = (exit_prices / entry_prices - 1) * 100
pct_cum_pnl_no_fee = np.cumsum(pnl_pct_no_fee)
# Sharpe: per-trade fractional returns annualized by the observed trade
# frequency. annual = (mean / std) * sqrt(trades_per_year).
returns = factor - 1
returns_no_fee = exit_prices / entry_prices - 1
sharpe = 0.0
sharpe_no_fee = 0.0
if total_trades > 1 and len(prices) > 0:
trades_per_year = total_trades * bars_per_year / len(prices)
sd = float(np.std(returns, ddof=1))
sd_no_fee = float(np.std(returns_no_fee, ddof=1))
if sd > 0:
sharpe = float(np.mean(returns)) / sd * np.sqrt(trades_per_year)
if sd_no_fee > 0:
sharpe_no_fee = float(np.mean(returns_no_fee)) / sd_no_fee * np.sqrt(trades_per_year)
# Cumulative (running) Sharpe — Sharpe-to-date after each trade.
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,
"pct_cum_pnl_no_fee": pct_cum_pnl_no_fee,
"equity": equity,
"equity_no_fee": equity_no_fee,
"fees": fees,
"cum_fees": cum_fees,
"sharpe": sharpe,
"sharpe_no_fee": sharpe_no_fee,
"cum_sharpe": cum_sharpe,
}
Charting¶
This section visualises every metric on a shared timeline, one line per metric, so the strategy's behaviour across the sequence of trades can be read at a glance: where equity grows, when the win rate stabilises, how fees accumulate, and how the rolling Sharpe ratio evolves.
def charts(results):
symbols_list = list(results.keys())
n_cols = len(symbols_list)
metric_titles = [
"pct_prices",
"pnl_pct",
"cum_winrate",
"pct_cum_pnl, pct_cum_pnl_no_fee",
"equity, equity_no_fee",
"fees",
"cum_fees",
"cum_sharpe",
]
subplot_titles = [f"{sym} — {title}" for title in metric_titles for sym in symbols_list]
col_width = 500
row_height = 170
gap_px = 60
total_w = col_width * n_cols + gap_px * max(n_cols - 1, 0)
total_h = row_height * 8
h_spacing = gap_px / total_w if n_cols > 1 else 0 # constant pixel gap
fig = make_subplots(
rows=8,
cols=n_cols,
shared_xaxes=True,
vertical_spacing=0.02,
horizontal_spacing=h_spacing,
subplot_titles=subplot_titles,
)
for col_idx, symbol in enumerate(symbols_list, start=1):
a = results[symbol]
prices = a["prices"]
opens = a["opens"]
pnl_pct = a["pnl_pct"]
cum_winrate = a["cum_winrate"]
pct_cum_pnl = a["pct_cum_pnl"]
pct_cum_pnl_no_fee = a["pct_cum_pnl_no_fee"]
equity = a["equity"]
equity_no_fee = a["equity_no_fee"]
fees = a["fees"]
cum_fees = a["cum_fees"]
cum_sharpe = a["cum_sharpe"]
pct_prices = (prices / prices[0] - 1) * 100
bar_width = max(len(prices) / 200, 5)
win_colors = np.where(pnl_pct > 0, "#00d4aa", "#ff3b30")
fig.add_trace(
go.Scatter(
y=pct_prices, mode="lines", name="pct_prices", line=dict(color="#888", width=1)
),
row=1,
col=col_idx,
)
fig.add_trace(
go.Bar(
x=opens,
y=pnl_pct,
name="pnl_pct",
marker=dict(color=win_colors, line=dict(width=0)),
width=bar_width,
),
row=2,
col=col_idx,
)
fig.add_trace(
go.Scatter(
x=opens,
y=cum_winrate,
mode="lines",
name="cum_winrate",
line=dict(color="#f1c40f", width=1),
),
row=3,
col=col_idx,
)
fig.add_hline(y=50, line=dict(color="#bbb", dash="dash"), row=3, col=col_idx)
fig.add_trace(
go.Scatter(
x=opens,
y=pct_cum_pnl_no_fee,
mode="lines",
name="pct_cum_pnl_no_fee",
line=dict(color="#7fb069", width=1, dash="dot"),
),
row=4,
col=col_idx,
)
fig.add_trace(
go.Scatter(
x=opens,
y=pct_cum_pnl,
mode="lines",
name="pct_cum_pnl",
line=dict(color="#00d4aa", width=1),
fill="tozeroy",
fillcolor="rgba(0,212,170,0.10)",
),
row=4,
col=col_idx,
)
fig.add_hline(y=0, line=dict(color="#bbb", dash="dash"), row=4, col=col_idx)
fig.add_trace(
go.Scatter(
x=opens,
y=equity_no_fee,
mode="lines",
name="equity_no_fee",
line=dict(color="#7fb069", width=1, dash="dot"),
),
row=5,
col=col_idx,
)
fig.add_trace(
go.Scatter(
x=opens,
y=equity,
mode="lines",
name="equity",
line=dict(color="#1f77b4", width=1),
fill="tozeroy",
fillcolor="rgba(31,119,180,0.10)",
),
row=5,
col=col_idx,
)
fig.add_hline(y=INITIAL_CASH, line=dict(color="#bbb", dash="dash"), row=5, col=col_idx)
fig.add_trace(
go.Bar(
x=opens,
y=fees,
name="fees",
marker=dict(color="#d35400", line=dict(width=0)),
width=bar_width,
),
row=6,
col=col_idx,
)
fig.add_trace(
go.Scatter(
x=opens,
y=cum_fees,
mode="lines",
name="cum_fees",
line=dict(color="#d35400", width=1),
fill="tozeroy",
fillcolor="rgba(211,84,0,0.15)",
),
row=7,
col=col_idx,
)
fig.add_trace(
go.Scatter(
x=opens,
y=cum_sharpe,
mode="lines",
name="cum_sharpe",
line=dict(color="#bb86fc", width=1),
fill="tozeroy",
fillcolor="rgba(187,134,252,0.12)",
),
row=8,
col=col_idx,
)
fig.add_hline(y=0, line=dict(color="#bbb", dash="dash"), row=8, col=col_idx)
fig.add_hline(y=1, line=dict(color="#ccc", dash="dot"), row=8, col=col_idx)
fig.update_layout(
template="plotly_white",
height=total_h,
width=total_w,
showlegend=False,
hovermode="x unified",
margin=dict(l=60, r=20, t=40, b=40),
bargap=0,
)
fig.update_xaxes(showgrid=True)
fig.update_yaxes(showgrid=True, zeroline=True)
fig.update_yaxes(range=[-3, 3], row=8)
fig.update_yaxes(showline=True, linewidth=1, row=2)
fig.update_yaxes(showline=True, linewidth=1, row=6)
for annotation in fig.layout.annotations:
annotation.update(font=dict(size=12))
fig.show()
Run on Real Data¶
This section runs the strategy on real market data: a basket of liquid symbols evaluated across several timeframes, each in its own subsection below, so its behaviour can be compared across both markets and horizons.
prices = {
(symbol, timeframe): pd.read_csv(f"{DATA_DIR}/{symbol}_{timeframe}.csv")["close"].to_numpy()
for symbol in SYMBOLS
for timeframe in TIMEFRAMES
}
30m¶
The strategy on the 30-minute timeframe across the full symbol basket.
results_30m = {
symbol: analytics(
symbol=symbol,
prices=prices[(symbol, "30m")],
bars_per_year=BARS_PER_YEAR["30m"],
)
for symbol in SYMBOLS
}
charts(results=results_30m)