Optimization
Walk Forward Optimization
Introduction
Walk forward optimization is the practice of periodically adjusting the logic or parameters of a strategy to optimize some objective function over a trailing window of time. For example, say your strategy is to hold a long position when an asset is trading above its simple moving average (SMA) and to hold a short position when it's trading below the SMA. If your objective is to maximize returns, how many trailing data points should you use to compute the SMA?
Benefits
The parameter values that maximize your algorithm's performance are typically a function of the start and end dates of your training dataset. Market conditions change over time and a strategy that works well during one period may not perform as well during another. By continually adjusting the strategy parameters based on recent data, walk-forward optimization tailors your strategy to the latest market trends and patterns.
Optimization Frequency
There is a tradeoff when it comes to how often you optimize your algorithm's parameters. On one hand, if you frequently optimize your parameters, your parameters will be fit to the most recent, most meaningful market data. On the other hand, if you wait a long time between optimization sessions, your algorithm will run faster and you reduce the chance of overfitting. To define your optimization schedule, pass DateRules and TimeRules arguments to the train method.
// To define the optimization schedule, pass self.date_rules and self.time_rules arguments to the train method. symbol = Symbol.create('SPY', SecurityType.EQUITY, Market.USA) self.train( self.date_rules.month_start(symbol), self.time_rules.midnight, lambda: self._do_wfo(self._optimization_func, max, objective) )
EMA Crossover Example
Follow these steps to implement walk forward optimization for an exponential moving average (EMA) crossover strategy:
- Define the backtest dates and starting cash.
- Enable the automatic indicator warm up setting.
- Subscribe to an asset.
- Add some members for the EMA indicators.
- Define the trading rules.
- Define a method that will adjust the algorithm's behavior, given a set of optimal parameters.
- Back in
initialize
method, generate all the possible combinations of parameter values. - Define the objective function.
- Define the optimization function.
- Schedule the optimization sessions.
- Add a warm-up period.
# Set the start date, end date, and starting cash. def initialize(self): self.set_start_date(2000, 1, 1) self.set_end_date(2024, 1, 1) self.set_cash(100_000)
# Set the automatic_indicator_warm_up property to true to enable automatic algorithm warm-up. def initialize(self): # . . . self.settings.automatic_indicator_warm_up = True
# Add daily SPY Equity data. def initialize(self): # . . . self._security = self.add_equity("SPY", Resolution.DAILY) self._symbol = self._security.symbol
# Initialize members for short and long EMA indicators. def initialize(self): # . . . self._short_ema = None self._long_ema = None
# Define the trading logic. def on_data(self, data): if self.is_warming_up: return # Case 1: Short EMA is above long EMA. if (self._short_ema > self._long_ema and not self._security.holdings.is_long): self.set_holdings(self._symbol, 1) # Case 2: Short EMA is below long EMA. elif (self._short_ema < self._long_ema and not self._security.holdings.is_short): self.set_holdings(self._symbol, 0)
# Define a method to adjust the algorithm's behavior based on the new optimal parameters. def _update_algorithm_logic(self, optimal_parameters): # Remove the old indicators. if self._short_ema: self.deregister_indicator(self._short_ema) if self._long_ema: self.deregister_indicator(self._long_ema) # Create the new indicators. self._short_ema = self.ema( self._symbol, optimal_parameters['long_ema'], Resolution.DAILY ) self._long_ema = self.ema( self._symbol, optimal_parameters['short_ema'], Resolution.DAILY )
# In the initialize method, generate all possible combinations of parameter values. import itertools def initialize(self): # . . . self._parameter_sets = self._generate_parameter_sets( { 'short_ema': (10, 50, 10), # min, max, step 'long_ema': (60, 200, 10) } ) # . . . def _generate_parameter_sets(self, search_space): # Create ranges for each parameter. ranges = { parameter_name: np.arange(min_, max_ + step_size, step_size) for parameter_name, (min_, max_, step_size) in search_space.items() } # Calculate the cartesian product and create a list of dictionaries for # the parameter sets. return [ dict(zip(ranges.keys(), combination)) for combination in list(itertools.product(*ranges.values())) ]
# Define an objective function to optimize the algorithm's overall performance. def initialize(self): # . . . objective = self._cumulative_return # . . . def _cumulative_return(self, daily_returns): return (daily_returns + 1).cumprod()[-1] - 1
# Evaluate parameter sets with the objective function. def _optimization_func(self, data, parameter_set, objective): p1 = parameter_set['short_ema'] p2 = parameter_set['long_ema'] short_ema = data['close'].ewm(p1, min_periods=p1).mean() long_ema = data['close'].ewm(p2, min_periods=p2).mean() exposure = (short_ema - long_ema).dropna().apply(np.sign)\ .replace(0, pd.NA).ffill().shift(1) # ^ shift(1) because we enter the position on the next day. asset_daily_returns = data['open'].pct_change().shift(-1) # ^ shift(-1) because we want each entry to be the return from # the current day to the next day. strategy_daily_returns = (exposure * asset_daily_returns).dropna() return objective(strategy_daily_returns)
# Schedule the WFO process to run at the start of each month at midnight. def initialize(self): # . . . self.train( self.date_rules.month_start(self._symbol), self.time_rules.midnight, lambda: self._do_wfo(self._optimization_func, max, objective) ) # . . . def _do_wfo(self, optimization_func, min_max, objective): # Get the historical data you need to calculate the scores. prices = self.history( self._symbol, timedelta(365), Resolution.DAILY ).loc[self._symbol] # Calculate the score of each parameter set. scores = [ optimization_func(prices, parameter_set, objective) for parameter_set in self._parameter_sets ] # Find the parameter set that minimizes/maximizes the objective function. optimal_parameters = self._parameter_sets[scores.index(min_max(scores))] # Adjust the algorithm's logic. self._update_algorithm_logic(optimal_parameters)
The warm-up period ensures that the algorithm calls the _do_wfo
method at least one time before the algorithm starts trading.
# Set the warm-up period. def initialize(self): # . . . self.set_warm_up(timedelta(45))
For a full working example algorithm, see the following backtest:
Ideas
The following list provides some ideas on how you could use walk forward optimization:
- Find the range of PE ratios that mazimize your algorithm's performance, and then use that range in your fundamental universe filter.
- Experiment with asset universe diversity and counts.
- Take 10 fundamental factors (PE, market cap, etc...). For each optimization session, step through and picks one factor at a time, using it as a ranking factor for the universe. Find the factor that leads to the greatest return over the lookback window and use it as the sole factor for the following month.