Time Series Momentum Effect

Introduction

Time series momentum is related to, but different from the phenomenon known as “momentum” in the finance literature, which is primarily cross-sectional in nature. The momentum literature focuses on the relative performance of securities in the cross section, finding that securities that recently outperformed their peers over the past 3 to 12 months continue to do so on average over the next month. Rather than focus on the relative returns of securities in the cross section, this time series momentum strategy focuses purely on the past returns of each individual Futures contract. Every month, the investor considers whether the excess return of each asset over the past 12 months is positive or negative and goes long on the contract if it is positive and short if negative. The position size is set to be inversely proportional to the volatility of the security's returns. A univariate GARCH model could be used to estimate volatility. However, other simple models could probably be easily used with good results (for example, the easiest one would be using historical volatility). For the sake of simplicity, we will use historical volatility. The portfolio is rebalanced monthly.

Method

We manually create a universe of tradable commodity Futures from all available commodity Futures traded on CME and ICE. The subscribed data resolution is daily.

Step 1: Import the data

As the strategy needs the continuous Futures contract, we subscribe to continuous Futures data by AddFuture in daily resolution.

for symbol in self.symbols:
    future = self.AddFuture(symbol,
        resolution = Resolution.Daily,
        extendedMarketHours = True,
        dataNormalizationMode = DataNormalizationMode.BackwardsRatio,
        dataMappingMode = DataMappingMode.OpenInterest,
        contractDepthOffset = 0
    )
    future.SetLeverage(1)

Step 2: Set the portfolio target volatility and decide rebalance schedule

def Initialize(self):
    # Last trading date tracker to achieve rebalancing the portfolio every month
    self.RebalancingTime = self.Time

Step 3: Set up SymbolData class to hold momentum and other information

Here we use a 12-month RateOfChange indicator to simulate the momentum returns. We will use a SymbolData class object to hold the indicator per contract, as well as other information like their canonical symbol and mapped contract's symbol. All SymbolData objects are saved in the dictionary self.symbol_data.

self.period = 252
self.symbol_data = {}
for symbol in self.symbols:
    self.symbol_data[future.Symbol] = SymbolData(self, future, self.period)

class SymbolData:
    def __init__(self, algorithm, future, period):
        self._future = future
        self.Symbol = future.Symbol
        self.ROC = algorithm.ROC(future.Symbol, period)
        algorithm.WarmUpIndicator(future.Symbol, self.ROC, Resolution.Daily)

    @property
    def Value(self):
        return self.ROC.Current.Value

    @property
    def Mapped(self):
        return self._future.Mapped

    @property
    def IsReady(self):
        return self.ROC.IsReady

Step 4: Construct/Rebalance the Portfolio

We set up handler for Future contracts rollover events, as well as flag-check of rebalancing time

def OnData(self, data):
    # Rollover for future contract mapping change
    for symbol, symbol_data in self.symbol_data.items():
        if data.SymbolChangedEvents.ContainsKey(symbol):
            changed_event = data.SymbolChangedEvents[symbol]
            old_symbol = changed_event.OldSymbol
            new_symbol = changed_event.NewSymbol
            tag = f"Rollover - Symbol changed at {self.Time}: {old_symbol} -> {new_symbol}"
            quantity = self.Portfolio[old_symbol].Quantity

            # Rolling over: to liquidate any position of the old mapped contract and switch to the newly mapped contract
            self.Liquidate(old_symbol, tag = tag)
            self.MarketOrder(new_symbol, quantity // self.Securities[new_symbol].SymbolProperties.ContractMultiplier, tag = tag)

    # skip if less than 30 days passed since the last trading date
    if self.Time < self.RebalancingTime:
        return

We use a history request to obtain historical prices. Here, history is the daily close-price returns, which will be used to calculate volatilities.

history = self.History(list(self.symbol_data.keys()), self.period, Resolution.Daily)
history = history.droplevel([0]).unstack(level=0).close.pct_change().dropna()

Then we calculate the historical volatilities and place orders. Note that the weights are inversely proportional to volatilities and np.sign determines whether to long or short.

# Compute the inverse of the volatility
# The weights are the normalized inverse of the volatility
vol_inv = 1 / history.std(ddof=1)
vol_sum = vol_inv.sum()
weights = (vol_inv / vol_sum).fillna(0).to_dict()

#np.sign(roc.Current.Value) tells us whether to long or short
for symbol, symbol_data in self.symbol_data.items():
    symbol_id = symbol.ID.ToString()
    if symbol_id in weights:
        weight = np.sign(symbol_data.Value) * weights[symbol_id] *.5

        mapped = symbol_data.Mapped
        qty = self.CalculateOrderQuantity(mapped, np.clip(weight, -1, 1))
        multiplier = self.Securities[mapped].SymbolProperties.ContractMultiplier
        order_qty = (qty - self.Portfolio[mapped].Quantity) // multiplier
        self.MarketOrder(mapped, order_qty)

At last, we need to reset the rebalancing timestamp

# Set next rebalance time
self.RebalancingTime = Expiry.EndOfMonth(self.Time)

---

---

---