Active Collar Strategy

1import numpy as np
2from AlgorithmImports import *
3from dateutil.relativedelta import relativedelta
4class ActiveCollarStrategy(QCAlgorithm):
5def Initialize(self):
6 self.SetStartDate(2015, 1, 1)
7 self.SetCash(100000)
8 
9 # collar settings
10 self.targets = np.array([0.95, 1.05])   # initial target
11 self.vix_signal = 0
12 self.sma_signal_set = False
13 self.vix_signal_set = False 
14 self.macro_signal_set = False        
15 
16 option = self.AddOption("QQQ", Resolution.Minute)
17 option.SetFilter(-60, +60, timedelta(0), timedelta(35))
18 
19 # index and sma
20 data = self.AddEquity("QQQ", Resolution.Minute)
21 data.SetLeverage(10)
22 self.symbol = data.Symbol
23 
24 self.sma_5 = self.SMA(self.symbol, 5, Resolution.Daily)
25 self.sma_50 = self.SMA(self.symbol, 50, Resolution.Daily)
26 self.sma_150 = self.SMA(self.symbol, 150, Resolution.Daily)
27 self.sma_200 = self.SMA(self.symbol, 200, Resolution.Daily)
28 
29 self.index_smas = [
30     (None, self.sma_50),
31     (self.sma_5, self.sma_150),
32     (None, self.sma_200)
33     ]
34     
35 # vix and SMAs
36 self.vix = self.AddData(CBOE, 'VIX', Resolution.Daily).Symbol
37 self.vix_sma_5 = self.SMA(self.vix, 5, Resolution.Daily)
38 self.vix_std_5 = self.STD(self.vix, 5, Resolution.Daily)
39 self.vix_sma_150 = self.SMA(self.vix, 150, Resolution.Daily)
40 self.vix_std_150 = self.STD(self.vix, 150, Resolution.Daily)
41 self.vix_sma_250 = self.SMA(self.vix, 250, Resolution.Daily)
42 self.vix_std_250 = self.STD(self.vix, 250, Resolution.Daily)
43 
44 self.vix_sma_std = [
45     (self.vix_sma_5, self.vix_std_5),
46     (self.vix_sma_150, self.vix_std_150),
47     (self.vix_sma_250, self.vix_std_250)
48     ]
49 # recession indicator, claims data and SMAs
50 self.us_recession = self.AddData(FREDData, 'USREC', Resolution.Daily).Symbol     # monthly data  
51 self.initial_claims = self.AddData(FREDData, 'ICSA', Resolution.Daily).Symbol    # weekly data
52 self.claims_sma_10 = self.SMA(self.initial_claims, 10, Resolution.Daily)
53 self.claims_sma_30 = self.SMA(self.initial_claims, 30, Resolution.Daily)
54 self.claims_sma_40 = self.SMA(self.initial_claims, 40, Resolution.Daily)
55 self.claims_smas = [
56     self.claims_sma_10,
57     self.claims_sma_30,
58     self.claims_sma_40
59     ]        
60 
61 self.SetWarmUp(250, Resolution.Daily)
62 
63 # Next expiry date.
64 self.expiry_date = None
65 self.recession_signal_lagged = RollingWindow[float](2)
66 self.recession_signal_lagged.Add(-1)
67 
68 self.last_day = -1
69 
70def OnData(self, slice: Slice) -> None:
71 # Open new trades only on market close.
72 if not (self.Time.hour == 15 and self.Time.minute == 59):
73     return
74 
75 last_update_date:Dict[str, datetime.date] = FREDData.get_last_update_date()
76 # data stopped comming in
77 if not all( self.Securities[x].GetLastData() and x.Value in last_update_date and self.Time.date() <= last_update_date[x.Value] for x in [self.us_recession, self.initial_claims] ):
78     self.Liquidate()
79     return
80 # on option roll date
81 if self.expiry_date:
82     if self.Time.date() < self.expiry_date.date():
83         return
84     else:
85         # update lagged recession signal
86         if self.Securities.ContainsKey(self.us_recession):
87             recession_signal = self.Securities[self.us_recession].Price
88             self.recession_signal_lagged.Add(recession_signal)
89 
90         # SMA signal calculation - widened or tightened collar
91         for sma_pair in self.index_smas:
92             if sma_pair[1].IsReady:
93                 self.sma_signal_set = True
94                 
95                 long_sma = sma_pair[1].Current.Value
96                 
97                 if sma_pair[0] is not None:
98                     if sma_pair[0].IsReady:
99                         short_sma = sma_pair[0].Current.Value
100                         
101                         if short_sma > long_sma:
102                             self.targets += np.array([-0.01, 0.01])
103                             # sma_signal += 1
104                         else:
105                             self.targets += np.array([+0.01, -0.01])
106                             # sma_signal -= 1
107                 else:
108                     price = self.Securities[self.symbol].Price
109                     if price > long_sma:
110                         self.targets += np.array([-0.01, 0.01])
111                         # sma_signal += 1
112                     else:
113                         # sma_signal -= 1
114                         self.targets += np.array([+0.01, -0.01])
115 
116         # VIX signal calculation - quantity
117         for sma_std in self.vix_sma_std:
118             if sma_std[0].IsReady and sma_std[1].IsReady:
119                 sma = sma_std[0].Current.Value
120                 std = sma_std[1].Current.Value
121                 current_vix = self.Securities[self.vix].Price
122                 self.vix_signal_set = True
123                 if current_vix > sma + 1*std:
124                     self.vix_signal += 0.75
125                 elif current_vix < sma - 1*std:
126                     self.vix_signal += 1.25
127                 
128         # macroeconomic signal - colar shift
129         if self.Securities.ContainsKey(self.initial_claims) and self.recession_signal_lagged.IsReady:
130             recession_signal = self.recession_signal_lagged[1]
131             if recession_signal != -1:
132                 claims_value = self.Securities[self.initial_claims].Price
133                 for claims_sma in self.claims_smas:
134                     if claims_sma.IsReady:
135                         self.macro_signal_set = True
136                         if claims_value > claims_sma.Current.Value:
137                             if recession_signal == 1:
138                                 self.targets += np.array([0.01])
139                             else:
140                                 self.targets -= np.array([0.01])                
141 
142 for i in slice.OptionChains:
143     chains = i.Value
144     if not self.Portfolio.Invested:
145         calls = list(filter(lambda x: x.Right == OptionRight.Call, chains))
146         puts = list(filter(lambda x: x.Right == OptionRight.Put, chains))
147         if not calls or not puts: return
148     
149         underlying_price = self.Securities[self.symbol].Price
150         call_expiries = [i.Expiry for i in calls]
151         call_strikes = [i.Strike for i in calls]
152         # 1-month to expiration call
153         call_expiry = min(call_expiries, key=lambda x: abs((x.date() - self.Time.date()).days - 30))
154         put_expiries = [i.Expiry for i in puts]
155         put_strikes = [i.Strike for i in puts]
156         # 6-months to expiration put
157         # put_expiry = min(put_expiries, key=lambda x: abs((x.date()-self.Time.date()).days-180))
158         put_expiry = min(put_expiries, key=lambda x: abs((x.date()-self.Time.date()).days-30))  # one month expiration is used instead of 6 months
159         # determine strikes
160         put_strike = min(put_strikes, key = lambda x:abs(x - float(self.targets[0]) * underlying_price))    # changed by macro
161         call_strike = min(call_strikes, key = lambda x:abs(x - float(self.targets[1]) * underlying_price))    # changed by macro
162         
163         put = [i for i in puts if i.Expiry == put_expiry and i.Strike == put_strike]
164         call = [i for i in calls if i.Expiry == call_expiry and i.Strike == call_strike]
165         
166         if call_expiry:
167             self.expiry_date = call_expiry   # store shorter expiry date
168         
169         if put and call:
170             # All three signals were set to trade.
171             if self.sma_signal_set and self.vix_signal_set and self.macro_signal_set:
172                 options_q:int = int(self.Portfolio.TotalPortfolioValue / (underlying_price * 100))  # changed by vix
173                 if options_q >= 1:
174                     # buy index.
175                     self.SetHoldings(self.symbol, 1)
176                     
177                     # sell call
178                     self.Sell(call[0].Symbol, self.vix_signal)
179                     
180                     # buy put
181                     self.Buy(put[0].Symbol, options_q)
182                 
183                 # monthly signal reset
184                 self.vix_signal = 0
185                 self.sma_signal_set = False
186                 self.vix_signal_set = False 
187                 self.macro_signal_set = False
188                 self.targets = np.array([0.95, 1.05])
189         else:
190             pass
191 invested = [x.Key for x in self.Portfolio if x.Value.Invested]
192 if len(invested) == 1:
193     self.Liquidate(self.symbol)
194# Source: https://fred.stlouisfed.org/series/T10Y3M
195class FREDData(PythonData):
196def GetSource(self, config:SubscriptionDataConfig, date:datetime, isLiveMode:bool) -> SubscriptionDataSource:
197 return SubscriptionDataSource(f'data.quantpedia.com/backtesting_data/economic/{config.Symbol.Value}.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
198_last_update_date:Dict[str, datetime.date] = {}
199@staticmethod
200def get_last_update_date() -> Dict[str, datetime.date]:
201return FREDData._last_update_date
202def Reader(self, config:SubscriptionDataConfig, line:str, date:datetime, isLiveMode:bool) -> BaseData:
203 data = FREDData()
204 data.Symbol = config.Symbol
205 if not line[0].isdigit(): return None
206 split = line.split(';')
207 
208 # Parse the CSV file's columns into the custom data class
209 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + relativedelta(months=1)
210 if split[1] != '.':
211     data.Value = float(split[1])
212 # store last update date
213 if config.Symbol.Value not in FREDData._last_update_date:
214     FREDData._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
215 if data.Time.date() > FREDData._last_update_date[config.Symbol.Value]:
216     FREDData._last_update_date[config.Symbol.Value] = data.Time.date()
217 
218 return data