Backtesting exercises¶
Retrieve historical candles, implement example signals and examine simulated performance. The simulation uses simplified execution, cost and sizing rules; it does not reproduce OANDA margin, liquidity, order handling or account-specific restrictions. Parameter searches and resampled returns do not establish expected future performance.
Setup¶
Use a dedicated OANDA practice account and set both FIVETWENTY_OANDA_TOKEN and FIVETWENTY_OANDA_ACCOUNT in the kernel's environment. Review the configuration cell before making requests. Install the packages imported by the setup cell, select that Python environment as the Jupyter kernel, and run cells in order.
The repository's uv run poe docs-validate-notebooks command executes a temporary copy with synthetic HTTP responses. Running this notebook normally uses its configured API credentials; the offline validation result is not a live account test.
In [ ]:
Copied!
# Required imports
import os
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Any
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
# FiveTwenty imports
from fivetwenty import AsyncClient, Environment
from fivetwenty.models import CandlestickGranularity
# Configure plotting
plt.style.use("default")
sns.set_palette("husl")
%matplotlib inline
# Required imports
import os
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Any
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
# FiveTwenty imports
from fivetwenty import AsyncClient, Environment
from fivetwenty.models import CandlestickGranularity
# Configure plotting
plt.style.use("default")
sns.set_palette("husl")
%matplotlib inline
1. Configuration and Setup¶
First, let's set up our configuration and create the OANDA client.
In [ ]:
Copied!
# Configuration - Replace with your actual token
API_TOKEN = os.getenv("FIVETWENTY_OANDA_TOKEN", "your-oanda-api-token-here")
ENVIRONMENT = Environment.PRACTICE # Use PRACTICE for backtesting
ACCOUNT_ID = os.getenv("FIVETWENTY_OANDA_ACCOUNT", "your-account-id-here")
# Backtesting parameters
INSTRUMENT = "EUR_USD"
GRANULARITY = CandlestickGranularity.H1 # 1-hour candles
START_DATE = datetime(2023, 1, 1, tzinfo=timezone.utc)
END_DATE = datetime(2023, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
INITIAL_BALANCE = Decimal("10000") # Starting with $10,000
print("Backtesting Configuration:")
print(f"Instrument: {INSTRUMENT}")
print(f"Granularity: {GRANULARITY}")
print(f"Period: {START_DATE} to {END_DATE}")
print(f"Initial Balance: ${INITIAL_BALANCE}")
# Configuration - Replace with your actual token
API_TOKEN = os.getenv("FIVETWENTY_OANDA_TOKEN", "your-oanda-api-token-here")
ENVIRONMENT = Environment.PRACTICE # Use PRACTICE for backtesting
ACCOUNT_ID = os.getenv("FIVETWENTY_OANDA_ACCOUNT", "your-account-id-here")
# Backtesting parameters
INSTRUMENT = "EUR_USD"
GRANULARITY = CandlestickGranularity.H1 # 1-hour candles
START_DATE = datetime(2023, 1, 1, tzinfo=timezone.utc)
END_DATE = datetime(2023, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
INITIAL_BALANCE = Decimal("10000") # Starting with $10,000
print("Backtesting Configuration:")
print(f"Instrument: {INSTRUMENT}")
print(f"Granularity: {GRANULARITY}")
print(f"Period: {START_DATE} to {END_DATE}")
print(f"Initial Balance: ${INITIAL_BALANCE}")
2. Historical Data Retrieval¶
Let's create a function to fetch historical candlestick data from OANDA.
In [ ]:
Copied!
async def fetch_historical_data(client: AsyncClient, instrument: str, granularity: CandlestickGranularity, start_time: datetime, end_time: datetime) -> pd.DataFrame:
"""
Fetch historical candlestick data and convert to pandas DataFrame.
Args:
client: OANDA AsyncClient
instrument: Trading instrument (e.g., 'EUR_USD')
granularity: Candlestick granularity
start_time: Start of the range (timezone-aware datetime)
end_time: End of the range (timezone-aware datetime)
Returns:
DataFrame with OHLCV data
"""
print(f"Fetching historical data for {instrument}...")
# OANDA has a limit on candles per request (typically 5000)
# For longer periods, we need to make multiple requests
all_candles = []
current_start = start_time
while current_start < end_time:
try:
response = await client.instruments.get_instrument_candles(
instrument=instrument,
granularity=granularity,
from_time=current_start,
to_time=end_time,
include_first=current_start == start_time,
)
if not response["candles"]:
break
all_candles.extend(response["candles"])
# Update start time for next batch
current_start = response["candles"][-1].time
print(f"Fetched {len(response['candles'])} candles (total: {len(all_candles)})")
except Exception as e:
print(f"Error fetching data: {e}")
break
# Convert to DataFrame
data = []
for candle in all_candles:
if candle.mid: # Use mid prices for backtesting
data.append({"timestamp": pd.to_datetime(candle.time), "open": float(candle.mid.o), "high": float(candle.mid.h), "low": float(candle.mid.l), "close": float(candle.mid.c), "volume": candle.volume})
df = pd.DataFrame(data)
df.set_index("timestamp", inplace=True)
df.sort_index(inplace=True)
print(f"Historical data loaded: {len(df)} candles")
print(f"Date range: {df.index[0]} to {df.index[-1]}")
return df
async def fetch_historical_data(client: AsyncClient, instrument: str, granularity: CandlestickGranularity, start_time: datetime, end_time: datetime) -> pd.DataFrame:
"""
Fetch historical candlestick data and convert to pandas DataFrame.
Args:
client: OANDA AsyncClient
instrument: Trading instrument (e.g., 'EUR_USD')
granularity: Candlestick granularity
start_time: Start of the range (timezone-aware datetime)
end_time: End of the range (timezone-aware datetime)
Returns:
DataFrame with OHLCV data
"""
print(f"Fetching historical data for {instrument}...")
# OANDA has a limit on candles per request (typically 5000)
# For longer periods, we need to make multiple requests
all_candles = []
current_start = start_time
while current_start < end_time:
try:
response = await client.instruments.get_instrument_candles(
instrument=instrument,
granularity=granularity,
from_time=current_start,
to_time=end_time,
include_first=current_start == start_time,
)
if not response["candles"]:
break
all_candles.extend(response["candles"])
# Update start time for next batch
current_start = response["candles"][-1].time
print(f"Fetched {len(response['candles'])} candles (total: {len(all_candles)})")
except Exception as e:
print(f"Error fetching data: {e}")
break
# Convert to DataFrame
data = []
for candle in all_candles:
if candle.mid: # Use mid prices for backtesting
data.append({"timestamp": pd.to_datetime(candle.time), "open": float(candle.mid.o), "high": float(candle.mid.h), "low": float(candle.mid.l), "close": float(candle.mid.c), "volume": candle.volume})
df = pd.DataFrame(data)
df.set_index("timestamp", inplace=True)
df.sort_index(inplace=True)
print(f"Historical data loaded: {len(df)} candles")
print(f"Date range: {df.index[0]} to {df.index[-1]}")
return df
In [ ]:
Copied!
# Fetch the historical data
async with AsyncClient(token=API_TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
historical_data = await fetch_historical_data(client=client, instrument=INSTRUMENT, granularity=GRANULARITY, start_time=START_DATE, end_time=END_DATE)
# Display basic information about the data
print("\nData Summary:")
print(historical_data.describe())
# Plot the price data
plt.figure(figsize=(15, 8))
plt.plot(historical_data.index, historical_data["close"], label="Close Price", linewidth=0.8)
plt.title(f"{INSTRUMENT} Price Chart")
plt.xlabel("Date")
plt.ylabel("Price")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Fetch the historical data
async with AsyncClient(token=API_TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
historical_data = await fetch_historical_data(client=client, instrument=INSTRUMENT, granularity=GRANULARITY, start_time=START_DATE, end_time=END_DATE)
# Display basic information about the data
print("\nData Summary:")
print(historical_data.describe())
# Plot the price data
plt.figure(figsize=(15, 8))
plt.plot(historical_data.index, historical_data["close"], label="Close Price", linewidth=0.8)
plt.title(f"{INSTRUMENT} Price Chart")
plt.xlabel("Date")
plt.ylabel("Price")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
3. Technical Indicators¶
Let's implement some common technical indicators for our strategies.
In [ ]:
Copied!
def calculate_sma(data: pd.Series, window: int) -> pd.Series:
"""Calculate Simple Moving Average."""
return data.rolling(window=window).mean()
def calculate_ema(data: pd.Series, window: int) -> pd.Series:
"""Calculate Exponential Moving Average."""
return data.ewm(span=window, adjust=False).mean()
def calculate_rsi(data: pd.Series, window: int = 14) -> pd.Series:
"""Calculate Relative Strength Index."""
delta = data.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def calculate_bollinger_bands(data: pd.Series, window: int = 20, num_std: float = 2) -> tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate Bollinger Bands."""
sma = calculate_sma(data, window)
std = data.rolling(window=window).std()
upper_band = sma + (std * num_std)
lower_band = sma - (std * num_std)
return upper_band, sma, lower_band
def calculate_macd(data: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate MACD (Moving Average Convergence Divergence)."""
ema_fast = calculate_ema(data, fast)
ema_slow = calculate_ema(data, slow)
macd_line = ema_fast - ema_slow
signal_line = calculate_ema(macd_line, signal)
histogram = macd_line - signal_line
return macd_line, signal_line, histogram
# Calculate technical indicators for our dataset
print("Calculating technical indicators...")
# Moving averages
historical_data["sma_20"] = calculate_sma(historical_data["close"], 20)
historical_data["sma_50"] = calculate_sma(historical_data["close"], 50)
historical_data["ema_20"] = calculate_ema(historical_data["close"], 20)
# RSI
historical_data["rsi"] = calculate_rsi(historical_data["close"])
# Bollinger Bands
bb_upper, bb_middle, bb_lower = calculate_bollinger_bands(historical_data["close"])
historical_data["bb_upper"] = bb_upper
historical_data["bb_middle"] = bb_middle
historical_data["bb_lower"] = bb_lower
# MACD
macd_line, macd_signal, macd_histogram = calculate_macd(historical_data["close"])
historical_data["macd"] = macd_line
historical_data["macd_signal"] = macd_signal
historical_data["macd_histogram"] = macd_histogram
print("Technical indicators calculated successfully!")
print(f"Data shape: {historical_data.shape}")
def calculate_sma(data: pd.Series, window: int) -> pd.Series:
"""Calculate Simple Moving Average."""
return data.rolling(window=window).mean()
def calculate_ema(data: pd.Series, window: int) -> pd.Series:
"""Calculate Exponential Moving Average."""
return data.ewm(span=window, adjust=False).mean()
def calculate_rsi(data: pd.Series, window: int = 14) -> pd.Series:
"""Calculate Relative Strength Index."""
delta = data.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def calculate_bollinger_bands(data: pd.Series, window: int = 20, num_std: float = 2) -> tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate Bollinger Bands."""
sma = calculate_sma(data, window)
std = data.rolling(window=window).std()
upper_band = sma + (std * num_std)
lower_band = sma - (std * num_std)
return upper_band, sma, lower_band
def calculate_macd(data: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate MACD (Moving Average Convergence Divergence)."""
ema_fast = calculate_ema(data, fast)
ema_slow = calculate_ema(data, slow)
macd_line = ema_fast - ema_slow
signal_line = calculate_ema(macd_line, signal)
histogram = macd_line - signal_line
return macd_line, signal_line, histogram
# Calculate technical indicators for our dataset
print("Calculating technical indicators...")
# Moving averages
historical_data["sma_20"] = calculate_sma(historical_data["close"], 20)
historical_data["sma_50"] = calculate_sma(historical_data["close"], 50)
historical_data["ema_20"] = calculate_ema(historical_data["close"], 20)
# RSI
historical_data["rsi"] = calculate_rsi(historical_data["close"])
# Bollinger Bands
bb_upper, bb_middle, bb_lower = calculate_bollinger_bands(historical_data["close"])
historical_data["bb_upper"] = bb_upper
historical_data["bb_middle"] = bb_middle
historical_data["bb_lower"] = bb_lower
# MACD
macd_line, macd_signal, macd_histogram = calculate_macd(historical_data["close"])
historical_data["macd"] = macd_line
historical_data["macd_signal"] = macd_signal
historical_data["macd_histogram"] = macd_histogram
print("Technical indicators calculated successfully!")
print(f"Data shape: {historical_data.shape}")
4. Strategy Framework¶
Let's create a flexible framework for implementing trading strategies.
In [ ]:
Copied!
class Signal(Enum):
"""Trading signals."""
BUY = 1
SELL = -1
HOLD = 0
@dataclass
class Trade:
"""Represents a single trade."""
timestamp: pd.Timestamp
signal: Signal
price: float
units: int
stop_loss: float | None = None
take_profit: float | None = None
@dataclass
class BacktestPosition:
"""Represents current position."""
units: int = 0
average_price: float = 0.0
unrealized_pnl: float = 0.0
realized_pnl: float = 0.0
class Strategy:
"""Base class for trading strategies."""
def __init__(self, name: str):
self.name = name
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
"""Generate trading signal based on current market data.
Args:
data: Historical price data with indicators
index: Current bar index
Returns:
Trading signal (BUY, SELL, or HOLD)
"""
raise NotImplementedError("Subclasses must implement generate_signal")
def calculate_position_size(self, data: pd.DataFrame, index: int, balance: float) -> int:
"""Calculate position size for the trade.
Args:
data: Historical price data
index: Current bar index
balance: Current account balance
Returns:
Number of units to trade
"""
# Default: risk 2% of balance per trade
risk_per_trade = 0.02
price = data.iloc[index]["close"]
units = int((balance * risk_per_trade) / price)
return max(1000, units) # Minimum position size
class MovingAverageCrossover(Strategy):
"""Simple moving average crossover strategy."""
def __init__(self, fast_period: int = 20, slow_period: int = 50):
super().__init__(f"MA_Crossover_{fast_period}_{slow_period}")
self.fast_period = fast_period
self.slow_period = slow_period
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
if index < self.slow_period:
return Signal.HOLD
current_fast = data.iloc[index][f"sma_{self.fast_period}"]
current_slow = data.iloc[index][f"sma_{self.slow_period}"]
previous_fast = data.iloc[index - 1][f"sma_{self.fast_period}"]
previous_slow = data.iloc[index - 1][f"sma_{self.slow_period}"]
# Check for crossover
if previous_fast <= previous_slow and current_fast > current_slow:
return Signal.BUY
if previous_fast >= previous_slow and current_fast < current_slow:
return Signal.SELL
return Signal.HOLD
class RSIMeanReversion(Strategy):
"""RSI-based mean reversion strategy."""
def __init__(self, rsi_period: int = 14, oversold: float = 30, overbought: float = 70):
super().__init__(f"RSI_MeanReversion_{rsi_period}_{oversold}_{overbought}")
self.rsi_period = rsi_period
self.oversold = oversold
self.overbought = overbought
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
if index < self.rsi_period:
return Signal.HOLD
current_rsi = data.iloc[index]["rsi"]
previous_rsi = data.iloc[index - 1]["rsi"]
# Buy when RSI crosses above oversold level
if previous_rsi <= self.oversold and current_rsi > self.oversold:
return Signal.BUY
# Sell when RSI crosses below overbought level
if previous_rsi >= self.overbought and current_rsi < self.overbought:
return Signal.SELL
return Signal.HOLD
class BollingerBandStrategy(Strategy):
"""Bollinger Band breakout strategy."""
def __init__(self, bb_period: int = 20, bb_std: float = 2.0):
super().__init__(f"BollingerBand_{bb_period}_{bb_std}")
self.bb_period = bb_period
self.bb_std = bb_std
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
if index < self.bb_period:
return Signal.HOLD
current_price = data.iloc[index]["close"]
previous_price = data.iloc[index - 1]["close"]
bb_upper = data.iloc[index]["bb_upper"]
bb_lower = data.iloc[index]["bb_lower"]
prev_bb_upper = data.iloc[index - 1]["bb_upper"]
prev_bb_lower = data.iloc[index - 1]["bb_lower"]
# Buy on breakout above upper band
if previous_price <= prev_bb_upper and current_price > bb_upper:
return Signal.BUY
# Sell on breakdown below lower band
if previous_price >= prev_bb_lower and current_price < bb_lower:
return Signal.SELL
return Signal.HOLD
print("Strategy framework implemented successfully!")
class Signal(Enum):
"""Trading signals."""
BUY = 1
SELL = -1
HOLD = 0
@dataclass
class Trade:
"""Represents a single trade."""
timestamp: pd.Timestamp
signal: Signal
price: float
units: int
stop_loss: float | None = None
take_profit: float | None = None
@dataclass
class BacktestPosition:
"""Represents current position."""
units: int = 0
average_price: float = 0.0
unrealized_pnl: float = 0.0
realized_pnl: float = 0.0
class Strategy:
"""Base class for trading strategies."""
def __init__(self, name: str):
self.name = name
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
"""Generate trading signal based on current market data.
Args:
data: Historical price data with indicators
index: Current bar index
Returns:
Trading signal (BUY, SELL, or HOLD)
"""
raise NotImplementedError("Subclasses must implement generate_signal")
def calculate_position_size(self, data: pd.DataFrame, index: int, balance: float) -> int:
"""Calculate position size for the trade.
Args:
data: Historical price data
index: Current bar index
balance: Current account balance
Returns:
Number of units to trade
"""
# Default: risk 2% of balance per trade
risk_per_trade = 0.02
price = data.iloc[index]["close"]
units = int((balance * risk_per_trade) / price)
return max(1000, units) # Minimum position size
class MovingAverageCrossover(Strategy):
"""Simple moving average crossover strategy."""
def __init__(self, fast_period: int = 20, slow_period: int = 50):
super().__init__(f"MA_Crossover_{fast_period}_{slow_period}")
self.fast_period = fast_period
self.slow_period = slow_period
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
if index < self.slow_period:
return Signal.HOLD
current_fast = data.iloc[index][f"sma_{self.fast_period}"]
current_slow = data.iloc[index][f"sma_{self.slow_period}"]
previous_fast = data.iloc[index - 1][f"sma_{self.fast_period}"]
previous_slow = data.iloc[index - 1][f"sma_{self.slow_period}"]
# Check for crossover
if previous_fast <= previous_slow and current_fast > current_slow:
return Signal.BUY
if previous_fast >= previous_slow and current_fast < current_slow:
return Signal.SELL
return Signal.HOLD
class RSIMeanReversion(Strategy):
"""RSI-based mean reversion strategy."""
def __init__(self, rsi_period: int = 14, oversold: float = 30, overbought: float = 70):
super().__init__(f"RSI_MeanReversion_{rsi_period}_{oversold}_{overbought}")
self.rsi_period = rsi_period
self.oversold = oversold
self.overbought = overbought
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
if index < self.rsi_period:
return Signal.HOLD
current_rsi = data.iloc[index]["rsi"]
previous_rsi = data.iloc[index - 1]["rsi"]
# Buy when RSI crosses above oversold level
if previous_rsi <= self.oversold and current_rsi > self.oversold:
return Signal.BUY
# Sell when RSI crosses below overbought level
if previous_rsi >= self.overbought and current_rsi < self.overbought:
return Signal.SELL
return Signal.HOLD
class BollingerBandStrategy(Strategy):
"""Bollinger Band breakout strategy."""
def __init__(self, bb_period: int = 20, bb_std: float = 2.0):
super().__init__(f"BollingerBand_{bb_period}_{bb_std}")
self.bb_period = bb_period
self.bb_std = bb_std
def generate_signal(self, data: pd.DataFrame, index: int) -> Signal:
if index < self.bb_period:
return Signal.HOLD
current_price = data.iloc[index]["close"]
previous_price = data.iloc[index - 1]["close"]
bb_upper = data.iloc[index]["bb_upper"]
bb_lower = data.iloc[index]["bb_lower"]
prev_bb_upper = data.iloc[index - 1]["bb_upper"]
prev_bb_lower = data.iloc[index - 1]["bb_lower"]
# Buy on breakout above upper band
if previous_price <= prev_bb_upper and current_price > bb_upper:
return Signal.BUY
# Sell on breakdown below lower band
if previous_price >= prev_bb_lower and current_price < bb_lower:
return Signal.SELL
return Signal.HOLD
print("Strategy framework implemented successfully!")
5. Backtesting Engine¶
The following engine implements a simplified local simulation; it is not an OANDA execution simulator.
In [ ]:
Copied!
class BacktestEngine:
"""Comprehensive backtesting engine."""
def __init__(
self,
initial_balance: float = 10000,
commission_rate: float = 0.0001, # 1 basis point
slippage: float = 0.0001, # 1 basis point
leverage: float = 1.0,
):
self.initial_balance = initial_balance
self.commission_rate = commission_rate
self.slippage = slippage
self.leverage = leverage
# Reset for each backtest
self.reset()
def reset(self):
"""Reset the engine for a new backtest."""
self.balance = self.initial_balance
self.equity_curve = []
self.trades = []
self.positions = BacktestPosition()
self.total_commission = 0.0
self.total_slippage = 0.0
def calculate_commission(self, units: int, price: float) -> float:
"""Calculate commission for a trade."""
return abs(units) * price * self.commission_rate
def apply_slippage(self, price: float, signal: Signal) -> float:
"""Apply slippage to trade price."""
if signal == Signal.BUY:
return price * (1 + self.slippage)
if signal == Signal.SELL:
return price * (1 - self.slippage)
return price
def execute_trade(self, signal: Signal, price: float, units: int, timestamp: pd.Timestamp):
"""Execute a trade and update positions."""
if signal == Signal.HOLD:
return
# Apply slippage
execution_price = self.apply_slippage(price, signal)
# Calculate commission
commission = self.calculate_commission(units, execution_price)
self.total_commission += commission
# Track slippage cost
slippage_cost = abs(units) * abs(execution_price - price)
self.total_slippage += slippage_cost
# Determine trade direction and units
if signal == Signal.BUY:
trade_units = units
else: # SELL
trade_units = -units
# Update position
if self.positions.units == 0:
# New position
self.positions.units = trade_units
self.positions.average_price = execution_price
elif (self.positions.units > 0 and trade_units > 0) or (self.positions.units < 0 and trade_units < 0):
# Add to existing position
total_value = (self.positions.units * self.positions.average_price) + (trade_units * execution_price)
self.positions.units += trade_units
self.positions.average_price = total_value / self.positions.units
# Partial or full close
elif abs(trade_units) >= abs(self.positions.units):
# Full close (or reverse)
pnl = self.positions.units * (execution_price - self.positions.average_price)
self.positions.realized_pnl += pnl
self.balance += pnl
# Check if reversing position
remaining_units = trade_units + self.positions.units
if remaining_units != 0:
self.positions.units = remaining_units
self.positions.average_price = execution_price
else:
self.positions.units = 0
self.positions.average_price = 0.0
else:
# Partial close
close_units = -trade_units
pnl = close_units * (execution_price - self.positions.average_price)
self.positions.realized_pnl += pnl
self.balance += pnl
self.positions.units += trade_units
# Deduct commission from balance
self.balance -= commission
# Record the trade
trade = Trade(timestamp=timestamp, signal=signal, price=execution_price, units=trade_units)
self.trades.append(trade)
def update_unrealized_pnl(self, current_price: float):
"""Update unrealized P&L based on current market price."""
if self.positions.units != 0:
self.positions.unrealized_pnl = self.positions.units * (current_price - self.positions.average_price)
else:
self.positions.unrealized_pnl = 0.0
def get_total_equity(self) -> float:
"""Calculate total account equity (balance + unrealized P&L)."""
return self.balance + self.positions.unrealized_pnl
def run_backtest(self, data: pd.DataFrame, strategy: Strategy) -> dict[str, Any]:
"""Run backtest on historical data using specified strategy."""
self.reset()
print(f"Running backtest for strategy: {strategy.name}")
print(f"Data period: {data.index[0]} to {data.index[-1]}")
print(f"Total bars: {len(data)}")
# Iterate through historical data
for i in range(len(data)):
current_bar = data.iloc[i]
timestamp = data.index[i]
price = current_bar["close"]
# Generate trading signal
signal = strategy.generate_signal(data, i)
# Execute trade if signal is not HOLD
if signal != Signal.HOLD:
units = strategy.calculate_position_size(data, i, self.balance)
self.execute_trade(signal, price, units, timestamp)
# Update unrealized P&L
self.update_unrealized_pnl(price)
# Record equity curve
equity = self.get_total_equity()
self.equity_curve.append({"timestamp": timestamp, "equity": equity, "balance": self.balance, "unrealized_pnl": self.positions.unrealized_pnl, "position_units": self.positions.units})
# Close any remaining position at final price
if self.positions.units != 0:
final_price = data.iloc[-1]["close"]
final_signal = Signal.SELL if self.positions.units > 0 else Signal.BUY
self.execute_trade(final_signal, final_price, abs(self.positions.units), data.index[-1])
print(f"Backtest completed. Total trades: {len(self.trades)}")
return self.calculate_performance_metrics()
def calculate_performance_metrics(self) -> dict[str, Any]:
"""Calculate comprehensive performance metrics."""
if not self.equity_curve:
return {}
# Convert equity curve to DataFrame for easier analysis
equity_df = pd.DataFrame(self.equity_curve)
equity_df.set_index("timestamp", inplace=True)
# Calculate returns
equity_df["returns"] = equity_df["equity"].pct_change()
total_return = (equity_df["equity"].iloc[-1] / self.initial_balance - 1) * 100
# Calculate drawdown
equity_df["cummax"] = equity_df["equity"].cummax()
equity_df["drawdown"] = (equity_df["equity"] / equity_df["cummax"] - 1) * 100
max_drawdown = equity_df["drawdown"].min()
# Risk metrics
returns = equity_df["returns"].dropna()
volatility = returns.std() * np.sqrt(252 * 24) * 100 # Annualized volatility for hourly data
sharpe_ratio = (returns.mean() * 252 * 24) / (returns.std() * np.sqrt(252 * 24)) if returns.std() > 0 else 0
# Trade statistics
winning_trades = [t for t in self.trades if self._calculate_trade_pnl(t) > 0]
losing_trades = [t for t in self.trades if self._calculate_trade_pnl(t) < 0]
win_rate = len(winning_trades) / len(self.trades) * 100 if self.trades else 0
avg_win = np.mean([self._calculate_trade_pnl(t) for t in winning_trades]) if winning_trades else 0
avg_loss = np.mean([self._calculate_trade_pnl(t) for t in losing_trades]) if losing_trades else 0
profit_factor = abs(avg_win * len(winning_trades)) / abs(avg_loss * len(losing_trades)) if losing_trades and avg_loss != 0 else float("inf")
metrics = {
"initial_balance": self.initial_balance,
"final_balance": equity_df["equity"].iloc[-1],
"total_return_pct": total_return,
"max_drawdown_pct": max_drawdown,
"volatility_pct": volatility,
"sharpe_ratio": sharpe_ratio,
"total_trades": len(self.trades),
"winning_trades": len(winning_trades),
"losing_trades": len(losing_trades),
"win_rate_pct": win_rate,
"avg_win": avg_win,
"avg_loss": avg_loss,
"profit_factor": profit_factor,
"total_commission": self.total_commission,
"total_slippage": self.total_slippage,
"equity_curve": equity_df,
"trades": self.trades,
}
return metrics
def _calculate_trade_pnl(self, trade: Trade) -> float:
"""Calculate P&L for a single trade (simplified for demonstration)."""
# This is a simplified calculation - in practice you'd track entry/exit pairs
return 0 # Placeholder
print("Backtesting engine implemented successfully!")
class BacktestEngine:
"""Comprehensive backtesting engine."""
def __init__(
self,
initial_balance: float = 10000,
commission_rate: float = 0.0001, # 1 basis point
slippage: float = 0.0001, # 1 basis point
leverage: float = 1.0,
):
self.initial_balance = initial_balance
self.commission_rate = commission_rate
self.slippage = slippage
self.leverage = leverage
# Reset for each backtest
self.reset()
def reset(self):
"""Reset the engine for a new backtest."""
self.balance = self.initial_balance
self.equity_curve = []
self.trades = []
self.positions = BacktestPosition()
self.total_commission = 0.0
self.total_slippage = 0.0
def calculate_commission(self, units: int, price: float) -> float:
"""Calculate commission for a trade."""
return abs(units) * price * self.commission_rate
def apply_slippage(self, price: float, signal: Signal) -> float:
"""Apply slippage to trade price."""
if signal == Signal.BUY:
return price * (1 + self.slippage)
if signal == Signal.SELL:
return price * (1 - self.slippage)
return price
def execute_trade(self, signal: Signal, price: float, units: int, timestamp: pd.Timestamp):
"""Execute a trade and update positions."""
if signal == Signal.HOLD:
return
# Apply slippage
execution_price = self.apply_slippage(price, signal)
# Calculate commission
commission = self.calculate_commission(units, execution_price)
self.total_commission += commission
# Track slippage cost
slippage_cost = abs(units) * abs(execution_price - price)
self.total_slippage += slippage_cost
# Determine trade direction and units
if signal == Signal.BUY:
trade_units = units
else: # SELL
trade_units = -units
# Update position
if self.positions.units == 0:
# New position
self.positions.units = trade_units
self.positions.average_price = execution_price
elif (self.positions.units > 0 and trade_units > 0) or (self.positions.units < 0 and trade_units < 0):
# Add to existing position
total_value = (self.positions.units * self.positions.average_price) + (trade_units * execution_price)
self.positions.units += trade_units
self.positions.average_price = total_value / self.positions.units
# Partial or full close
elif abs(trade_units) >= abs(self.positions.units):
# Full close (or reverse)
pnl = self.positions.units * (execution_price - self.positions.average_price)
self.positions.realized_pnl += pnl
self.balance += pnl
# Check if reversing position
remaining_units = trade_units + self.positions.units
if remaining_units != 0:
self.positions.units = remaining_units
self.positions.average_price = execution_price
else:
self.positions.units = 0
self.positions.average_price = 0.0
else:
# Partial close
close_units = -trade_units
pnl = close_units * (execution_price - self.positions.average_price)
self.positions.realized_pnl += pnl
self.balance += pnl
self.positions.units += trade_units
# Deduct commission from balance
self.balance -= commission
# Record the trade
trade = Trade(timestamp=timestamp, signal=signal, price=execution_price, units=trade_units)
self.trades.append(trade)
def update_unrealized_pnl(self, current_price: float):
"""Update unrealized P&L based on current market price."""
if self.positions.units != 0:
self.positions.unrealized_pnl = self.positions.units * (current_price - self.positions.average_price)
else:
self.positions.unrealized_pnl = 0.0
def get_total_equity(self) -> float:
"""Calculate total account equity (balance + unrealized P&L)."""
return self.balance + self.positions.unrealized_pnl
def run_backtest(self, data: pd.DataFrame, strategy: Strategy) -> dict[str, Any]:
"""Run backtest on historical data using specified strategy."""
self.reset()
print(f"Running backtest for strategy: {strategy.name}")
print(f"Data period: {data.index[0]} to {data.index[-1]}")
print(f"Total bars: {len(data)}")
# Iterate through historical data
for i in range(len(data)):
current_bar = data.iloc[i]
timestamp = data.index[i]
price = current_bar["close"]
# Generate trading signal
signal = strategy.generate_signal(data, i)
# Execute trade if signal is not HOLD
if signal != Signal.HOLD:
units = strategy.calculate_position_size(data, i, self.balance)
self.execute_trade(signal, price, units, timestamp)
# Update unrealized P&L
self.update_unrealized_pnl(price)
# Record equity curve
equity = self.get_total_equity()
self.equity_curve.append({"timestamp": timestamp, "equity": equity, "balance": self.balance, "unrealized_pnl": self.positions.unrealized_pnl, "position_units": self.positions.units})
# Close any remaining position at final price
if self.positions.units != 0:
final_price = data.iloc[-1]["close"]
final_signal = Signal.SELL if self.positions.units > 0 else Signal.BUY
self.execute_trade(final_signal, final_price, abs(self.positions.units), data.index[-1])
print(f"Backtest completed. Total trades: {len(self.trades)}")
return self.calculate_performance_metrics()
def calculate_performance_metrics(self) -> dict[str, Any]:
"""Calculate comprehensive performance metrics."""
if not self.equity_curve:
return {}
# Convert equity curve to DataFrame for easier analysis
equity_df = pd.DataFrame(self.equity_curve)
equity_df.set_index("timestamp", inplace=True)
# Calculate returns
equity_df["returns"] = equity_df["equity"].pct_change()
total_return = (equity_df["equity"].iloc[-1] / self.initial_balance - 1) * 100
# Calculate drawdown
equity_df["cummax"] = equity_df["equity"].cummax()
equity_df["drawdown"] = (equity_df["equity"] / equity_df["cummax"] - 1) * 100
max_drawdown = equity_df["drawdown"].min()
# Risk metrics
returns = equity_df["returns"].dropna()
volatility = returns.std() * np.sqrt(252 * 24) * 100 # Annualized volatility for hourly data
sharpe_ratio = (returns.mean() * 252 * 24) / (returns.std() * np.sqrt(252 * 24)) if returns.std() > 0 else 0
# Trade statistics
winning_trades = [t for t in self.trades if self._calculate_trade_pnl(t) > 0]
losing_trades = [t for t in self.trades if self._calculate_trade_pnl(t) < 0]
win_rate = len(winning_trades) / len(self.trades) * 100 if self.trades else 0
avg_win = np.mean([self._calculate_trade_pnl(t) for t in winning_trades]) if winning_trades else 0
avg_loss = np.mean([self._calculate_trade_pnl(t) for t in losing_trades]) if losing_trades else 0
profit_factor = abs(avg_win * len(winning_trades)) / abs(avg_loss * len(losing_trades)) if losing_trades and avg_loss != 0 else float("inf")
metrics = {
"initial_balance": self.initial_balance,
"final_balance": equity_df["equity"].iloc[-1],
"total_return_pct": total_return,
"max_drawdown_pct": max_drawdown,
"volatility_pct": volatility,
"sharpe_ratio": sharpe_ratio,
"total_trades": len(self.trades),
"winning_trades": len(winning_trades),
"losing_trades": len(losing_trades),
"win_rate_pct": win_rate,
"avg_win": avg_win,
"avg_loss": avg_loss,
"profit_factor": profit_factor,
"total_commission": self.total_commission,
"total_slippage": self.total_slippage,
"equity_curve": equity_df,
"trades": self.trades,
}
return metrics
def _calculate_trade_pnl(self, trade: Trade) -> float:
"""Calculate P&L for a single trade (simplified for demonstration)."""
# This is a simplified calculation - in practice you'd track entry/exit pairs
return 0 # Placeholder
print("Backtesting engine implemented successfully!")
6. Running Backtests¶
Now let's run backtests for our different strategies and compare their performance.
In [ ]:
Copied!
# Create backtesting engine
engine = BacktestEngine(
initial_balance=float(INITIAL_BALANCE),
commission_rate=0.0002, # 2 basis points
slippage=0.0001, # 1 basis point
leverage=1.0,
)
# Define strategies to test
strategies = [MovingAverageCrossover(fast_period=20, slow_period=50), RSIMeanReversion(rsi_period=14, oversold=30, overbought=70), BollingerBandStrategy(bb_period=20, bb_std=2.0)]
# Run backtests
results = {}
for strategy in strategies:
print(f"\n{'=' * 50}")
metrics = engine.run_backtest(historical_data, strategy)
results[strategy.name] = metrics
# Display key metrics
print(f"\nStrategy: {strategy.name}")
print(f"Total Return: {metrics['total_return_pct']:.2f}%")
print(f"Max Drawdown: {metrics['max_drawdown_pct']:.2f}%")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f"Total Trades: {metrics['total_trades']}")
print(f"Win Rate: {metrics['win_rate_pct']:.1f}%")
print(f"Profit Factor: {metrics['profit_factor']:.2f}")
print(f"Total Commission: ${metrics['total_commission']:.2f}")
print(f"Total Slippage: ${metrics['total_slippage']:.2f}")
print(f"\n{'=' * 50}")
print("All backtests completed!")
# Create backtesting engine
engine = BacktestEngine(
initial_balance=float(INITIAL_BALANCE),
commission_rate=0.0002, # 2 basis points
slippage=0.0001, # 1 basis point
leverage=1.0,
)
# Define strategies to test
strategies = [MovingAverageCrossover(fast_period=20, slow_period=50), RSIMeanReversion(rsi_period=14, oversold=30, overbought=70), BollingerBandStrategy(bb_period=20, bb_std=2.0)]
# Run backtests
results = {}
for strategy in strategies:
print(f"\n{'=' * 50}")
metrics = engine.run_backtest(historical_data, strategy)
results[strategy.name] = metrics
# Display key metrics
print(f"\nStrategy: {strategy.name}")
print(f"Total Return: {metrics['total_return_pct']:.2f}%")
print(f"Max Drawdown: {metrics['max_drawdown_pct']:.2f}%")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f"Total Trades: {metrics['total_trades']}")
print(f"Win Rate: {metrics['win_rate_pct']:.1f}%")
print(f"Profit Factor: {metrics['profit_factor']:.2f}")
print(f"Total Commission: ${metrics['total_commission']:.2f}")
print(f"Total Slippage: ${metrics['total_slippage']:.2f}")
print(f"\n{'=' * 50}")
print("All backtests completed!")
7. Performance Visualization¶
Let's create visualizations of our backtest results.
In [ ]:
Copied!
def plot_strategy_comparison(results: dict[str, dict]):
"""Plot comparison of different strategies."""
_fig, axes = plt.subplots(2, 2, figsize=(20, 15))
# Equity curves
ax1 = axes[0, 0]
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
ax1.plot(equity_curve.index, equity_curve["equity"], label=strategy_name, linewidth=2)
ax1.axhline(y=float(INITIAL_BALANCE), color="black", linestyle="--", alpha=0.5, label="Initial Balance")
ax1.set_title("Equity Curves Comparison", fontsize=14, fontweight="bold")
ax1.set_xlabel("Date")
ax1.set_ylabel("Account Equity ($)")
ax1.legend()
ax1.grid(True, alpha=0.3)
# Drawdown curves
ax2 = axes[0, 1]
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
ax2.fill_between(equity_curve.index, equity_curve["drawdown"], 0, alpha=0.3, label=strategy_name)
ax2.set_title("Drawdown Comparison", fontsize=14, fontweight="bold")
ax2.set_xlabel("Date")
ax2.set_ylabel("Drawdown (%)")
ax2.legend()
ax2.grid(True, alpha=0.3)
# Performance metrics bar chart
ax3 = axes[1, 0]
strategy_names = list(results.keys())
total_returns = [results[name]["total_return_pct"] for name in strategy_names]
sharpe_ratios = [results[name]["sharpe_ratio"] for name in strategy_names]
x = np.arange(len(strategy_names))
width = 0.35
ax3.bar(x - width / 2, total_returns, width, label="Total Return (%)", alpha=0.8)
ax3_twin = ax3.twinx()
ax3_twin.bar(x + width / 2, sharpe_ratios, width, label="Sharpe Ratio", alpha=0.8, color="orange")
ax3.set_title("Performance Metrics Comparison", fontsize=14, fontweight="bold")
ax3.set_xlabel("Strategy")
ax3.set_ylabel("Total Return (%)", color="blue")
ax3_twin.set_ylabel("Sharpe Ratio", color="orange")
ax3.set_xticks(x)
ax3.set_xticklabels([name.replace("_", "\n") for name in strategy_names], rotation=45, ha="right")
ax3.grid(True, alpha=0.3)
# Risk-Return scatter plot
ax4 = axes[1, 1]
volatilities = [results[name]["volatility_pct"] for name in strategy_names]
scatter = ax4.scatter(volatilities, total_returns, s=100, alpha=0.7, c=sharpe_ratios, cmap="viridis")
for i, name in enumerate(strategy_names):
ax4.annotate(name.replace("_", "\n"), (volatilities[i], total_returns[i]), xytext=(5, 5), textcoords="offset points", fontsize=10)
ax4.set_title("Risk-Return Profile", fontsize=14, fontweight="bold")
ax4.set_xlabel("Volatility (%)")
ax4.set_ylabel("Total Return (%)")
ax4.grid(True, alpha=0.3)
# Add colorbar for Sharpe ratio
cbar = plt.colorbar(scatter, ax=ax4)
cbar.set_label("Sharpe Ratio")
plt.tight_layout()
plt.show()
def create_performance_summary_table(results: dict[str, dict]) -> pd.DataFrame:
"""Create a comprehensive performance summary table."""
summary_data = []
for strategy_name, metrics in results.items():
summary_data.append(
{
"Strategy": strategy_name,
"Total Return (%)": f"{metrics['total_return_pct']:.2f}",
"Max Drawdown (%)": f"{metrics['max_drawdown_pct']:.2f}",
"Volatility (%)": f"{metrics['volatility_pct']:.2f}",
"Sharpe Ratio": f"{metrics['sharpe_ratio']:.2f}",
"Total Trades": metrics["total_trades"],
"Win Rate (%)": f"{metrics['win_rate_pct']:.1f}",
"Profit Factor": f"{metrics['profit_factor']:.2f}",
"Final Balance ($)": f"{metrics['final_balance']:.2f}",
"Commission ($)": f"{metrics['total_commission']:.2f}",
"Slippage ($)": f"{metrics['total_slippage']:.2f}",
}
)
return pd.DataFrame(summary_data)
# Generate visualizations
plot_strategy_comparison(results)
# Display performance summary table
summary_table = create_performance_summary_table(results)
print("\nPerformance Summary:")
print(summary_table.to_string(index=False))
def plot_strategy_comparison(results: dict[str, dict]):
"""Plot comparison of different strategies."""
_fig, axes = plt.subplots(2, 2, figsize=(20, 15))
# Equity curves
ax1 = axes[0, 0]
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
ax1.plot(equity_curve.index, equity_curve["equity"], label=strategy_name, linewidth=2)
ax1.axhline(y=float(INITIAL_BALANCE), color="black", linestyle="--", alpha=0.5, label="Initial Balance")
ax1.set_title("Equity Curves Comparison", fontsize=14, fontweight="bold")
ax1.set_xlabel("Date")
ax1.set_ylabel("Account Equity ($)")
ax1.legend()
ax1.grid(True, alpha=0.3)
# Drawdown curves
ax2 = axes[0, 1]
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
ax2.fill_between(equity_curve.index, equity_curve["drawdown"], 0, alpha=0.3, label=strategy_name)
ax2.set_title("Drawdown Comparison", fontsize=14, fontweight="bold")
ax2.set_xlabel("Date")
ax2.set_ylabel("Drawdown (%)")
ax2.legend()
ax2.grid(True, alpha=0.3)
# Performance metrics bar chart
ax3 = axes[1, 0]
strategy_names = list(results.keys())
total_returns = [results[name]["total_return_pct"] for name in strategy_names]
sharpe_ratios = [results[name]["sharpe_ratio"] for name in strategy_names]
x = np.arange(len(strategy_names))
width = 0.35
ax3.bar(x - width / 2, total_returns, width, label="Total Return (%)", alpha=0.8)
ax3_twin = ax3.twinx()
ax3_twin.bar(x + width / 2, sharpe_ratios, width, label="Sharpe Ratio", alpha=0.8, color="orange")
ax3.set_title("Performance Metrics Comparison", fontsize=14, fontweight="bold")
ax3.set_xlabel("Strategy")
ax3.set_ylabel("Total Return (%)", color="blue")
ax3_twin.set_ylabel("Sharpe Ratio", color="orange")
ax3.set_xticks(x)
ax3.set_xticklabels([name.replace("_", "\n") for name in strategy_names], rotation=45, ha="right")
ax3.grid(True, alpha=0.3)
# Risk-Return scatter plot
ax4 = axes[1, 1]
volatilities = [results[name]["volatility_pct"] for name in strategy_names]
scatter = ax4.scatter(volatilities, total_returns, s=100, alpha=0.7, c=sharpe_ratios, cmap="viridis")
for i, name in enumerate(strategy_names):
ax4.annotate(name.replace("_", "\n"), (volatilities[i], total_returns[i]), xytext=(5, 5), textcoords="offset points", fontsize=10)
ax4.set_title("Risk-Return Profile", fontsize=14, fontweight="bold")
ax4.set_xlabel("Volatility (%)")
ax4.set_ylabel("Total Return (%)")
ax4.grid(True, alpha=0.3)
# Add colorbar for Sharpe ratio
cbar = plt.colorbar(scatter, ax=ax4)
cbar.set_label("Sharpe Ratio")
plt.tight_layout()
plt.show()
def create_performance_summary_table(results: dict[str, dict]) -> pd.DataFrame:
"""Create a comprehensive performance summary table."""
summary_data = []
for strategy_name, metrics in results.items():
summary_data.append(
{
"Strategy": strategy_name,
"Total Return (%)": f"{metrics['total_return_pct']:.2f}",
"Max Drawdown (%)": f"{metrics['max_drawdown_pct']:.2f}",
"Volatility (%)": f"{metrics['volatility_pct']:.2f}",
"Sharpe Ratio": f"{metrics['sharpe_ratio']:.2f}",
"Total Trades": metrics["total_trades"],
"Win Rate (%)": f"{metrics['win_rate_pct']:.1f}",
"Profit Factor": f"{metrics['profit_factor']:.2f}",
"Final Balance ($)": f"{metrics['final_balance']:.2f}",
"Commission ($)": f"{metrics['total_commission']:.2f}",
"Slippage ($)": f"{metrics['total_slippage']:.2f}",
}
)
return pd.DataFrame(summary_data)
# Generate visualizations
plot_strategy_comparison(results)
# Display performance summary table
summary_table = create_performance_summary_table(results)
print("\nPerformance Summary:")
print(summary_table.to_string(index=False))
8. Advanced Analysis¶
Let's perform some advanced analysis including rolling performance and trade analysis.
In [ ]:
Copied!
def analyze_rolling_performance(results: dict[str, dict], window_days: int = 30):
"""Analyze rolling performance metrics."""
_fig, axes = plt.subplots(2, 2, figsize=(20, 12))
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
# Calculate rolling metrics
window_hours = window_days * 24 # Convert to hours for hourly data
rolling_returns = equity_curve["returns"].rolling(window_hours).mean() * 100
rolling_volatility = equity_curve["returns"].rolling(window_hours).std() * 100
rolling_sharpe = rolling_returns / rolling_volatility
# Plot rolling returns
axes[0, 0].plot(equity_curve.index, rolling_returns, label=strategy_name, linewidth=1.5)
# Plot rolling volatility
axes[0, 1].plot(equity_curve.index, rolling_volatility, label=strategy_name, linewidth=1.5)
# Plot rolling Sharpe ratio
axes[1, 0].plot(equity_curve.index, rolling_sharpe, label=strategy_name, linewidth=1.5)
# Plot underwater curve (drawdown)
axes[1, 1].fill_between(equity_curve.index, equity_curve["drawdown"], 0, alpha=0.3, label=strategy_name)
# Configure subplots
axes[0, 0].set_title(f"Rolling Returns ({window_days}-day window)", fontweight="bold")
axes[0, 0].set_ylabel("Returns (%)")
axes[0, 0].legend()
axes[0, 0].grid(True, alpha=0.3)
axes[0, 1].set_title(f"Rolling Volatility ({window_days}-day window)", fontweight="bold")
axes[0, 1].set_ylabel("Volatility (%)")
axes[0, 1].legend()
axes[0, 1].grid(True, alpha=0.3)
axes[1, 0].set_title(f"Rolling Sharpe Ratio ({window_days}-day window)", fontweight="bold")
axes[1, 0].set_ylabel("Sharpe Ratio")
axes[1, 0].legend()
axes[1, 0].grid(True, alpha=0.3)
axes[1, 1].set_title("Underwater Curve (Drawdown)", fontweight="bold")
axes[1, 1].set_ylabel("Drawdown (%)")
axes[1, 1].legend()
axes[1, 1].grid(True, alpha=0.3)
for ax in axes.flat:
ax.set_xlabel("Date")
plt.tight_layout()
plt.show()
def analyze_monthly_returns(results: dict[str, dict]):
"""Create monthly returns heatmap for each strategy."""
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
# Calculate monthly returns
monthly_equity = equity_curve["equity"].resample("ME").last()
monthly_returns = monthly_equity.pct_change().dropna() * 100
# Create pivot table for heatmap
monthly_returns_df = pd.DataFrame({"Year": monthly_returns.index.year, "Month": monthly_returns.index.month, "Return": monthly_returns.values})
pivot_table = monthly_returns_df.pivot(index="Year", columns="Month", values="Return")
# Plot heatmap
plt.figure(figsize=(12, 6))
sns.heatmap(pivot_table, annot=True, fmt=".1f", cmap="RdYlBu_r", center=0, cbar_kws={"label": "Monthly Return (%)"}, square=False)
plt.title(f"Monthly Returns Heatmap - {strategy_name}", fontweight="bold")
plt.xlabel("Month")
plt.ylabel("Year")
# Set month labels
month_labels = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
plt.gca().set_xticklabels([month_labels[month - 1] for month in pivot_table.columns])
plt.tight_layout()
plt.show()
# Perform advanced analysis
print("Generating advanced performance analysis...")
analyze_rolling_performance(results, window_days=30)
analyze_monthly_returns(results)
def analyze_rolling_performance(results: dict[str, dict], window_days: int = 30):
"""Analyze rolling performance metrics."""
_fig, axes = plt.subplots(2, 2, figsize=(20, 12))
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
# Calculate rolling metrics
window_hours = window_days * 24 # Convert to hours for hourly data
rolling_returns = equity_curve["returns"].rolling(window_hours).mean() * 100
rolling_volatility = equity_curve["returns"].rolling(window_hours).std() * 100
rolling_sharpe = rolling_returns / rolling_volatility
# Plot rolling returns
axes[0, 0].plot(equity_curve.index, rolling_returns, label=strategy_name, linewidth=1.5)
# Plot rolling volatility
axes[0, 1].plot(equity_curve.index, rolling_volatility, label=strategy_name, linewidth=1.5)
# Plot rolling Sharpe ratio
axes[1, 0].plot(equity_curve.index, rolling_sharpe, label=strategy_name, linewidth=1.5)
# Plot underwater curve (drawdown)
axes[1, 1].fill_between(equity_curve.index, equity_curve["drawdown"], 0, alpha=0.3, label=strategy_name)
# Configure subplots
axes[0, 0].set_title(f"Rolling Returns ({window_days}-day window)", fontweight="bold")
axes[0, 0].set_ylabel("Returns (%)")
axes[0, 0].legend()
axes[0, 0].grid(True, alpha=0.3)
axes[0, 1].set_title(f"Rolling Volatility ({window_days}-day window)", fontweight="bold")
axes[0, 1].set_ylabel("Volatility (%)")
axes[0, 1].legend()
axes[0, 1].grid(True, alpha=0.3)
axes[1, 0].set_title(f"Rolling Sharpe Ratio ({window_days}-day window)", fontweight="bold")
axes[1, 0].set_ylabel("Sharpe Ratio")
axes[1, 0].legend()
axes[1, 0].grid(True, alpha=0.3)
axes[1, 1].set_title("Underwater Curve (Drawdown)", fontweight="bold")
axes[1, 1].set_ylabel("Drawdown (%)")
axes[1, 1].legend()
axes[1, 1].grid(True, alpha=0.3)
for ax in axes.flat:
ax.set_xlabel("Date")
plt.tight_layout()
plt.show()
def analyze_monthly_returns(results: dict[str, dict]):
"""Create monthly returns heatmap for each strategy."""
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
# Calculate monthly returns
monthly_equity = equity_curve["equity"].resample("ME").last()
monthly_returns = monthly_equity.pct_change().dropna() * 100
# Create pivot table for heatmap
monthly_returns_df = pd.DataFrame({"Year": monthly_returns.index.year, "Month": monthly_returns.index.month, "Return": monthly_returns.values})
pivot_table = monthly_returns_df.pivot(index="Year", columns="Month", values="Return")
# Plot heatmap
plt.figure(figsize=(12, 6))
sns.heatmap(pivot_table, annot=True, fmt=".1f", cmap="RdYlBu_r", center=0, cbar_kws={"label": "Monthly Return (%)"}, square=False)
plt.title(f"Monthly Returns Heatmap - {strategy_name}", fontweight="bold")
plt.xlabel("Month")
plt.ylabel("Year")
# Set month labels
month_labels = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
plt.gca().set_xticklabels([month_labels[month - 1] for month in pivot_table.columns])
plt.tight_layout()
plt.show()
# Perform advanced analysis
print("Generating advanced performance analysis...")
analyze_rolling_performance(results, window_days=30)
analyze_monthly_returns(results)
9. Risk Analysis¶
Let's perform detailed risk analysis including Value at Risk (VaR) and other risk metrics.
In [ ]:
Copied!
def calculate_risk_metrics(results: dict[str, dict]):
"""Calculate comprehensive risk metrics."""
risk_analysis = {}
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
returns = equity_curve["returns"].dropna()
if len(returns) == 0:
continue
# Value at Risk (VaR)
var_95 = np.percentile(returns, 5) * 100
var_99 = np.percentile(returns, 1) * 100
# Conditional VaR (Expected Shortfall)
cvar_95 = returns[returns <= np.percentile(returns, 5)].mean() * 100
cvar_99 = returns[returns <= np.percentile(returns, 1)].mean() * 100
# Maximum consecutive losses
consecutive_losses = 0
max_consecutive_losses = 0
for ret in returns:
if ret < 0:
consecutive_losses += 1
max_consecutive_losses = max(max_consecutive_losses, consecutive_losses)
else:
consecutive_losses = 0
# Calmar Ratio (Annual Return / Max Drawdown)
annual_return = (equity_curve["equity"].iloc[-1] / equity_curve["equity"].iloc[0] - 1) * 100
calmar_ratio = annual_return / abs(metrics["max_drawdown_pct"]) if metrics["max_drawdown_pct"] != 0 else float("inf")
# Sortino Ratio (return / downside deviation)
downside_returns = returns[returns < 0]
downside_std = downside_returns.std() if len(downside_returns) > 0 else 0
sortino_ratio = (returns.mean() * 252 * 24) / (downside_std * np.sqrt(252 * 24)) if downside_std > 0 else float("inf")
# Skewness and Kurtosis
skewness = returns.skew()
kurtosis = returns.kurtosis()
risk_analysis[strategy_name] = {"VaR_95": var_95, "VaR_99": var_99, "CVaR_95": cvar_95, "CVaR_99": cvar_99, "Max_Consecutive_Losses": max_consecutive_losses, "Calmar_Ratio": calmar_ratio, "Sortino_Ratio": sortino_ratio, "Skewness": skewness, "Kurtosis": kurtosis}
return risk_analysis
def plot_return_distributions(results: dict[str, dict]):
"""Plot return distributions for each strategy."""
_fig, axes = plt.subplots(1, len(results), figsize=(5 * len(results), 6))
if len(results) == 1:
axes = [axes]
for i, (strategy_name, metrics) in enumerate(results.items()):
equity_curve = metrics["equity_curve"]
returns = equity_curve["returns"].dropna() * 100
# Plot histogram
axes[i].hist(returns, bins=50, alpha=0.7, density=True, edgecolor="black")
# Plot normal distribution for comparison
mu, sigma = returns.mean(), returns.std()
x = np.linspace(returns.min(), returns.max(), 100)
normal_dist = (1 / (sigma * np.sqrt(2 * np.pi))) * np.exp(-0.5 * ((x - mu) / sigma) ** 2)
axes[i].plot(x, normal_dist, "r-", linewidth=2, label="Normal Distribution")
# Add VaR lines
var_95 = np.percentile(returns, 5)
var_99 = np.percentile(returns, 1)
axes[i].axvline(var_95, color="orange", linestyle="--", label=f"VaR 95%: {var_95:.2f}%")
axes[i].axvline(var_99, color="red", linestyle="--", label=f"VaR 99%: {var_99:.2f}%")
axes[i].set_title(f"{strategy_name}\nReturn Distribution", fontweight="bold")
axes[i].set_xlabel("Returns (%)")
axes[i].set_ylabel("Density")
axes[i].legend()
axes[i].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Perform risk analysis
print("Calculating risk metrics...")
risk_metrics = calculate_risk_metrics(results)
# Create risk metrics table
risk_df = pd.DataFrame(risk_metrics).T
risk_df = risk_df.round(2)
print("\nRisk Analysis Summary:")
print(risk_df.to_string())
# Plot return distributions
plot_return_distributions(results)
def calculate_risk_metrics(results: dict[str, dict]):
"""Calculate comprehensive risk metrics."""
risk_analysis = {}
for strategy_name, metrics in results.items():
equity_curve = metrics["equity_curve"]
returns = equity_curve["returns"].dropna()
if len(returns) == 0:
continue
# Value at Risk (VaR)
var_95 = np.percentile(returns, 5) * 100
var_99 = np.percentile(returns, 1) * 100
# Conditional VaR (Expected Shortfall)
cvar_95 = returns[returns <= np.percentile(returns, 5)].mean() * 100
cvar_99 = returns[returns <= np.percentile(returns, 1)].mean() * 100
# Maximum consecutive losses
consecutive_losses = 0
max_consecutive_losses = 0
for ret in returns:
if ret < 0:
consecutive_losses += 1
max_consecutive_losses = max(max_consecutive_losses, consecutive_losses)
else:
consecutive_losses = 0
# Calmar Ratio (Annual Return / Max Drawdown)
annual_return = (equity_curve["equity"].iloc[-1] / equity_curve["equity"].iloc[0] - 1) * 100
calmar_ratio = annual_return / abs(metrics["max_drawdown_pct"]) if metrics["max_drawdown_pct"] != 0 else float("inf")
# Sortino Ratio (return / downside deviation)
downside_returns = returns[returns < 0]
downside_std = downside_returns.std() if len(downside_returns) > 0 else 0
sortino_ratio = (returns.mean() * 252 * 24) / (downside_std * np.sqrt(252 * 24)) if downside_std > 0 else float("inf")
# Skewness and Kurtosis
skewness = returns.skew()
kurtosis = returns.kurtosis()
risk_analysis[strategy_name] = {"VaR_95": var_95, "VaR_99": var_99, "CVaR_95": cvar_95, "CVaR_99": cvar_99, "Max_Consecutive_Losses": max_consecutive_losses, "Calmar_Ratio": calmar_ratio, "Sortino_Ratio": sortino_ratio, "Skewness": skewness, "Kurtosis": kurtosis}
return risk_analysis
def plot_return_distributions(results: dict[str, dict]):
"""Plot return distributions for each strategy."""
_fig, axes = plt.subplots(1, len(results), figsize=(5 * len(results), 6))
if len(results) == 1:
axes = [axes]
for i, (strategy_name, metrics) in enumerate(results.items()):
equity_curve = metrics["equity_curve"]
returns = equity_curve["returns"].dropna() * 100
# Plot histogram
axes[i].hist(returns, bins=50, alpha=0.7, density=True, edgecolor="black")
# Plot normal distribution for comparison
mu, sigma = returns.mean(), returns.std()
x = np.linspace(returns.min(), returns.max(), 100)
normal_dist = (1 / (sigma * np.sqrt(2 * np.pi))) * np.exp(-0.5 * ((x - mu) / sigma) ** 2)
axes[i].plot(x, normal_dist, "r-", linewidth=2, label="Normal Distribution")
# Add VaR lines
var_95 = np.percentile(returns, 5)
var_99 = np.percentile(returns, 1)
axes[i].axvline(var_95, color="orange", linestyle="--", label=f"VaR 95%: {var_95:.2f}%")
axes[i].axvline(var_99, color="red", linestyle="--", label=f"VaR 99%: {var_99:.2f}%")
axes[i].set_title(f"{strategy_name}\nReturn Distribution", fontweight="bold")
axes[i].set_xlabel("Returns (%)")
axes[i].set_ylabel("Density")
axes[i].legend()
axes[i].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Perform risk analysis
print("Calculating risk metrics...")
risk_metrics = calculate_risk_metrics(results)
# Create risk metrics table
risk_df = pd.DataFrame(risk_metrics).T
risk_df = risk_df.round(2)
print("\nRisk Analysis Summary:")
print(risk_df.to_string())
# Plot return distributions
plot_return_distributions(results)
10. Strategy Optimization¶
This grid search compares parameter choices on the supplied data. Selecting the best result on that same data can overfit; evaluate chosen settings on separate data.
In [ ]:
Copied!
def optimize_ma_crossover(data: pd.DataFrame, engine: BacktestEngine):
"""Optimize Moving Average Crossover parameters."""
print("Optimizing Moving Average Crossover parameters...")
# Parameter ranges
fast_periods = [10, 15, 20, 25, 30]
slow_periods = [40, 50, 60, 70, 80]
optimization_results = []
data = data.copy()
for period in sorted(set(fast_periods) | set(slow_periods)):
data[f"sma_{period}"] = calculate_sma(data["close"], period)
for fast in fast_periods:
for slow in slow_periods:
if fast >= slow:
continue
strategy = MovingAverageCrossover(fast_period=fast, slow_period=slow)
metrics = engine.run_backtest(data, strategy)
optimization_results.append({"fast_period": fast, "slow_period": slow, "total_return": metrics["total_return_pct"], "max_drawdown": metrics["max_drawdown_pct"], "sharpe_ratio": metrics["sharpe_ratio"], "total_trades": metrics["total_trades"], "win_rate": metrics["win_rate_pct"]})
# Convert to DataFrame for analysis
opt_df = pd.DataFrame(optimization_results)
# Find best parameters by different criteria
best_return = opt_df.loc[opt_df["total_return"].idxmax()]
best_sharpe = opt_df.loc[opt_df["sharpe_ratio"].idxmax()]
best_drawdown = opt_df.loc[opt_df["max_drawdown"].idxmin()] # Minimum drawdown
print("\nOptimization Results:")
print(f"Best Total Return: Fast={best_return['fast_period']}, Slow={best_return['slow_period']}, Return={best_return['total_return']:.2f}%")
print(f"Best Sharpe Ratio: Fast={best_sharpe['fast_period']}, Slow={best_sharpe['slow_period']}, Sharpe={best_sharpe['sharpe_ratio']:.2f}")
print(f"Best Drawdown: Fast={best_drawdown['fast_period']}, Slow={best_drawdown['slow_period']}, Drawdown={best_drawdown['max_drawdown']:.2f}%")
# Create heatmap of results
pivot_return = opt_df.pivot(index="slow_period", columns="fast_period", values="total_return")
pivot_sharpe = opt_df.pivot(index="slow_period", columns="fast_period", values="sharpe_ratio")
_fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
# Total return heatmap
sns.heatmap(pivot_return, annot=True, fmt=".1f", cmap="RdYlBu_r", ax=ax1, cbar_kws={"label": "Total Return (%)"})
ax1.set_title("Total Return Optimization Heatmap", fontweight="bold")
ax1.set_xlabel("Fast Period")
ax1.set_ylabel("Slow Period")
# Sharpe ratio heatmap
sns.heatmap(pivot_sharpe, annot=True, fmt=".2f", cmap="RdYlBu_r", ax=ax2, cbar_kws={"label": "Sharpe Ratio"})
ax2.set_title("Sharpe Ratio Optimization Heatmap", fontweight="bold")
ax2.set_xlabel("Fast Period")
ax2.set_ylabel("Slow Period")
plt.tight_layout()
plt.show()
return opt_df
# Run optimization (using a subset of data for speed)
optimization_data = historical_data.iloc[:2000] # Use first 2000 bars for optimization
opt_engine = BacktestEngine(initial_balance=float(INITIAL_BALANCE))
optimization_results = optimize_ma_crossover(optimization_data, opt_engine)
# Display top 10 parameter combinations
print("\nTop 10 Parameter Combinations (by Sharpe Ratio):")
top_10 = optimization_results.nlargest(10, "sharpe_ratio")[["fast_period", "slow_period", "total_return", "sharpe_ratio", "max_drawdown"]]
print(top_10.to_string(index=False))
def optimize_ma_crossover(data: pd.DataFrame, engine: BacktestEngine):
"""Optimize Moving Average Crossover parameters."""
print("Optimizing Moving Average Crossover parameters...")
# Parameter ranges
fast_periods = [10, 15, 20, 25, 30]
slow_periods = [40, 50, 60, 70, 80]
optimization_results = []
data = data.copy()
for period in sorted(set(fast_periods) | set(slow_periods)):
data[f"sma_{period}"] = calculate_sma(data["close"], period)
for fast in fast_periods:
for slow in slow_periods:
if fast >= slow:
continue
strategy = MovingAverageCrossover(fast_period=fast, slow_period=slow)
metrics = engine.run_backtest(data, strategy)
optimization_results.append({"fast_period": fast, "slow_period": slow, "total_return": metrics["total_return_pct"], "max_drawdown": metrics["max_drawdown_pct"], "sharpe_ratio": metrics["sharpe_ratio"], "total_trades": metrics["total_trades"], "win_rate": metrics["win_rate_pct"]})
# Convert to DataFrame for analysis
opt_df = pd.DataFrame(optimization_results)
# Find best parameters by different criteria
best_return = opt_df.loc[opt_df["total_return"].idxmax()]
best_sharpe = opt_df.loc[opt_df["sharpe_ratio"].idxmax()]
best_drawdown = opt_df.loc[opt_df["max_drawdown"].idxmin()] # Minimum drawdown
print("\nOptimization Results:")
print(f"Best Total Return: Fast={best_return['fast_period']}, Slow={best_return['slow_period']}, Return={best_return['total_return']:.2f}%")
print(f"Best Sharpe Ratio: Fast={best_sharpe['fast_period']}, Slow={best_sharpe['slow_period']}, Sharpe={best_sharpe['sharpe_ratio']:.2f}")
print(f"Best Drawdown: Fast={best_drawdown['fast_period']}, Slow={best_drawdown['slow_period']}, Drawdown={best_drawdown['max_drawdown']:.2f}%")
# Create heatmap of results
pivot_return = opt_df.pivot(index="slow_period", columns="fast_period", values="total_return")
pivot_sharpe = opt_df.pivot(index="slow_period", columns="fast_period", values="sharpe_ratio")
_fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
# Total return heatmap
sns.heatmap(pivot_return, annot=True, fmt=".1f", cmap="RdYlBu_r", ax=ax1, cbar_kws={"label": "Total Return (%)"})
ax1.set_title("Total Return Optimization Heatmap", fontweight="bold")
ax1.set_xlabel("Fast Period")
ax1.set_ylabel("Slow Period")
# Sharpe ratio heatmap
sns.heatmap(pivot_sharpe, annot=True, fmt=".2f", cmap="RdYlBu_r", ax=ax2, cbar_kws={"label": "Sharpe Ratio"})
ax2.set_title("Sharpe Ratio Optimization Heatmap", fontweight="bold")
ax2.set_xlabel("Fast Period")
ax2.set_ylabel("Slow Period")
plt.tight_layout()
plt.show()
return opt_df
# Run optimization (using a subset of data for speed)
optimization_data = historical_data.iloc[:2000] # Use first 2000 bars for optimization
opt_engine = BacktestEngine(initial_balance=float(INITIAL_BALANCE))
optimization_results = optimize_ma_crossover(optimization_data, opt_engine)
# Display top 10 parameter combinations
print("\nTop 10 Parameter Combinations (by Sharpe Ratio):")
top_10 = optimization_results.nlargest(10, "sharpe_ratio")[["fast_period", "slow_period", "total_return", "sharpe_ratio", "max_drawdown"]]
print(top_10.to_string(index=False))
11. Walk-Forward Analysis¶
This example divides data into successive evaluation windows. Inspect whether fitting and evaluation remain separate in each window; the label alone does not establish an out-of-sample test.
In [ ]:
Copied!
def walk_forward_analysis(data: pd.DataFrame, strategy: Strategy, window_months: int = 3):
"""Perform walk-forward analysis of a strategy."""
print(f"Performing walk-forward analysis for {strategy.name}...")
# Calculate window size in hours (assuming hourly data)
window_size = window_months * 30 * 24 # Approximate hours per month
# Ensure we have enough data
if len(data) < window_size * 2:
print("Not enough data for walk-forward analysis")
return None
walk_forward_results = []
engine = BacktestEngine(initial_balance=10000)
# Start from the minimum window size
start_idx = window_size
while start_idx + window_size < len(data):
# Define the analysis window
end_idx = start_idx + window_size
window_data = data.iloc[start_idx:end_idx]
# Run backtest on this window
metrics = engine.run_backtest(window_data, strategy)
walk_forward_results.append({"start_date": window_data.index[0], "end_date": window_data.index[-1], "total_return": metrics["total_return_pct"], "max_drawdown": metrics["max_drawdown_pct"], "sharpe_ratio": metrics["sharpe_ratio"], "total_trades": metrics["total_trades"], "win_rate": metrics["win_rate_pct"]})
# Move window forward by 1 month
start_idx += 30 * 24 # Move forward by approximately 1 month
# Convert to DataFrame
wf_df = pd.DataFrame(walk_forward_results)
wf_df.set_index("start_date", inplace=True)
# Plot results
_fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# Rolling returns
axes[0, 0].plot(wf_df.index, wf_df["total_return"], marker="o", linewidth=2)
axes[0, 0].axhline(y=0, color="black", linestyle="--", alpha=0.5)
axes[0, 0].set_title(f"Walk-Forward Returns - {strategy.name}", fontweight="bold")
axes[0, 0].set_ylabel("Total Return (%)")
axes[0, 0].grid(True, alpha=0.3)
# Rolling Sharpe ratio
axes[0, 1].plot(wf_df.index, wf_df["sharpe_ratio"], marker="o", linewidth=2, color="green")
axes[0, 1].axhline(y=1, color="red", linestyle="--", alpha=0.5, label="Sharpe = 1")
axes[0, 1].set_title(f"Walk-Forward Sharpe Ratio - {strategy.name}", fontweight="bold")
axes[0, 1].set_ylabel("Sharpe Ratio")
axes[0, 1].legend()
axes[0, 1].grid(True, alpha=0.3)
# Rolling drawdown
axes[1, 0].plot(wf_df.index, wf_df["max_drawdown"], marker="o", linewidth=2, color="red")
axes[1, 0].set_title(f"Walk-Forward Max Drawdown - {strategy.name}", fontweight="bold")
axes[1, 0].set_ylabel("Max Drawdown (%)")
axes[1, 0].grid(True, alpha=0.3)
# Rolling win rate
axes[1, 1].plot(wf_df.index, wf_df["win_rate"], marker="o", linewidth=2, color="purple")
axes[1, 1].axhline(y=50, color="black", linestyle="--", alpha=0.5, label="50% Win Rate")
axes[1, 1].set_title(f"Walk-Forward Win Rate - {strategy.name}", fontweight="bold")
axes[1, 1].set_ylabel("Win Rate (%)")
axes[1, 1].legend()
axes[1, 1].grid(True, alpha=0.3)
for ax in axes.flat:
ax.set_xlabel("Date")
plt.tight_layout()
plt.show()
# Print summary statistics
print(f"\nWalk-Forward Analysis Summary for {strategy.name}:")
print(f"Average Return: {wf_df['total_return'].mean():.2f}% (std: {wf_df['total_return'].std():.2f}%)")
print(f"Average Sharpe: {wf_df['sharpe_ratio'].mean():.2f} (std: {wf_df['sharpe_ratio'].std():.2f})")
print(f"Average Drawdown: {wf_df['max_drawdown'].mean():.2f}% (std: {wf_df['max_drawdown'].std():.2f}%)")
print(f"Win Rate Consistency: {wf_df['win_rate'].mean():.1f}% (std: {wf_df['win_rate'].std():.1f}%)")
print(f"Positive Return Periods: {(wf_df['total_return'] > 0).sum()} out of {len(wf_df)} ({(wf_df['total_return'] > 0).mean() * 100:.1f}%)")
return wf_df
# Run walk-forward analysis on the best strategy from our earlier results
best_strategy_name = max(results.keys(), key=lambda x: results[x]["sharpe_ratio"])
print(f"Running walk-forward analysis on best strategy: {best_strategy_name}")
# Create the strategy (you might need to adjust parameters based on your optimization results)
if "MovingAverage" in best_strategy_name:
best_strategy = MovingAverageCrossover(fast_period=20, slow_period=50)
elif "RSI" in best_strategy_name:
best_strategy = RSIMeanReversion()
else:
best_strategy = BollingerBandStrategy()
walk_forward_results = walk_forward_analysis(historical_data, best_strategy, window_months=2)
def walk_forward_analysis(data: pd.DataFrame, strategy: Strategy, window_months: int = 3):
"""Perform walk-forward analysis of a strategy."""
print(f"Performing walk-forward analysis for {strategy.name}...")
# Calculate window size in hours (assuming hourly data)
window_size = window_months * 30 * 24 # Approximate hours per month
# Ensure we have enough data
if len(data) < window_size * 2:
print("Not enough data for walk-forward analysis")
return None
walk_forward_results = []
engine = BacktestEngine(initial_balance=10000)
# Start from the minimum window size
start_idx = window_size
while start_idx + window_size < len(data):
# Define the analysis window
end_idx = start_idx + window_size
window_data = data.iloc[start_idx:end_idx]
# Run backtest on this window
metrics = engine.run_backtest(window_data, strategy)
walk_forward_results.append({"start_date": window_data.index[0], "end_date": window_data.index[-1], "total_return": metrics["total_return_pct"], "max_drawdown": metrics["max_drawdown_pct"], "sharpe_ratio": metrics["sharpe_ratio"], "total_trades": metrics["total_trades"], "win_rate": metrics["win_rate_pct"]})
# Move window forward by 1 month
start_idx += 30 * 24 # Move forward by approximately 1 month
# Convert to DataFrame
wf_df = pd.DataFrame(walk_forward_results)
wf_df.set_index("start_date", inplace=True)
# Plot results
_fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# Rolling returns
axes[0, 0].plot(wf_df.index, wf_df["total_return"], marker="o", linewidth=2)
axes[0, 0].axhline(y=0, color="black", linestyle="--", alpha=0.5)
axes[0, 0].set_title(f"Walk-Forward Returns - {strategy.name}", fontweight="bold")
axes[0, 0].set_ylabel("Total Return (%)")
axes[0, 0].grid(True, alpha=0.3)
# Rolling Sharpe ratio
axes[0, 1].plot(wf_df.index, wf_df["sharpe_ratio"], marker="o", linewidth=2, color="green")
axes[0, 1].axhline(y=1, color="red", linestyle="--", alpha=0.5, label="Sharpe = 1")
axes[0, 1].set_title(f"Walk-Forward Sharpe Ratio - {strategy.name}", fontweight="bold")
axes[0, 1].set_ylabel("Sharpe Ratio")
axes[0, 1].legend()
axes[0, 1].grid(True, alpha=0.3)
# Rolling drawdown
axes[1, 0].plot(wf_df.index, wf_df["max_drawdown"], marker="o", linewidth=2, color="red")
axes[1, 0].set_title(f"Walk-Forward Max Drawdown - {strategy.name}", fontweight="bold")
axes[1, 0].set_ylabel("Max Drawdown (%)")
axes[1, 0].grid(True, alpha=0.3)
# Rolling win rate
axes[1, 1].plot(wf_df.index, wf_df["win_rate"], marker="o", linewidth=2, color="purple")
axes[1, 1].axhline(y=50, color="black", linestyle="--", alpha=0.5, label="50% Win Rate")
axes[1, 1].set_title(f"Walk-Forward Win Rate - {strategy.name}", fontweight="bold")
axes[1, 1].set_ylabel("Win Rate (%)")
axes[1, 1].legend()
axes[1, 1].grid(True, alpha=0.3)
for ax in axes.flat:
ax.set_xlabel("Date")
plt.tight_layout()
plt.show()
# Print summary statistics
print(f"\nWalk-Forward Analysis Summary for {strategy.name}:")
print(f"Average Return: {wf_df['total_return'].mean():.2f}% (std: {wf_df['total_return'].std():.2f}%)")
print(f"Average Sharpe: {wf_df['sharpe_ratio'].mean():.2f} (std: {wf_df['sharpe_ratio'].std():.2f})")
print(f"Average Drawdown: {wf_df['max_drawdown'].mean():.2f}% (std: {wf_df['max_drawdown'].std():.2f}%)")
print(f"Win Rate Consistency: {wf_df['win_rate'].mean():.1f}% (std: {wf_df['win_rate'].std():.1f}%)")
print(f"Positive Return Periods: {(wf_df['total_return'] > 0).sum()} out of {len(wf_df)} ({(wf_df['total_return'] > 0).mean() * 100:.1f}%)")
return wf_df
# Run walk-forward analysis on the best strategy from our earlier results
best_strategy_name = max(results.keys(), key=lambda x: results[x]["sharpe_ratio"])
print(f"Running walk-forward analysis on best strategy: {best_strategy_name}")
# Create the strategy (you might need to adjust parameters based on your optimization results)
if "MovingAverage" in best_strategy_name:
best_strategy = MovingAverageCrossover(fast_period=20, slow_period=50)
elif "RSI" in best_strategy_name:
best_strategy = RSIMeanReversion()
else:
best_strategy = BollingerBandStrategy()
walk_forward_results = walk_forward_analysis(historical_data, best_strategy, window_months=2)
Interpreting the results¶
Check the printed inputs, timestamps and handled errors before interpreting output. A completed cell can still report an API failure. Example calculations and simulations do not establish a profitable strategy, a guaranteed loss cap or readiness for unattended execution.
For further work, test boundary cases, currency conversion, incomplete data and recovery after a disconnect. Keep signal evaluation separate from order submission and reconcile any account changes.
In [ ]:
Copied!
# Summary of key findings
print("=" * 60)
print("BACKTESTING SUMMARY")
print("=" * 60)
print("\n📊 STRATEGIES TESTED:")
for i, strategy_name in enumerate(results.keys(), 1):
metrics = results[strategy_name]
print(f"{i}. {strategy_name}:")
print(f" Return: {metrics['total_return_pct']:.2f}% | Sharpe: {metrics['sharpe_ratio']:.2f} | Max DD: {metrics['max_drawdown_pct']:.2f}%")
best_strategy = max(results.keys(), key=lambda x: results[x]["sharpe_ratio"])
print(f"\n🏆 BEST PERFORMING STRATEGY (by Sharpe Ratio): {best_strategy}")
print(f" Final Performance: {results[best_strategy]['total_return_pct']:.2f}% return with {results[best_strategy]['sharpe_ratio']:.2f} Sharpe ratio")
print("\n💡 KEY INSIGHTS:")
print("• Transaction costs and slippage significantly impact performance")
print("• Strategy performance varies significantly over different time periods")
print("• Risk-adjusted returns (Sharpe ratio) often differ from raw returns")
print("• Drawdown analysis is crucial for understanding downside risk")
print("\n🚀 NEXT STEPS FOR IMPROVEMENT:")
print("1. ADVANCED STRATEGIES:")
print(" • Multi-timeframe analysis")
print(" • Machine learning-based signals")
print(" • Portfolio-based strategies")
print("\n2. RISK MANAGEMENT:")
print(" • Dynamic position sizing")
print(" • Stop-loss and take-profit optimization")
print(" • Correlation analysis across instruments")
print("\n3. EXECUTION IMPROVEMENTS:")
print(" • Market impact modeling")
print(" • Order book analysis")
print(" • Latency considerations")
print("\n4. STATISTICAL ROBUSTNESS:")
print(" • Bootstrap analysis")
print(" • Monte Carlo simulation")
print(" • Out-of-sample validation")
print("\n5. LIVE TRADING PREPARATION:")
print(" • Paper trading implementation")
print(" • Real-time data integration")
print(" • Performance monitoring dashboard")
print("\n" + "=" * 60)
print("For production deployment, see our 'How-to Deploy SDK to Production' guide")
print("For advanced strategies, check our 'How-to Optimize High-Frequency Trading' guide")
print("=" * 60)
# Summary of key findings
print("=" * 60)
print("BACKTESTING SUMMARY")
print("=" * 60)
print("\n📊 STRATEGIES TESTED:")
for i, strategy_name in enumerate(results.keys(), 1):
metrics = results[strategy_name]
print(f"{i}. {strategy_name}:")
print(f" Return: {metrics['total_return_pct']:.2f}% | Sharpe: {metrics['sharpe_ratio']:.2f} | Max DD: {metrics['max_drawdown_pct']:.2f}%")
best_strategy = max(results.keys(), key=lambda x: results[x]["sharpe_ratio"])
print(f"\n🏆 BEST PERFORMING STRATEGY (by Sharpe Ratio): {best_strategy}")
print(f" Final Performance: {results[best_strategy]['total_return_pct']:.2f}% return with {results[best_strategy]['sharpe_ratio']:.2f} Sharpe ratio")
print("\n💡 KEY INSIGHTS:")
print("• Transaction costs and slippage significantly impact performance")
print("• Strategy performance varies significantly over different time periods")
print("• Risk-adjusted returns (Sharpe ratio) often differ from raw returns")
print("• Drawdown analysis is crucial for understanding downside risk")
print("\n🚀 NEXT STEPS FOR IMPROVEMENT:")
print("1. ADVANCED STRATEGIES:")
print(" • Multi-timeframe analysis")
print(" • Machine learning-based signals")
print(" • Portfolio-based strategies")
print("\n2. RISK MANAGEMENT:")
print(" • Dynamic position sizing")
print(" • Stop-loss and take-profit optimization")
print(" • Correlation analysis across instruments")
print("\n3. EXECUTION IMPROVEMENTS:")
print(" • Market impact modeling")
print(" • Order book analysis")
print(" • Latency considerations")
print("\n4. STATISTICAL ROBUSTNESS:")
print(" • Bootstrap analysis")
print(" • Monte Carlo simulation")
print(" • Out-of-sample validation")
print("\n5. LIVE TRADING PREPARATION:")
print(" • Paper trading implementation")
print(" • Real-time data integration")
print(" • Performance monitoring dashboard")
print("\n" + "=" * 60)
print("For production deployment, see our 'How-to Deploy SDK to Production' guide")
print("For advanced strategies, check our 'How-to Optimize High-Frequency Trading' guide")
print("=" * 60)
13. Save Results and Export¶
Finally, let's save our backtesting results for future reference and analysis.
In [ ]:
Copied!
import json
def save_backtest_results(results: dict[str, dict], filename_prefix: str = "backtest_results"):
"""Save backtest results to files."""
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
# Prepare data for JSON serialization
json_results = {}
for strategy_name, metrics in results.items():
# Convert equity curve to serializable format
equity_curve = metrics["equity_curve"]
equity_data = {
"timestamps": equity_curve.index.strftime("%Y-%m-%d %H:%M:%S").tolist(),
"equity": equity_curve["equity"].tolist(),
"balance": equity_curve["balance"].tolist(),
"unrealized_pnl": equity_curve["unrealized_pnl"].tolist(),
"returns": equity_curve["returns"].fillna(0).tolist(),
"drawdown": equity_curve["drawdown"].fillna(0).tolist(),
}
# Convert trades to serializable format
trades_data = []
for trade in metrics["trades"]:
trades_data.append({"timestamp": trade.timestamp.strftime("%Y-%m-%d %H:%M:%S"), "signal": trade.signal.name, "price": trade.price, "units": trade.units})
json_results[strategy_name] = {
"performance_metrics": {
"initial_balance": metrics["initial_balance"],
"final_balance": metrics["final_balance"],
"total_return_pct": metrics["total_return_pct"],
"max_drawdown_pct": metrics["max_drawdown_pct"],
"volatility_pct": metrics["volatility_pct"],
"sharpe_ratio": metrics["sharpe_ratio"],
"total_trades": metrics["total_trades"],
"winning_trades": metrics["winning_trades"],
"losing_trades": metrics["losing_trades"],
"win_rate_pct": metrics["win_rate_pct"],
"profit_factor": metrics["profit_factor"],
"total_commission": metrics["total_commission"],
"total_slippage": metrics["total_slippage"],
},
"equity_curve": equity_data,
"trades": trades_data,
}
# Save to JSON file
json_filename = f"{filename_prefix}_{timestamp}.json"
with open(json_filename, "w") as f:
json.dump(json_results, f, indent=2)
print(f"Results saved to: {json_filename}")
# Also save summary table as CSV
summary_table = create_performance_summary_table(results)
csv_filename = f"{filename_prefix}_summary_{timestamp}.csv"
summary_table.to_csv(csv_filename, index=False)
print(f"Summary table saved to: {csv_filename}")
return json_filename, csv_filename
# Save the results
print("Saving backtest results...")
json_file, csv_file = save_backtest_results(results, f"oanda_backtest_{INSTRUMENT}")
print("\n✅ BACKTESTING COMPLETE!")
print(f"📁 Results saved as: {json_file} and {csv_file}")
print("\n📚 For more advanced trading strategies and deployment guides, check out:")
print(" • docs/how-to-guides/implement-stop-loss-strategies.md")
print(" • docs/how-to-guides/optimize-high-frequency-trading.md")
print(" • docs/how-to-guides/deploy-sdk-to-production.md")
import json
def save_backtest_results(results: dict[str, dict], filename_prefix: str = "backtest_results"):
"""Save backtest results to files."""
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
# Prepare data for JSON serialization
json_results = {}
for strategy_name, metrics in results.items():
# Convert equity curve to serializable format
equity_curve = metrics["equity_curve"]
equity_data = {
"timestamps": equity_curve.index.strftime("%Y-%m-%d %H:%M:%S").tolist(),
"equity": equity_curve["equity"].tolist(),
"balance": equity_curve["balance"].tolist(),
"unrealized_pnl": equity_curve["unrealized_pnl"].tolist(),
"returns": equity_curve["returns"].fillna(0).tolist(),
"drawdown": equity_curve["drawdown"].fillna(0).tolist(),
}
# Convert trades to serializable format
trades_data = []
for trade in metrics["trades"]:
trades_data.append({"timestamp": trade.timestamp.strftime("%Y-%m-%d %H:%M:%S"), "signal": trade.signal.name, "price": trade.price, "units": trade.units})
json_results[strategy_name] = {
"performance_metrics": {
"initial_balance": metrics["initial_balance"],
"final_balance": metrics["final_balance"],
"total_return_pct": metrics["total_return_pct"],
"max_drawdown_pct": metrics["max_drawdown_pct"],
"volatility_pct": metrics["volatility_pct"],
"sharpe_ratio": metrics["sharpe_ratio"],
"total_trades": metrics["total_trades"],
"winning_trades": metrics["winning_trades"],
"losing_trades": metrics["losing_trades"],
"win_rate_pct": metrics["win_rate_pct"],
"profit_factor": metrics["profit_factor"],
"total_commission": metrics["total_commission"],
"total_slippage": metrics["total_slippage"],
},
"equity_curve": equity_data,
"trades": trades_data,
}
# Save to JSON file
json_filename = f"{filename_prefix}_{timestamp}.json"
with open(json_filename, "w") as f:
json.dump(json_results, f, indent=2)
print(f"Results saved to: {json_filename}")
# Also save summary table as CSV
summary_table = create_performance_summary_table(results)
csv_filename = f"{filename_prefix}_summary_{timestamp}.csv"
summary_table.to_csv(csv_filename, index=False)
print(f"Summary table saved to: {csv_filename}")
return json_filename, csv_filename
# Save the results
print("Saving backtest results...")
json_file, csv_file = save_backtest_results(results, f"oanda_backtest_{INSTRUMENT}")
print("\n✅ BACKTESTING COMPLETE!")
print(f"📁 Results saved as: {json_file} and {csv_file}")
print("\n📚 For more advanced trading strategies and deployment guides, check out:")
print(" • docs/how-to-guides/implement-stop-loss-strategies.md")
print(" • docs/how-to-guides/optimize-high-frequency-trading.md")
print(" • docs/how-to-guides/deploy-sdk-to-production.md")