Trade currencies based on the one-year volatility risk premium (VRP), going long on currencies with the highest VRP and short on those with the lowest, with equally weighted, monthly rebalanced portfolios.
ASSET CLASS: CFDs, forwards, futures | REGION: Global | FREQUENCY:
Monthly | MARKET: currencies | KEYWORD: Volatility
I. STRATEGY IN A NUTSHELL
Long currencies with highest Volatility Risk Premium (VRP), short those with lowest VRP across 10 major currencies vs USD. Equally weighted, monthly rebalanced.
II. ECONOMIC RATIONALE
Currencies with cheap volatility insurance tend to appreciate, while those with expensive insurance tend to depreciate. Returns are influenced by funding liquidity, risk aversion, and capital flows, rather than traditional risk factors, highlighting limits to arbitrage and market frictions.
III. SOURCE PAPER
Volatility Risk Premia and Exchange Rate Predictability [Click to Open PDF]
Della Corte, Pasquale, Imperial College Business School; Ramadorai, Tarun, Imperial College London; Sarno, Lucio, University of Cambridge – Judge Business School
<Abstract>
We discover a new currency strategy with highly desirable return and diversification properties, which uses the predictive capability of currency volatility risk premia for currency returns. The volatility risk premium — the difference between expected realized volatility and model-free implied volatility — reflects the costs of insuring against currency volatility fluctuations, and the strategy sells high-insurance-cost currencies and buys low-insurance-cost currencies. The returns to the strategy are mainly generated by movements in spot exchange rates rather than interest rate differentials, and the strategy carries a large weight in a minimum-variance portfolio of commonly employed currency strategies. We explore alternative explanations for the profitability of the strategy, which cannot be understood using traditional risk factors.
IV. BACKTEST PERFORMANCE
| Annualised Return | 4.95% |
| Volatility | 8.15% |
| Beta | -0.051 |
| Sharpe Ratio | 0.61 |
| Sortino Ratio | -0.159 |
| Maximum Drawdown | -17% |
| Win Rate | 51% |
V. FULL PYTHON CODE
from AlgorithmImports import *
class VolatilityRiskPremiumInCurrencies2(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.min_expiry = 25
self.max_expiry = 35
self.period = 12 * 21 # need 12 months of daily prices
self.prices = {} # storing daily prices
self.contracts = {} # storing option contracts
self.symbols_by_ticker = {} # storing symbols under their tickers
self.tickers_currencies = {} # storing currencies symbols keyed by etf tickers
self.vol_difference = {} # storing volatility differences for each etf
self.etf_currencies = {
'FXA': "CME_AD1", # Australia
'FXC': "CME_CD1", # Canada
'FXE': "CME_EC1", # Euro
'FXY': "CME_JY1", # Japan
'BNZ': "CME_NE1", # New Zealand
'FXF': "CME_SF1", # Switzerland
'FXB': "CME_BP1", # Great Britain
}
for etf_ticker, currency_ticker in self.etf_currencies.items():
# subscribe to etf
security = self.AddEquity(etf_ticker, Resolution.Minute)
# change normalization to raw to allow adding etf contracts
security.SetDataNormalizationMode(DataNormalizationMode.Raw)
# get etf symbol
etf_symbol = security.Symbol
# Subscribe to future
security = self.AddData(QuantpediaFutures, currency_ticker, Resolution.Daily)
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(5)
currency_symbol = security.Symbol
# store etf symbol under etf ticker
self.symbols_by_ticker[etf_ticker] = etf_symbol
# create RollingWindow for daily prices
self.prices[etf_symbol] = RollingWindow[float](self.period)
# create pair etf ticker and currency symbol
self.tickers_currencies[etf_ticker] = currency_symbol
# create object from Contracts class for etf symbol
self.contracts[etf_symbol] = Contracts(self.Time.date(), 0, [])
self.day = -1
self.selection_flag = False
def OnData(self, data):
# not every IV comes at 9:30...
if self.selection_flag and self.Time.hour == 9:
for kvp in data.OptionChains:
chain = kvp.Value
ticker = chain.Underlying.Symbol.Value
# get etf symbol
symbol = self.symbols_by_ticker[ticker]
# currency symbol
currency_symbol = self.tickers_currencies[ticker]
# check if data is still coming
if self.securities[currency_symbol].get_last_data() and self.time.date() > QuantpediaFutures.get_last_update_date()[currency_symbol]:
self.liquidate()
return
# get contracts
contracts = [x for x in chain]
# check if there are enough contracts for option, daily prices are ready and volatility difference wasn't calculated
if len(contracts) < 2 or not self.prices[symbol].IsReady or currency_symbol in self.vol_difference:
continue
# get call and put implied volatility
call_iv, put_iv = self.GetImpliedVolatilities(contracts)
if call_iv and put_iv:
# make mean from call implied volatility and put implied volatility
iv = (call_iv + put_iv) / 2
# get historical volatility
hv = self.GetHistoricalVolatility(self.prices[symbol])
# store difference between historical and implied volatility
self.vol_difference[currency_symbol] = hv - iv
elif self.selection_flag and self.Time.hour != 9:
# can't perform selection
if len(self.vol_difference) < 5:
self.vol_difference.clear()
self.Liquidate()
return
# perform selection
quintile = int(len(self.vol_difference) / 5)
sorted_by_vol_dif = [x[0] for x in sorted(self.vol_difference.items(), key=lambda item: item[1])]
# long top etfs
long = sorted_by_vol_dif[-quintile:]
# short bottom etfs
short = sorted_by_vol_dif[:quintile]
# trade execution
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long + short:
self.Liquidate(symbol)
long_length = len(long)
short_length = len(short)
for symbol in long:
self.SetHoldings(symbol, 1 / long_length)
for symbol in short:
self.SetHoldings(symbol, -1 / short_length)
# clear dictionary with volatility differences and make sure, there will be no rebalance before contracts expirations
self.vol_difference.clear()
self.selection_flag = False
# execute once a day
if self.day == self.Time.day:
return
self.day = self.Time.day
for _, symbol in self.symbols_by_ticker.items():
if symbol in self.contracts and self.contracts[symbol].expiry_date <= self.Time.date():
# remove expired contracts
for contract in self.contracts[symbol].contracts:
self.RemoveSecurity(contract)
# remove Contracts object for current symbol
del self.contracts[symbol]
# set selection flag when there's no active contracts or every active contract expired
if len(self.contracts) == 0:
self.selection_flag = True
for _, symbol in self.symbols_by_ticker.items():
# update RollingWindow with daily prices
if symbol in data and data[symbol]:
self.prices[symbol].Add(data[symbol].Value)
# select new contracts only after expiration of the last ones
if self.selection_flag and symbol not in self.contracts:
# get all contracts for current etf
contracts = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
# get current price for etf
underlying_price = self.Securities[symbol].Price
# get strikes from commodity future contracts
strikes = [i.ID.StrikePrice for i in contracts]
# can't filter contracts, if there isn't any strike price
if len(strikes) <= 0:
continue
# filter calls and puts contracts with one month expiry
calls, puts = self.FilterContracts(strikes, contracts, underlying_price)
# make sure, there is at least one call and put contract
if len(calls) and len(puts):
# sort by expiry
call = sorted(calls, key = lambda x: x.ID.Date)[0]
put = sorted(puts, key = lambda x: x.ID.Date)[0]
# add call contract
self.AddContract(call)
# add put contract
self.AddContract(put)
# retrieve expiry date for contracts
expiry_date = call.ID.Date.date()
# store contracts with expiry date under etf symbol
self.contracts[symbol] = Contracts(expiry_date, underlying_price, [call, put])
def FilterContracts(self, strikes, contracts, underlying_price):
''' filter call and put contracts from contracts parameter '''
''' return call and put contracts '''
# get min of strike based on etf underlying price
atm_strike:float = min(strikes, key=lambda x: abs(x-underlying_price))
calls = [] # storing call contracts
puts = [] # storing put contracts
for contract in contracts:
# check if contract has one month expiry
if self.min_expiry < (contract.ID.Date - self.Time).days < self.max_expiry:
# check if contract is call
if contract.ID.OptionRight == OptionRight.Call and contract.ID.StrikePrice == atm_strike:
calls.append(contract)
# check if contract is put
elif contract.ID.OptionRight == OptionRight.Put and contract.ID.StrikePrice == atm_strike:
puts.append(contract)
# return filtered calls and puts with one month expiry
return calls, puts
def AddContract(self, contract):
''' subscribe option contract, set price mondel and normalization mode '''
option = self.AddOptionContract(contract, Resolution.Minute)
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
option.SetDataNormalizationMode(DataNormalizationMode.Raw)
def GetImpliedVolatilities(self, contracts):
''' retrieve implied volatility of contracts from contracts parameteres '''
''' returns call and put implied volatility '''
call_iv = None
put_iv = None
# go through option contracts
for c in contracts:
if c.Right == OptionRight.Call:
# found call option
call_iv = c.ImpliedVolatility
else:
# found put option
put_iv = c.ImpliedVolatility
return call_iv, put_iv
def GetHistoricalVolatility(self, rolling_window_prices):
''' calculate historical volatility based on daily prices in rolling_window_prices parameter '''
prices = np.array([x for x in rolling_window_prices])
returns = (prices[:-1] - prices[1:]) / prices[1:]
return np.std(returns)
class Contracts():
def __init__(self, expiry_date, underlying_price, contracts):
self.expiry_date = expiry_date
self.underlying_price = underlying_price
self.contracts = contracts
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaFutures(PythonData):
_last_update_date:Dict[Symbol, datetime.date] = {}
@staticmethod
def get_last_update_date() -> Dict[Symbol, datetime.date]:
return QuantpediaFutures._last_update_date
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaFutures()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split = line.split(';')
data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
data['back_adjusted'] = float(split[1])
data['spliced'] = float(split[2])
data.Value = float(split[1])
if config.Symbol not in QuantpediaFutures._last_update_date:
QuantpediaFutures._last_update_date[config.Symbol] = datetime(1,1,1).date()
if data.Time.date() > QuantpediaFutures._last_update_date[config.Symbol]:
QuantpediaFutures._last_update_date[config.Symbol] = data.Time.date()
return data
# custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))VI. Backtest Performance
