A Walk-Forward Backtest of Adaptive Kalman Filter Strategy
In the world of algorithmic trading, a stellar backtest report can be both exhilarating and dangerously misleading. A strategy that performs brilliantly on historical data can often fail spectacularly in live trading. This gap between past performance and future results is frequently caused by overfitting, where a strategy’s parameters are so finely tuned to a specific historical dataset that they lose their predictive power.
To bridge this gap, we need a more rigorous testing methodology. This article breaks down a powerful technique called Walk-Forward Optimization (WFO) and explores a complete Python framework designed to implement it. This framework automatically extracts a strategy’s parameters, optimizes them on a slice of historical data, and then validates their performance on unseen, “out-of-sample” data, providing a much more realistic assessment of a strategy’s viability.
Section 1: The Problem with a Single Backtest
A standard backtest optimizes and evaluates a strategy on the same large block of historical data. This is like giving a student a test and then letting them study the exact same test questions before grading them. They are bound to score well, but it proves little about their actual knowledge.
Walk-forward optimization solves this by breaking the historical data into sequential chunks. The process is as follows:
- Optimize: Find the best strategy parameters on an “in-sample” period (e.g., 3 months of data).
- Validate: Apply these “best” parameters to the next, unseen “out-of-sample” period (the subsequent 3 months).
- Repeat: Slide the window forward and repeat the process.
The “out-of-sample” periods are the true test. The strategy has never seen this data before. Consistent performance across multiple out-of-sample periods gives a much higher degree of confidence that the strategy is robust and not merely overfitted.
Section 2: Automating the Process: The Walk-Forward Framework
The provided Python code creates a reusable framework that can apply
this walk-forward process to virtually any backtrader
strategy. It’s built on two main classes.
The
ParameterExtractor: Dynamic and Intelligent
Before we can optimize, we need to know what to optimize.
The ParameterExtractor class cleverly automates this by
inspecting the strategy code.
class ParameterExtractor:
"""Extract optimizable parameters from strategy class"""
@staticmethod
def get_strategy_parameters(strategy_class):
"""
Extract parameters from Backtrader strategy class params attribute
Returns dict with parameter names and default values
"""
try:
# ... [code to inspect strategy.params or __init__] ...
except Exception as e:
print(f"Error extracting parameters: {e}")
return {}
@staticmethod
def create_parameter_bounds(strategy_params, multiplier_range=(0.2, 3.0)):
"""
Create parameter bounds based on default values
"""
# ... [code to create a min/max range for each parameter] ...This class is the first step in creating a truly generic tool. It
automatically: 1. Finds Parameters: It looks inside a
given strategy for a params object (standard in
backtrader) or even the __init__ method to
find what parameters are available for optimization. 2. Sets
Bounds: It then intelligently creates a search space for the
optimization, defining a minimum and maximum value for each parameter,
typically as a multiple of its default value. This prevents the
optimizer from testing absurd parameter values (e.g., a 1-day moving
average or a 1000-day moving average if the default is 20).
The
SimpleWalkForwardBacktester: The Core Engine
This class is the heart of the framework. It orchestrates the entire walk-forward process.
class SimpleWalkForwardBacktester:
"""
Simple walk-forward backtester:
1. Optimize on window N
2. Trade on window N+1 with optimized parameters
3. Repeat
"""
def __init__(self, strategy_class, ticker="BTC-USD", ...):
self.strategy_class = strategy_class
# ... [initialization] ...
# Automatically extract parameters on creation
self.strategy_params = ParameterExtractor.get_strategy_parameters(strategy_class)
self.param_bounds = ParameterExtractor.create_parameter_bounds(self.strategy_params)
def run_walkforward_backtest(self, start="2020-01-01", end="2024-12-31",
window_months=3, ...):
# ... [main loop] ...
while True:
# Define optimization and trading windows
opt_end = current_start + rd.relativedelta(months=window_months)
trade_start = opt_end
trade_end = trade_start + rd.relativedelta(months=window_months)
# ...
# Step 1: Download optimization data
opt_data = self._download_data(current_start, opt_end)
# Step 2: Optimize parameters on current window
best_params = self._optimize_parameters(opt_data, ...)
# Step 3: Download trading period data
trade_data = self._download_data(trade_start, trade_end)
# Step 4: Run backtest on trading period with optimized parameters
result = self._run_single_backtest(trade_data, best_params)
# ... [store results and move to next period] ...
def _optimize_parameters(self, data, param_bounds, method='differential_evolution', ...):
# ... [uses scipy.optimize to find best parameters] ...
# The objective is to maximize the Sharpe Ratio
def objective_function(params_array):
result = self._run_single_backtest(data, ...)
return -result['sharpe_ratio'] # Negative because we minimize
# ... [calls differential_evolution or other optimizers] ...The workflow is clear: * The main
run_walkforward_backtest method contains a loop that
“walks” through the data. * In each loop, it calls
_optimize_parameters, which uses powerful scientific
libraries like SciPy to run dozens or hundreds of backtests
on the “in-sample” data. Its goal is to find the parameter set that
yields the highest Sharpe Ratio (a measure of
risk-adjusted return). * Once the best parameters are found, it runs a
single backtest on the “out-of-sample” data using those parameters and
records the performance. * This process repeats until the end of the
dataset is reached.
Section 3: Analysis and Visualization — Making Sense of the Results
Running the walk-forward test is only half the job. The final part of the framework provides tools to analyze and visualize the results, which is where the true insights are found.
def analyze_walkforward_results(df):
"""Analyze walk-forward backtest results"""
print("\nWALK-FORWARD BACKTEST RESULTS")
# ... [prints summary statistics like mean return, Sharpe, win rate] ...
cumulative_return = (1 + df['return_pct']/100).prod() - 1
print(f"Total Cumulative Return: {cumulative_return * 100:.2f}%")
def plot_walkforward_results(df):
"""Plot walk-forward results"""
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 1. Plot returns for each out-of-sample period
axes[0, 0].bar(...)
# 2. Plot the cumulative equity curve
cumulative = (1 + df['return_pct']/100).cumprod() - 1
axes[0, 1].plot(range(len(df)), cumulative * 100, ...)
# ... [other plots like rolling Sharpe and return distribution] ...
plt.tight_layout()
plt.show()
# --- Example Usage with a hypothetical strategy ---
if __name__ == '__main__':
from AdaptiveKalmanFilterStrategy import AdaptiveKalmanFilterStrategy
backtester = SimpleWalkForwardBacktester(
strategy_class=AdaptiveKalmanFilterStrategy, ...
)
results = backtester.run_walkforward_backtest(...)
analyze_walkforward_results(results)
plot_walkforward_results(results)The Strategy: Adaptive Kalman Filter
The Adaptive Kalman Filter strategy we covered before in an article, is used for this walk-forward backtest:
class AdaptiveKalmanFilterStrategy(bt.Strategy):
# declare plot‐lines and subplots
lines = (
'kf_price',
'kf_velocity',
'adaptive_R',
'adaptive_Q0',
'adaptive_Q1',
)
plotlines = dict(
kf_price = dict(_name='KF Price', subplot=False),
kf_velocity = dict(_name='KF Velocity', subplot=True),
adaptive_R = dict(_name='R', subplot=True),
adaptive_Q0 = dict(_name='Q[0,0]', subplot=True),
adaptive_Q1 = dict(_name='Q[1,1]', subplot=True),
)
params = dict(
vol_period = 20,
delta = 1e-4,
R_base = 0.1,
R_scale = 1.0,
Q_scale_factor = 0.5,
initial_cov = 1.0,
printlog = False,
)
def log(self, txt, dt=None, doprint=False):
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
def __init__(self):
# data
self.data_close = self.datas[0].close
# ——— Kalman state & matrices ———
self.x = np.zeros(2) # [level, velocity]
self.P = np.eye(2) * self.params.initial_cov
self.F = np.array([[1., 1.],
[0., 1.]])
self.H = np.array([[1., 0.]])
self.I = np.eye(2)
self.initialized = False
# Initialize Q and R so they'll exist before first next()
self.Q = np.eye(2) * self.params.delta
self.R = self.params.R_base
# ——— Indicators ———
# 1-bar log returns
self.log_returns = LogReturns(self.data_close, period=1)
# rolling volatility
self.volatility = bt.indicators.StandardDeviation(
self.log_returns.logret,
period=self.params.vol_period
)
def _initialize_kalman(self, price):
self.x[:] = [price, 0.0]
self.P = np.eye(2) * self.params.initial_cov
self.initialized = True
self.log(f'KF initialized at price={price:.2f}', doprint=True)
def next(self):
price = self.data_close[0]
# —— wait for enough bars to init KF & vol ——
if not self.initialized:
if len(self) > self.params.vol_period and not np.isnan(self.volatility[0]):
self._initialize_kalman(price)
return
vol = self.volatility[0]
# if vol or price is NaN, push NaNs to keep plot aligned
if np.isnan(vol) or np.isnan(price):
for ln in self.lines:
getattr(self.lines, ln)[0] = np.nan
return
# ——— Predict ———
self.x = self.F.dot(self.x)
self.P = self.F.dot(self.P).dot(self.F.T) + self.Q
# ——— Adapt Q & R ———
vol = max(vol, 1e-8)
self.R = self.params.R_base * (1 + self.params.R_scale * vol)
qvar = self.params.delta * (1 + self.params.Q_scale_factor * vol**2)
self.Q = np.diag([qvar, qvar])
# ——— Update ———
y = price - (self.H.dot(self.x))[0]
S = (self.H.dot(self.P).dot(self.H.T))[0, 0] + self.R
K = self.P.dot(self.H.T) / S
self.x = self.x + (K.flatten() * y)
self.P = (self.I - K.dot(self.H)).dot(self.P)
# ——— Record lines ———
self.lines.kf_price[0] = self.x[0]
self.lines.kf_velocity[0] = self.x[1]
self.lines.adaptive_R[0] = self.R
self.lines.adaptive_Q0[0] = self.Q[0, 0]
self.lines.adaptive_Q1[0] = self.Q[1, 1]
# ——— Trading: full long & short ———
vel = self.x[1]
if not self.position:
if vel > 0:
self.log(f'BUY (vel={vel:.4f})')
self.buy()
elif vel < 0:
self.log(f'SELL SHORT (vel={vel:.4f})')
self.sell()
elif self.position.size > 0 and vel < 0:
self.log(f'CLOSE LONG & SELL SHORT (vel={vel:.4f})')
self.close(); self.sell()
elif self.position.size < 0 and vel > 0:
self.log(f'CLOSE SHORT & BUY LONG (vel={vel:.4f})')
self.close(); self.buy()
def stop(self):
self.log(f'Ending Portfolio Value: {self.broker.getvalue():.2f}', doprint=True)
The output of this analysis is critical. Instead of one final number, you get a series of performance metrics, one for each out-of-sample period.
- The Period Returns Bar Chart is the most important visual. It shows the performance in each validation period. A robust strategy should show consistent profitability across most periods, not one huge gain and many losses.
- The Cumulative Returns Plot chains these period returns together to show what the strategy’s equity curve would have looked like under this more realistic testing scenario.
Conclusion: Embracing a More Honest Approach
This walk-forward framework is a powerful tool for any serious quantitative trader. It forces a strategy to prove its worth on unseen data, time and time again. By moving beyond a single, potentially overfitted backtest, it provides a more honest and reliable assessment of a strategy’s potential. While it doesn’t guarantee future success, it significantly increases the confidence that a strategy has a genuine edge and is not just a product of data mining. This framework allows a developer to test any strategy with this advanced methodology, bringing them one step closer to building systems that can withstand the test of time in live markets.