Optimization

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.

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)
)

Follow these steps to implement walk forward optimization for an exponential moving average (EMA) crossover strategy:

1. Define the backtest dates and starting cash.

# 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)

2. Enable the automatic indicator warm up setting.

# 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

3. Subscribe to an asset.

# Add daily SPY Equity data.
def initialize(self):
    # . . .
    self._security = self.add_equity("SPY", Resolution.DAILY)
    self._symbol = self._security.symbol

4. Add some members for the EMA indicators.

# Initialize members for short and long EMA indicators.
def initialize(self):
    # . . .
    self._short_ema = None
    self._long_ema = None

5. Define the trading rules.

# 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)

6. Define a method that will adjust the algorithm's behavior, given a set of optimal parameters.

# 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
    )

7. Back in initialize method, generate all the possible combinations of parameter values.

# 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()))
    ]

8. Define the objective function.

# 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

9. Define the optimization function.

# 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)

10. Schedule the optimization sessions.

# 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)

11. Add a warm-up period.

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:

The following list provides some ideas on how you could use walk forward optimization:

1. Find the range of PE ratios that mazimize your algorithm's performance, and then use that range in your fundamental universe filter.
2. Experiment with asset universe diversity and counts.
3. 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.