The strategy uses bilateral city distances and GDP size to compute a Gravity score for each country. Countries are ranked based on this score, and the top quintile is longed, rebalanced monthly.
ASSET CLASS: ETFs, stocks | REGION: Global | FREQUENCY:
Monthly | MARKET: equities | KEYWORD: Geographical, Momentum
I. 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.
II. 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.
III. SOURCE PAPER
Gravity in International Equity Markets [Click to Open PDF]
Joon Woo Bae,Rotman School of Management , University of Toronto
<Abstract>
The size of economies and geographical distance are significant determinants of the contemporaneous and cross-serial correlations in international equity market returns across countries. Larger countries lead returns of small-countries, and this cross-country predictability decreases with geographical distance of the two countries. A long-short trading strategy that exploits this relation yields risk-adjusted returns of 10% per annum. The lead-lag relation is not driven by cross-country differences in the average size or liquidity of rms, the degree of stock market development, or the industry composition. Decomposing stock market returns into cash-flow and discount rate news shows that the international transmission of discount-rate news is more pronounced than cash-flow news and that the size of economies and geographical distance are signi cant determinants for both components of returns
IV. BACKTEST PERFORMANCE
| Annualised Return | 12.69% |
| Volatility | 19.83% |
| Beta | 0.912 |
| Sharpe Ratio | 0.64 |
| Sortino Ratio | N/A |
| Maximum Drawdown | N/A |
| Win Rate | 58% |
V. 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'