The U.S. Dollar and Variance Risk Premia Imbalances

1from AlgorithmImports import *
2
3class TheUSDollarandVarianceRiskPremiaImbalances(QCAlgorithm):
4
5def Initialize(self):
6 self.SetStartDate(2010, 1, 1)
7 self.SetCash(100000)
8 
9 self.min_expiry:int = 20
10 self.max_expiry:int = 35
11 
12 self.period:int = 22
13 self.prices:Dict[Symbol, RollingWindow] = {}
14 self.contracts:Dict[str, tuple] = {} # storing option contracts
15 self.next_expiry:datetime.datetime.date = None
16
17 self.us:str = 'SPY'  # SPDR S&P 500 ETF
18
19 self.euroarea_gdp:Dict[str, str] = {
20     'EWQ' : 'FRA_GDPD',  # iShares MSCI France Index ETF
21     'EWG' : 'DEU_GDPD',  # iShares MSCI Germany Index ETF
22     'EWI' : 'ITA_GDPD',  # iShares MSCI Italy Index ETF
23     'EWN' : 'NLD_GDPD',  # iShares MSCI Netherlands Index ETF
24 }
25
26 # subscribe GDP data
27 for ticker, gdp_symbol in self.euroarea_gdp.items():
28     self.AddData(GDPData, gdp_symbol, Resolution.Daily)
29
30 self.rest_of_the_world:List[str] = [
31     'EWA',  # iShares MSCI Australia Index ETF
32     'EWC',  # iShares MSCI Canada Index ETF
33     'EWJ',  # iShares MSCI Japan Index ETF
34     'EWL',  # iShares MSCI Switzerland Index ETF
35     'EWU',  # iShares MSCI United Kingdom Index ETF
36 ]
37
38 self.tickers:List[str] = [self.us] + list(self.euroarea_gdp.keys()) + self.rest_of_the_world
39
40 for ticker in self.tickers:
41     # subscribe to country
42     security:Equity = self.AddEquity(ticker, Resolution.Daily)
43     
44     # change normalization to raw to allow adding option contracts
45     security.SetDataNormalizationMode(DataNormalizationMode.Raw)
46
47     # set fee model and leverage
48     security.SetFeeModel(CustomFeeModel())
49     security.SetLeverage(5)
50     
51     # create RollingWindow for daily prices
52     self.prices[ticker] = RollingWindow[float](self.period)
53 
54def OnData(self, data: Slice) -> None:
55 # update RollingWindow with daily prices
56 for ticker in self.tickers:
57     if ticker in data and data[ticker]:
58         self.prices[ticker].Add(data[ticker].Value)
59
60 # check date of expiry
61 if self.next_expiry and self.Time.date() >= self.next_expiry.date():
62     self.Liquidate()
63
64     for ticker in self.tickers:
65         if ticker in self.contracts:
66             # remove expired contracts
67             self.RemoveSecurity(self.contracts[ticker][0])
68             self.RemoveSecurity(self.contracts[ticker][1])
69             
70             # remove contracts from dictionary
71             del self.contracts[ticker]
72 
73 gdp_last_update_date:Dict[str, datetime.date] = GDPData.get_last_update_date()
74
75 # set selection flag when there're no active contracts or every active contract expired
76 if len(self.contracts) == 0:
77     for ticker in self.tickers:
78         if ticker not in self.contracts:
79             if ticker in self.euroarea_gdp and not (self.Securities.ContainsKey(self.euroarea_gdp[ticker]) and self.Securities[self.euroarea_gdp[ticker]].Price != 0):
80                 # ticker is in euroarea, yet does not have GDP data subscribed
81                 continue
82
83             # get all contracts for current country
84             contracts:List[Symbol] = list(self.OptionChainProvider.GetOptionContractList(ticker, self.Time))
85             # get current price for country etf
86             underlying_price:float = self.Securities[ticker].Price
87             
88             # get strikes from country contracts
89             strikes:List[float] = [i.ID.StrikePrice for i in contracts]
90             if len(strikes) > 0:
91                 # get at the money strike
92                 atm_strike:float = min(strikes, key=lambda x: abs(x-underlying_price))
93 
94                 atm_calls:List[Symbol] = [i for i in contracts if i.ID.OptionRight == OptionRight.Call and 
95                                                          i.ID.StrikePrice == atm_strike and 
96                                                          self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
97                 
98                 atm_puts:List[Symbol] = [i for i in contracts if i.ID.OptionRight == OptionRight.Put and 
99                                                          i.ID.StrikePrice == atm_strike and 
100                                                          self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
101                 
102                 if len(atm_calls) != 0 and len(atm_puts) != 0:
103                     # sort by expiry
104                     atm_call:List[Symbol] = sorted(atm_calls, key = lambda x: x.ID.Date)[0]
105                     atm_put:List[Symbol] = sorted(atm_puts, key = lambda x: x.ID.Date)[0]
106                     
107                     self.next_expiry = min(atm_call.ID.Date, atm_put.ID.Date)
108                     
109                     # add contracts
110                     option:Option = self.AddOptionContract(atm_call, Resolution.Daily)
111                     option.PriceModel = OptionPriceModels.CrankNicolsonFD()
112                     
113                     option:Option = self.AddOptionContract(atm_put, Resolution.Daily)
114                     option.PriceModel = OptionPriceModels.CrankNicolsonFD()
115                     
116                     # store atm contracts by symbol
117                     self.contracts[ticker] = (atm_call, atm_put)
118     
119 if not self.Portfolio.Invested:
120     VP:Dict[str, float] = {} # variance risk premium
121
122     if data.OptionChains.Count != 0:
123         for kvp in data.OptionChains:
124             chain = kvp.Value
125             contracts:List = [x for x in chain]
126
127             # check if there are enough contracts for option
128             if len(contracts) < 2 or not self.prices[ticker].IsReady:
129                 continue
130             
131             atm_call_iv:float = None
132             atm_put_iv:float = None
133             # get ticker
134             ticker:str = chain.Underlying.Symbol.Value
135             
136             # go through option contracts
137             for c in contracts:
138                 if c.Right == OptionRight.Call:
139                     # found atm call
140                     atm_call_iv = c.ImpliedVolatility
141                 else:
142                     # found put option
143                     atm_put_iv = c.ImpliedVolatility
144             
145             if atm_call_iv and atm_put_iv:
146                 # risk-neutral measure of the stock market return variance
147                 risk_neutral_variance:float = ((atm_call_iv + atm_put_iv) / 2) ** 2 # calculated from average atm put and call iv
148
149                 # physical measure of the stock market return variance
150                 prices:np.ndarray = np.array(list(self.prices[ticker]))
151                 returns:np.ndarray = prices[:-1] / prices[1:] - 1
152                 physical_variance:float = (np.std(returns) * np.sqrt(252)) ** 2
153
154                 # variance risk premium calculation
155                 VP[ticker] = risk_neutral_variance - physical_variance
156                 
157         # at least 2 tickers have VP data stored and one of them is US
158         if len(VP) > 1 and self.us in VP:
159             # we construct a Euro VP as the GDP weighted average of the VPs available for the Eurozone countries
160             euroarea_vp_tickers:List[str] = [x for x in VP if x in self.euroarea_gdp]
161             rest_of_the_world_tickers:List[str] = [x for x in VP if x in self.rest_of_the_world]
162
163             # latest known GDP data is used
164             total_euroarea_gdp:float = sum([self.Securities[self.euroarea_gdp[x]].Price for x in euroarea_vp_tickers])
165             euroarea_vp:float = sum([VP[x] * (self.Securities[self.euroarea_gdp[x]].Price / total_euroarea_gdp) for x in euroarea_vp_tickers])
166
167             vp_mean:float = np.mean([vp for ticker, vp in VP.items() if ticker in rest_of_the_world_tickers] + [euroarea_vp])
168             VPI:float = VP[self.us] - vp_mean
169             
170             # the timing strategy is long (short) in the Dollar factor, and short (long) the USD, whenever VPI is positive (negative)
171             traded_tickers:List[str] = euroarea_vp_tickers + rest_of_the_world_tickers
172             traded_len:int = len(traded_tickers)
173
174             if VPI > 0:
175                 for ticker in traded_tickers:
176                     self.SetHoldings(ticker, 1 / traded_len)
177             else:
178                 for ticker in traded_tickers:
179                     self.SetHoldings(ticker, -1 / traded_len)
180
181# custom fee model
182class CustomFeeModel(FeeModel):
183def GetOrderFee(self, parameters):
184 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
185 return OrderFee(CashAmount(fee, "USD"))
186
187# source: https://data.oecd.org/gdp/gross-domestic-product-gdp.htm
188class GDPData(PythonData):
189_last_update_date:Dict[str, datetime.date] = {}
190
191@staticmethod
192def get_last_update_date() -> Dict[str, datetime.date]:
193return GDPData._last_update_date
194
195def GetSource(self, config, date, isLiveMode):
196 return SubscriptionDataSource(f'data.quantpedia.com/backtesting_data/economic/gdp/{config.Symbol.Value}.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
197
198def Reader(self, config, line, date, isLiveMode):
199 data = GDPData()
200 data.Symbol = config.Symbol
201 if not line[0].isdigit(): return None
202 split = line.split(';')
203 
204 # Parse the CSV file's columns into the custom data class
205 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + relativedelta(months=2)
206 data.Value = float(split[1])
207
208 # store last update date
209 if config.Symbol.Value not in GDPData._last_update_date:
210     GDPData._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
211
212 if data.Time.date() > GDPData._last_update_date[config.Symbol.Value]:
213     GDPData._last_update_date[config.Symbol.Value] = data.Time.date()
214 
215 return data