Geopolitical Risk and Commodities
Log in to collectAcademic paper
Strategy in a nutshell
The investment universe consists of 29 commodity futures contracts across four categories: agriculture, energy, livestock, and metals. Data is sourced from the Chicago Mercantile Exchange (CME).
The historical geopolitical risk index (GPRH) is constructed using the proportion of articles on geopolitical risks from The New York Times, The Chicago Tribune, and The Washington Post, via ProQuest Historical Newspapers, dating back to 1900. The GPRH data is sourced from Caldara and Iacoviello.
To estimate the GPRH beta, a rolling regression is run:
Independent variable: Monthly excess return of each commodity.
Dependent variables: Change in GPRH (month t – t-1), average factor (excess return of a long position in all commodity futures), carry factor, and commodity-momentum factor.
Estimation window: 60 months with at least 24 observations.
Each month, commodities are sorted into three portfolios based on their previous month’s GPRH beta:
Low: 5 contracts with lowest GPRH beta.
High: 5 contracts with highest GPRH beta.
Medium: All other contracts.
The strategy is long high-beta portfolios and short low-beta portfolios, equally weighted and rebalanced monthly.
Economic rationale
The strategy exploits the geopolitical risk premium, rooted in uncertainty-driven investment behavior.
Investors recognize the impact of geopolitical events (e.g., wars, terrorist attacks) on economic uncertainty and asset returns. As a result:
Risk-averse investors demand higher expected returns for assets positively correlated with geopolitical risk.
Assets negatively correlated with geopolitical risk are viewed as safer and can command higher prices.
This aligns with preference-based theory and the intertemporal CAPM model (Merton, 1973): as uncertainty rises, investors prefer higher-yielding assets to hedge future investment and consumption possibilities. The strategy captures returns arising from these risk-adjusted preferences.
Backtest performance
Annualised return8.21%
Volatility8.62%
Beta-0.004
Sharpe ratio0.95
Win rate50%
Full Python code
from AlgorithmImports import *
#endregion
import data_tools
import statsmodels.api as sm
class GeopoliticalRiskAndCommodities(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.leverage:int = 5
self.part_length = 5
self.max_missing_days = 5
self.month_period = 21
self.regression_period = 60
self.period = 6 * self.month_period
self.data = {}
self.symbols = [
"CME_S1", # Soybean Futures, Continuous Contract #1
"CME_W1", # Wheat Futures, Continuous Contract #1
"CME_SM1", # Soybean Meal Futures, Continuous Contract #1
"CME_BO1", # Soybean Oil Futures, Continuous Contract #1
"CME_C1", # Corn Futures, Continuous Contract #1
"CME_O1", # Oats Futures, Continuous Contract #1
"CME_LC1", # Live Cattle Futures, Continuous Contract #1
"CME_FC1", # Feeder Cattle Futures, Continuous Contract #1
"CME_LN1", # Lean Hog Futures, Continuous Contract #1
"CME_GC1", # Gold Futures, Continuous Contract #1
"CME_SI1", # Silver Futures, Continuous Contract #1
"CME_PL1", # Platinum Futures, Continuous Contract #1
"CME_CL1", # Crude Oil Futures, Continuous Contract #1
"CME_HG1", # Copper Futures, Continuous Contract #1
"CME_LB1", # Random Length Lumber Futures, Continuous Contract #1
"CME_NG1", # Natural Gas (Henry Hub) Physical Futures, Continuous Contract #1
"CME_PA1", # Palladium Futures, Continuous Contract #1
"CME_RR1", # Rough Rice Futures, Continuous Contract #1
"CME_CU1", # Chicago Ethanol (Platts) Futures
"CME_DA1", # Class III Milk Futures
"ICE_CC1", # Cocoa Futures, Continuous Contract #1
"ICE_CT1", # Cotton No. 2 Futures, Continuous Contract #1
"ICE_KC1", # Coffee C Futures, Continuous Contract #1
"ICE_O1", # Heating Oil Futures, Continuous Contract #1
"ICE_OJ1", # Orange Juice Futures, Continuous Contract #1
"ICE_SB1" # Sugar No. 11 Futures, Continuous Contract #1
]
for symbol in self.symbols:
# subscribe price data
data = self.AddData(data_tools.QuantpediaFutures, symbol, Resolution.Daily)
data.SetFeeModel(data_tools.CustomFeeModel())
data.SetLeverage(self.leverage)
# quandl data
quandl_symbol_1 = 'CHRIS/' + symbol
quandl_symbol_2 = 'CHRIS/' + symbol.replace('1', '2')
# subscribe to quandl data
self.AddData(data_tools.QuandlFutures, quandl_symbol_1, Resolution.Daily)
self.AddData(data_tools.QuandlFutures, quandl_symbol_2, Resolution.Daily)
# store SymbolData
self.data[symbol] = data_tools.SymbolData([quandl_symbol_1, quandl_symbol_2], self.month_period, self.period, self.regression_period)
# subscribe to geopolitical risk data
self.geo_risk_index_symbol = self.AddData(data_tools.QuantpediaGeopoliticalRisk, 'GEOPOLITICAL_RISK_INDEX', Resolution.Daily).Symbol
self.geo_risk_index_data = data_tools.IndexData(self.month_period)
self.regression_x_data = data_tools.RegressionData(self.regression_period)
# rebalancing monthly based on self.curr_month variable
self.curr_month = -1
def OnData(self, data):
curr_date = self.Time.date()
# rebalance monthly
if self.curr_month != self.Time.month:
self.curr_month = self.Time.month
self.Rebalance(curr_date)
# check if geopolitical risk index data are ready
if self.geo_risk_index_symbol in data and data[self.geo_risk_index_symbol]:
# collect index data
index_value = data[self.geo_risk_index_symbol].Value
self.geo_risk_index_data.update(index_value, curr_date)
for symbol in self.data:
q_sym1, q_sym2 = self.data[symbol].quandl_symbols
# update futures daily prices
if symbol in data and data[symbol]:
close = data[symbol].Value
self.data[symbol].update(close, curr_date)
# calculate carry factor only when carry value wasn't set in this month
if self.data[symbol].carry_value == None and q_sym1 in data and q_sym2 in data and data[q_sym1] and data[q_sym2]:
value1 = data[q_sym1].Value
value2 = data[q_sym2].Value
self.data[symbol].update_carry_value(np.log(value2 / value1))
def Rebalance(self, curr_date):
''' collects needed data for regression, then perform regresion, selection and trade '''
# continue only if geopolitical index values are ready and new geopolitical index values are still coming
if not self.geo_risk_index_data.is_ready(curr_date, self.max_missing_days):
self.Liquidate()
return
symbol_values = {}
for symbol, symbol_obj in self.data.items():
# make sure future data are ready and are still coming
if symbol_obj.is_ready(curr_date, self.max_missing_days):
future_closes = [x for x in symbol_obj.closes]
# calculate future's monthly performance
short_term_momentum = self.CalculatePerformance(future_closes, self.month_period)
# calculate long term momentum for momentum factor
long_term_momentum = self.CalculatePerformance(future_closes, self.period)
# carry
carry_value = symbol_obj.carry_value
symbol_values[symbol] = data_tools.SymbolValue(short_term_momentum, carry_value, long_term_momentum)
# update future's performance
symbol_obj.update_momentum_values(short_term_momentum)
symbol_obj.reset_carry_value()
# to make sure X regression data has same length update only when condition isn't true
if len(symbol_values) < (self.part_length * 2):
self.Liquidate()
return
# update factor values needed for regression
average_factor_perf = np.mean([symbol_values[symbol].short_term_performance for symbol in symbol_values])
self.regression_x_data.update_average_factor_perf(average_factor_perf)
carry_factor_perf = self.CarryFactorCalculation(symbol_values)
self.regression_x_data.update_carry_factor_perf(carry_factor_perf)
momentum_factor_perf = self.MomentumFactorCalculation(symbol_values)
self.regression_x_data.update_momentum_factor_perf(momentum_factor_perf)
# update index change
index_values = [x for x in self.geo_risk_index_data.index_values]
index_change = self.CalculatePerformance(index_values, self.month_period)
self.regression_x_data.update_index_changes(index_change)
# make sure data, for variable X in regression, are ready
if not self.regression_x_data.is_ready():
self.Liquidate()
return
# create data structure from data for variable X in regression
regression_x: list[list[float]] = [
[x for x in self.regression_x_data.index_changes],
[x for x in self.regression_x_data.average_factor_perf],
[x for x in self.regression_x_data.carry_factor_perf],
[x for x in self.regression_x_data.momentum_factor_perf]
]
betas = {}
for symbol in symbol_values:
# make sure future has enough data for regression
if not self.data[symbol].are_momentum_values_ready():
continue
regression_y = [x for x in self.data[symbol].momentum_values]
regression_model = self.MultipleLinearRegression(regression_x, regression_y)
# get geopolitical risk index beta
beta = regression_model.params[1]
betas[symbol] = beta
# make sure there are enough futures for selection
if len(betas) < (self.part_length * 2):
self.Liquidate()
return
sorted_by_beta = [x[0] for x in sorted(betas.items(), key=lambda item: item[1])]
# long high portfolio
long_part = sorted_by_beta[-self.part_length:]
# short low portfolio
short_part = sorted_by_beta[:self.part_length]
# trade execution
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long_part + short_part:
self.Liquidate(symbol)
for symbol in long_part:
self.SetHoldings(symbol, 1 / self.part_length)
for symbol in short_part:
self.SetHoldings(symbol, -1 / self.part_length)
def CalculatePerformance(self, values: list, period: int) -> float:
values = values[:period]
performance = (values[0] - values[-1]) / values[-1]
return performance
def CarryFactorCalculation(self, symbol_values: dict) -> float:
''' based on carry factor change value select backwardation and contango futures and calculate total performance for current month '''
sorted_by_carry_value = [x[0] for x in sorted(symbol_values.items(), key=lambda item: item[1].carry_value)]
# long commodities that are most in backwardation
long_part = sorted_by_carry_value[:self.part_length]
# short the ones that are most in contango
short_part = sorted_by_carry_value[-self.part_length:]
carry_factor_perf = self.CalculateFactorReturn(symbol_values, long_part, short_part)
return carry_factor_perf
def MomentumFactorCalculation(self, symbol_values: dict) -> float:
''' select top n and low n futures based on momentum factor value and calculate their total performance for current month '''
sorted_by_mom = [x[0] for x in sorted(symbol_values.items(), key=lambda item: item[1].long_term_performance)]
# long highest n futures
long_part = sorted_by_mom[-self.part_length:]
# short lowest n futures
short_part = sorted_by_mom[:self.part_length]
momentum_factor_perf = self.CalculateFactorReturn(symbol_values, long_part, short_part)
return momentum_factor_perf
def CalculateFactorReturn(self, symbol_values: dict, long_part: list, short_part: list) -> float:
''' calculate total performance of factor for current month '''
long_value = np.mean([symbol_values[symbol].short_term_performance for symbol in long_part])
short_value = np.mean([-symbol_values[symbol].short_term_performance for symbol in short_part])
factor_value = (long_value + short_value) / 2
return factor_value
def MultipleLinearRegression(self, x, y):
x = np.array(x).T
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result