Part IV
A
Appendix A
The Reference Strategy
The strategy specified in Chapter 5, implemented twice. Implementing the same specification in two independent systems is one of the better bug-detection methods available (Chapter 5), and any discrepancy between these two implementations is a bug in one of them rather than a platform quirk.
Both versions are deliberately mediocre strategies. They exist to be measured correctly, not to make money.
A.1 - Pine Script (TradingView)
Use for fast iteration and visual inspection. Note the known limitation, marked in the code: the position is unprotected on the bar it fills, because exit orders cannot be placed until the fill is visible to the script.
pine
//@version=5
strategy("Reference Strategy - Book Appendix A", overlay=true,
margin_long=100, margin_short=100, initial_capital=10000,
commission_type=strategy.commission.percent, commission_value=0.1,
slippage=2, pyramiding=0, calc_on_every_tick=false)
// ---------------- Inputs ----------------
grpMA = "Moving Averages"
fastLen = input.int(13, "Fast EMA", minval=1, group=grpMA)
slowLen = input.int(21, "Slow EMA", minval=1, group=grpMA)
trendLen = input.int(200, "Trend EMA", minval=1, group=grpMA)
useTrendFilter = input.bool(true, "Require price on trend-EMA side", group=grpMA)
grpRisk = "Risk"
atrLen = input.int(14, "ATR Length", minval=1, group=grpRisk)
atrMultSL = input.float(2.0, "ATR Stop Multiplier", minval=0.1, step=0.1, group=grpRisk)
atrMultTP = input.float(3.0, "ATR Target Multiplier", minval=0.1, step=0.1, group=grpRisk)
useRiskSizing = input.bool(true, "Size position by risk %", group=grpRisk)
riskPct = input.float(1.0, "Risk % of equity per trade", minval=0.01, maxval=100, step=0.1, group=grpRisk)
grpBE = "Breakeven"
useBreakeven = input.bool(false, "Enable breakeven stop", group=grpBE)
breakevenRR = input.float(1.0, "Arm breakeven after R multiple", minval=0.1, step=0.1, group=grpBE)
beOffsetR = input.float(0.05, "Breakeven offset in R (covers costs)", minval=0, step=0.01, group=grpBE)
grpDate = "Backtest Window"
useDateFilter = input.bool(false, "Limit backtest window", group=grpDate)
startDate = input.time(timestamp("01 Jan 2020 00:00 +0000"), "Start", group=grpDate)
endDate = input.time(timestamp("01 Jan 2024 00:00 +0000"), "End", group=grpDate)
// ---------------- Series ----------------
fastMA = ta.ema(close, fastLen)
slowMA = ta.ema(close, slowLen)
trendMA = ta.ema(close, trendLen)
atr = ta.atr(atrLen)
inWindow = not useDateFilter or (time >= startDate and time <= endDate)
warmedUp = not na(trendMA) and not na(atr)
trendOkLong = not useTrendFilter or close > trendMA
trendOkShort = not useTrendFilter or close < trendMA
longCond = warmedUp and inWindow and ta.crossover(fastMA, slowMA) and trendOkLong
shortCond = warmedUp and inWindow and ta.crossunder(fastMA, slowMA) and trendOkShort
// ---------------- Entries ----------------
riskPerUnit = atr * atrMultSL
qty = riskPerUnit > 0 ? (strategy.equity * riskPct / 100) / riskPerUnit : na
if longCond
if useRiskSizing and not na(qty)
strategy.entry("Long", strategy.long, qty=qty)
else
strategy.entry("Long", strategy.long)
if shortCond
if useRiskSizing and not na(qty)
strategy.entry("Short", strategy.short, qty=qty)
else
strategy.entry("Short", strategy.short)
// ---------------- Trade state ----------------
// ATR is frozen at fill so stop/target levels do not drift (Chapter 5).
var float entryAtr = na
var bool beArmed = false
var float curSL = na
var float curTP = na
prevPos = nz(strategy.position_size[1])
newTrade = strategy.position_size != 0 and
(prevPos == 0 or math.sign(strategy.position_size) != math.sign(prevPos))
if newTrade
entryAtr := nz(atr[1], atr) // ATR as of the signal bar
beArmed := false
if strategy.position_size == 0
entryAtr := na
beArmed := false
curSL := na
curTP := na
// ---------------- Exits ----------------
// One exit ID per side, re-issued each bar so the order is replaced, not duplicated.
if strategy.position_size != 0 and not na(entryAtr)
risk = entryAtr * atrMultSL
entry = strategy.position_avg_price
if strategy.position_size > 0
if useBreakeven and high >= entry + risk * breakevenRR
beArmed := true
initSL = entry - risk
curSL := beArmed ? math.max(initSL, entry + risk * beOffsetR) : initSL
curTP := entry + entryAtr * atrMultTP
strategy.exit("Exit Long", from_entry="Long", stop=curSL, limit=curTP)
else
if useBreakeven and low <= entry - risk * breakevenRR
beArmed := true
initSL = entry + risk
curSL := beArmed ? math.min(initSL, entry - risk * beOffsetR) : initSL
curTP := entry - entryAtr * atrMultTP
strategy.exit("Exit Short", from_entry="Short", stop=curSL, limit=curTP)
if useDateFilter and not inWindow and inWindow[1] and strategy.position_size != 0
strategy.close_all("Window End")
// ---------------- Plots ----------------
plot(fastMA, color=color.teal, title="Fast EMA")
plot(slowMA, color=color.orange, title="Slow EMA")
plot(trendMA, color=color.purple, title="Trend EMA")
plot(curSL, "Stop", color=color.new(color.red, 0), style=plot.style_linebr)
plot(curTP, "Target", color=color.new(color.green, 0), style=plot.style_linebr)
if barstate.islastconfirmedhistory
var label lbl = na
label.delete(lbl)
pf = strategy.grossloss > 0 ? strategy.grossprofit / strategy.grossloss : na
lbl := label.new(bar_index, high,
"Trades: " + str.tostring(strategy.closedtrades) +
"\nPF: " + str.tostring(pf, "#.##"),
style=label.style_label_left, color=color.new(color.blue, 20),
textcolor=color.white)A.2 - Python (event-driven)
The reference implementation. Slower than a vectorised version and correct by construction: the strategy receives a MarketView that cannot return future data (Chapter 7).
Two things this version does better than the Pine one. It checks the stop on the fill bar itself, closing the one-bar protection gap. And it resolves intrabar ambiguity pessimistically - when a bar contains both stop and target, the stop is assumed to have hit first.
python
"""
Reference event-driven backtester and strategy.
Companion to Trading Systems for Software Engineers, Chapters 5 and 7.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Protocol
import numpy as np
import pandas as pd
# ---------------------------------------------------------------- data types
@dataclass(frozen=True)
class Signal:
side: int # +1 long, -1 short
stop_distance: float # in price units, frozen at signal time
@dataclass
class Trade:
side: int
entry_time: pd.Timestamp
entry_price: float
qty: float
risk_per_unit: float
stop: float
target: float
exit_time: pd.Timestamp | None = None
exit_price: float | None = None
exit_reason: str | None = None
@property
def pnl(self) -> float:
if self.exit_price is None:
return 0.0
return self.side * (self.exit_price - self.entry_price) * self.qty
@property
def r_multiple(self) -> float:
"""P&L expressed in units of initial risk. The unit of account (Ch 4)."""
risk = self.risk_per_unit * self.qty
return self.pnl / risk if risk else 0.0
@dataclass
class CostModel:
commission_pct: float = 0.001 # 0.1% per side
slippage_pct: float = 0.0002 # 0.02% per side, applied adversely
def fill_price(self, quoted: float, side: int) -> float:
"""Slippage always works against you. Never model it as symmetric noise."""
return quoted * (1 + side * self.slippage_pct)
def commission(self, price: float, qty: float) -> float:
return abs(price * qty) * self.commission_pct
# ---------------------------------------------------------------- market view
class MarketView:
"""
A window onto history that physically cannot see the future.
The cursor is the index of the most recently CLOSED bar. There is
deliberately no method returning data beyond it, so lookahead bias
requires reaching around the interface rather than a one-character typo.
"""
__slots__ = ("_frame", "_cursor")
def __init__(self, frame: pd.DataFrame, cursor: int):
self._frame = frame
self._cursor = cursor
def latest(self, column: str) -> float:
return float(self._frame[column].iat[self._cursor])
def previous(self, column: str, back: int = 1) -> float:
idx = self._cursor - back
if idx < 0:
return float("nan")
return float(self._frame[column].iat[idx])
def history(self, column: str, n: int) -> np.ndarray:
start = max(0, self._cursor - n + 1)
return self._frame[column].values[start : self._cursor + 1]
@property
def time(self) -> pd.Timestamp:
return self._frame.index[self._cursor]
class Strategy(Protocol):
def on_bar(self, view: MarketView) -> Signal | None: ...
# ---------------------------------------------------------------- indicators
def prepare_indicators(df: pd.DataFrame, fast: int = 13, slow: int = 21,
trend: int = 200, atr_len: int = 14) -> pd.DataFrame:
"""
Precomputed for speed. This is safe: an EMA or ATR at index i depends
only on data at or before i, so no future information leaks backwards.
Precomputing anything centred or forward-looking would NOT be safe.
"""
out = df.copy()
out["ema_fast"] = out["close"].ewm(span=fast, adjust=False).mean()
out["ema_slow"] = out["close"].ewm(span=slow, adjust=False).mean()
out["ema_trend"] = out["close"].ewm(span=trend, adjust=False).mean()
prev_close = out["close"].shift(1)
true_range = pd.concat([
out["high"] - out["low"],
(out["high"] - prev_close).abs(),
(out["low"] - prev_close).abs(),
], axis=1).max(axis=1)
out["atr"] = true_range.ewm(alpha=1 / atr_len, adjust=False).mean()
# Warm-up mask: do not trade before the slowest series is meaningful.
out["warmed_up"] = np.arange(len(out)) >= max(trend, atr_len)
return out
# ---------------------------------------------------------------- strategy
@dataclass
class EmaCrossStrategy:
atr_mult_sl: float = 2.0
use_trend_filter: bool = True
def on_bar(self, view: MarketView) -> Signal | None:
if not view.latest("warmed_up"):
return None
fast_now, fast_prev = view.latest("ema_fast"), view.previous("ema_fast")
slow_now, slow_prev = view.latest("ema_slow"), view.previous("ema_slow")
if np.isnan(fast_prev) or np.isnan(slow_prev):
return None
crossed_up = fast_prev <= slow_prev and fast_now > slow_now
crossed_down = fast_prev >= slow_prev and fast_now < slow_now
close = view.latest("close")
trend = view.latest("ema_trend")
atr = view.latest("atr")
if atr <= 0 or np.isnan(atr):
return None
stop_distance = atr * self.atr_mult_sl
if crossed_up and (not self.use_trend_filter or close > trend):
return Signal(side=1, stop_distance=stop_distance)
if crossed_down and (not self.use_trend_filter or close < trend):
return Signal(side=-1, stop_distance=stop_distance)
return None
# ---------------------------------------------------------------- backtester
@dataclass
class BacktestResult:
trades: list[Trade] = field(default_factory=list)
equity_curve: pd.Series | None = None
initial_equity: float = 10_000.0
final_equity: float = 10_000.0
@property
def r_multiples(self) -> np.ndarray:
return np.array([t.r_multiple for t in self.trades if t.exit_price is not None])
def summary(self) -> dict:
r = self.r_multiples
if len(r) == 0:
return {"trades": 0}
wins, losses = r[r > 0], r[r <= 0]
gross_profit, gross_loss = wins.sum(), -losses.sum()
# Standard error of expectancy - see the sample-size table in Chapter 4.
stderr = r.std(ddof=1) / np.sqrt(len(r)) if len(r) > 1 else float("nan")
return {
"trades": len(r),
"expectancy_R": r.mean(),
"expectancy_stderr": stderr,
"t_stat": r.mean() / stderr if stderr else float("nan"),
"win_rate": len(wins) / len(r),
"avg_win_R": wins.mean() if len(wins) else 0.0,
"avg_loss_R": losses.mean() if len(losses) else 0.0,
"profit_factor": gross_profit / gross_loss if gross_loss > 0 else float("inf"),
"total_R": r.sum(),
"return_pct": (self.final_equity / self.initial_equity - 1) * 100,
}
def run_backtest(
strategy: Strategy,
df: pd.DataFrame,
costs: CostModel = CostModel(),
initial_equity: float = 10_000.0,
risk_pct: float = 1.0,
atr_mult_tp: float = 3.0,
pessimistic_intrabar: bool = True,
) -> BacktestResult:
"""
Event-driven simulation.
Bar ordering, which is where correctness lives:
1. Fill any pending entry at this bar's OPEN (signal came from the
previous bar's close -- you cannot act on a close at that close).
2. Check exits against this bar's range, including on the fill bar.
3. At this bar's CLOSE, ask the strategy for a new signal.
Intrabar ambiguity (Chapter 7): when a bar contains both stop and target,
`pessimistic_intrabar` assumes the stop hit first. The information needed
to do better is not present in the bar.
"""
result = BacktestResult(initial_equity=initial_equity)
equity = initial_equity
equity_points: list[float] = []
open_trade: Trade | None = None
pending: Signal | None = None
highs = df["high"].values
lows = df["low"].values
opens = df["open"].values
for i in range(len(df)):
timestamp = df.index[i]
# --- 1. Fill pending entry at this bar's open -----------------------
if pending is not None and open_trade is None:
entry = costs.fill_price(opens[i], pending.side)
risk_per_unit = pending.stop_distance
qty = (equity * risk_pct / 100) / risk_per_unit if risk_per_unit > 0 else 0.0
if qty > 0:
target_distance = risk_per_unit * (atr_mult_tp / 2.0) # TP:SL ratio
open_trade = Trade(
side=pending.side,
entry_time=timestamp,
entry_price=entry,
qty=qty,
risk_per_unit=risk_per_unit,
stop=entry - pending.side * risk_per_unit,
target=entry + pending.side * target_distance,
)
equity -= costs.commission(entry, qty)
pending = None
# --- 2. Check exits, including on the fill bar ----------------------
if open_trade is not None:
hit_stop = (
lows[i] <= open_trade.stop if open_trade.side > 0
else highs[i] >= open_trade.stop
)
hit_target = (
highs[i] >= open_trade.target if open_trade.side > 0
else lows[i] <= open_trade.target
)
exit_price = exit_reason = None
if hit_stop and hit_target:
if pessimistic_intrabar:
exit_price, exit_reason = open_trade.stop, "stop (ambiguous bar)"
else:
exit_price, exit_reason = open_trade.target, "target (ambiguous bar)"
elif hit_stop:
exit_price, exit_reason = open_trade.stop, "stop"
elif hit_target:
exit_price, exit_reason = open_trade.target, "target"
if exit_price is not None:
filled = costs.fill_price(exit_price, -open_trade.side)
open_trade.exit_time = timestamp
open_trade.exit_price = filled
open_trade.exit_reason = exit_reason
equity += open_trade.pnl - costs.commission(filled, open_trade.qty)
result.trades.append(open_trade)
open_trade = None
# --- 3. Generate a signal from this bar's close ---------------------
if open_trade is None and pending is None:
pending = strategy.on_bar(MarketView(df, i))
equity_points.append(equity)
result.equity_curve = pd.Series(equity_points, index=df.index)
result.final_equity = equity
return resultUsage, including the null test from Chapter 7:
python
import numpy as np
df = prepare_indicators(load_ohlcv("data/eurusd_1h.parquet"))
# Baseline. Record it, then resist tuning (Chapter 5).
result = run_backtest(EmaCrossStrategy(), df)
print(result.summary())
# Null test: random entries with the same exits and costs must lose money.
class RandomEntry:
def __init__(self, seed: int, rate: float = 0.02, atr_mult_sl: float = 2.0):
self.rng = np.random.default_rng(seed)
self.rate = rate
self.atr_mult_sl = atr_mult_sl
def on_bar(self, view: MarketView) -> Signal | None:
if not view.latest("warmed_up") or self.rng.random() > self.rate:
return None
atr = view.latest("atr")
if atr <= 0 or np.isnan(atr):
return None
side = 1 if self.rng.random() < 0.5 else -1
return Signal(side=side, stop_distance=atr * self.atr_mult_sl)
null_expectancies = [
run_backtest(RandomEntry(seed=s), df).summary().get("expectancy_R", 0.0)
for s in range(200)
]
assert np.mean(null_expectancies) < 0, "Harness is broken: random entries profit"