Geographical Country Momentum
Log in to collectAcademic paper
Strategy in a nutshell
Trades 44 country equity indices using a Gravity measure that combines GDP size and distance between countries’ largest cities. Goes long on the top quintile of indices based on weighted past returns, with value-weighted portfolios rebalanced monthly.
Economic rationale
Countries’ economic size and proximity drive co-movements and return predictability. Larger economies exert influence on nearby smaller ones, and geographic distance proxies barriers, allowing weighted past returns to forecast future performance.
Backtest performance
Annualised return12.69%
Volatility19.83%
Beta0.912
Sharpe ratio0.64
Win rate58%
Full Python code
from AlgorithmImports import *
import numpy as np
#endregion
class GeographicalCountryMomentum(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.symbols = {
"Australia" : "EWA", # iShares MSCI Australia Index ETF
"Austria" : "EWO", # iShares MSCI Austria Investable Mkt Index ETF
"Belgium" : "EWK", # iShares MSCI Belgium Investable Market Index ETF
"Brazil" : "EWZ", # iShares MSCI Brazil Index ETF
"Canada" : "EWC", # iShares MSCI Canada Index ETF
"China" : "FXI", # iShares China Large-Cap ETF
"France" : "EWQ", # iShares MSCI France Index ETF
"Germany" : "EWG", # iShares MSCI Germany ETF
"Hong Kong" : "EWH", # iShares MSCI Hong Kong Index ETF
"Italy" : "EWI", # iShares MSCI Italy Index ETF
"Japan" : "EWJ", # iShares MSCI Japan Index ETF
"Malaysia" : "EWM", # iShares MSCI Malaysia Index ETF
"Mexico" : "EWW", # iShares MSCI Mexico Inv. Mt. Idx
"Netherlands" : "EWN", # iShares MSCI Netherlands Index ETF
"Singapore" : "EWS", # iShares MSCI Singapore Index ETF
"South Africa" : "EZA", # iShares MSCI South Africa Index ETF
"South Korea" : "EWY", # iShares MSCI South Korea ETF
"Spain" : "EWP", # iShares MSCI Spain Index ETF
"Sweden" : "EWD", # iShares MSCI Sweden Index ETF
"Switzerland" : "EWL", # iShares MSCI Switzerland Index ETF
"Taiwan" : "EWT", # iShares MSCI Taiwan Index ETF
"Thailand" : "THD", # iShares MSCI Thailand Index ETF
"United Kingdom" : "EWU", # iShares MSCI United Kingdom Index ETF
"United States" : "SPY", # SPDR S&P 500 ETF
}
self.country_data = {}
self.period = 21 # performance period
self.SetWarmUp(self.period, Resolution.Daily)
self.quantile = 5
self.max_missing_days = 365
csv_string_file = self.Download('data.quantpedia.com/backtesting_data/economic/city_distance_420.csv')
lines = csv_string_file.split('\r\n')
# header and line example:
# country;country_population;biggest_city;city_population;GDP_symbol;sydney_d;vienna_d;brussel_d;sao_paulo_d;toronto_d;shanghai_d;paris_d;berlin+_d;hong_kong_d;rome_d;tokyo_d;kota_bharu_d;mexico_city_d;amsterdam_d;singapore_d;cape_town_d;seoul_d;madrid_d;stockholm_d;zurich_d;taipei_d;bangkok_d;london_d;new_york_d
# Australia;24990000;Sydney;4627345;ODA/AUS_PPPGDP;0;15996;16734;13349;15558;7876;16950;16084;7371;16311;7819;6794;12965;16632;6302;11005;8324;17674;15586;16557;7255;7532;16983;15979
# skip header.
for line in lines[1:]:
split_line = line.split(';')
country = split_line[0]
country_pop = float(split_line[1])
agglomeration = split_line[2]
agglomeration_pop = float(split_line[3])
gdp_symbol = split_line[4]
self.country_data[country] = CountryData(self.period, country, country_pop, agglomeration, agglomeration_pop, gdp_symbol)
line_index = 5
for symbol_index, symbol in enumerate(self.symbols):
self.country_data[country].AgglomerationDistance[symbol] = float(split_line[line_index + symbol_index])
# etf data.
data = self.AddEquity(self.symbols[country], Resolution.Daily)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(5)
# gdp quandl data.
self.AddData(QuandlValue, gdp_symbol, Resolution.Daily)
# aggregate distance calc.
for country_i in self.country_data:
for country_d in self.country_data[country_i].AgglomerationDistance:
country_i_ratio = self.country_data[country_i].AgglomerationPopulation / self.country_data[country_i].CountryPopulation
country_d_ratio = self.country_data[country_d].AgglomerationPopulation / self.country_data[country_d].CountryPopulation
distance_ratio = country_i_ratio * country_d_ratio * self.country_data[country_i].AgglomerationDistance[country_d]
self.country_data[country_i].AgglomerationDistance[country_d] = distance_ratio
self.recent_month = -1
def OnData(self, data):
for country in self.country_data:
etf_obj = self.Symbol(self.symbols[country])
if etf_obj in data.Bars and data[etf_obj]:
price = data.Bars[etf_obj].Value
self.country_data[country].update(price)
if self.recent_month == self.Time.month:
return
self.recent_month = self.Time.month
# size zscore calc.
size = { x : np.log(self.Securities[self.country_data[x].GDPSymbol].Price) \
for x in self.country_data if self.Securities.ContainsKey(self.country_data[x].GDPSymbol) and self.Securities[self.country_data[x].GDPSymbol].GetLastData() and (self.Time.date() - self.Securities[self.country_data[x].GDPSymbol].GetLastData().Time.date()).days < self.max_missing_days}
size_values = [x[1] for x in size.items()]
size_avg = np.average(size_values)
size_std = np.std(size_values)
z_score_size = { x[0] : (x[1] - size_avg) / size_std for x in size.items() }
# distance zscore calc.
z_score_dist = {}
dist_avg = { x : np.average([np.log(y[1]) for y in self.country_data[x].AgglomerationDistance.items() if y[0] != x]) for x in self.country_data }
dist_std = { x : np.std([np.log(y[1]) for y in self.country_data[x].AgglomerationDistance.items() if y[0] != x]) for x in self.country_data }
for country_i in self.country_data:
z_score_dist[country_i] = {}
for country_d in self.country_data[country_i].AgglomerationDistance:
if country_i != country_d:
z_score_dist[country_i][country_d] = (self.country_data[country_i].AgglomerationDistance[country_d] - dist_avg[country_i]) / dist_std[country_i]
# gravity score calc.
z_score_grav = {}
for country_i in z_score_dist:
z_score_grav[country_i] = {}
for country_d in z_score_dist[country_i]:
if country_i in z_score_size:
if country_i != country_d:
z_score_grav[country_i][country_d] = z_score_size[country_i] - z_score_dist[country_i][country_d]
# weight calc.
w_i = {}
for country_i in z_score_grav:
if len(z_score_grav[country_i]) != 0:
min_grav_i = min([x[1] for x in z_score_grav[country_i].items() if x[0] != country_i])
adj_grav_i = np.array([x[1] + min_grav_i for x in z_score_grav[country_i].items() if x[0] != country_i])
w_i[country_i] = adj_grav_i / sum(adj_grav_i)
gravity = {x[0] : sum(x[1] * self.country_data[x[0]].performance()) for x in w_i.items() if self.country_data[x[0]].is_ready()}
if len(gravity) < self.quantile:
self.Liquidate()
return
sorted_by_gravity = sorted(gravity.items(), key = lambda x: x[1], reverse = True)
quintile = int(len(sorted_by_gravity) / self.quantile)
long = [self.symbols[x[0]] for x in sorted_by_gravity][:quintile]
# Trade execution.
invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long:
self.Liquidate(symbol)
for etf in long:
if etf in data and data[etf]:
self.SetHoldings(etf, 1 / len(long))
class CountryData():
def __init__(self, period, country, country_pop, agglomeration, agglomeration_pop, gdp_symbol):
self.Country = country
self.CountryPopulation = country_pop
self.Agglomeration = agglomeration
self.AgglomerationPopulation = agglomeration_pop
self.GDPSymbol = gdp_symbol
self.AgglomerationDistance = {}
self.Price = RollingWindow[float](period)
def update(self, price):
self.Price.Add(price)
def is_ready(self) -> bool:
return self.Price.IsReady
def performance(self, values_to_skip = 0) -> float:
closes = [x for x in self.Price][values_to_skip:]
return (closes[0] / closes[-1] - 1)
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
# Quandl "value" data
class QuandlValue(NasdaqDataLink):
def __init__(self):
self.ValueColumnName = 'Value'