横截面六个月减去一个月股票ATM跨式日历交易策略

1from AlgorithmImports import *
2from typing import List, Dict
3from pandas.core.frame import DataFrame
4from pandas.core.series import Series
5#endregion
6class CrossSectionalSixMinusOneMonthEquityATMStraddleCalendarTradingStrategy(QCAlgorithm):
7def Initialize(self) -> None:
8self.SetStartDate(2012, 1, 1)
9self.SetCash(1000000)
10
11self.long_term_min_expiry: int = 150
12self.long_term_max_expiry: int = 230
13
14self.short_term_min_expiry: int = 20
15self.short_term_max_expiry: int = 45
16
17self.long_term_period: int = 6 * 21 # need n of stock daily prices
18self.short_term_period: int = 21 # need n of stock daily prices
19self.percentage_traded: float = 0.2
20self.min_contracts: int = 4
21self.leverage: int = 10
22self.min_share_price: int = 5
23self.quantile: int = 5
24
25self.data: Dict[Symbol, RollingWindow] = {}
26self.symbols_by_ticker: Dict[str, Symbol] = {}
27self.subscribed_contracts: Dict[Symbol, Contracts] = {}
28
29symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
30
31self.day: int = -1
32self.fundamental_count: int = 50
33self.fundamental_sorting_key = lambda x: x.DollarVolume
34self.selection_flag: bool = False
35self.settings.daily_precise_end_time = False
36self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
37self.UniverseSettings.Resolution = Resolution.Daily
38self.AddUniverse(self.FundamentalSelectionFunction)
39self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
40self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
41self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.BeforeMarketClose(symbol, 0), self.Selection)
42def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
43for security in changes.AddedSecurities:
44    security.SetFeeModel(CustomFeeModel())
45    security.SetLeverage(self.leverage)
46def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
47# update daily prices of stocks in self.data dictionary
48for stock in fundamental:
49    symbol: Symbol = stock.Symbol
50    
51    if symbol in self.data:
52        self.data[symbol].Add(stock.AdjustedPrice)
53
54# rebalance monthly
55if not self.selection_flag:
56    return Universe.Unchanged
57
58# select top n stocks by dollar volume
59selected: List[Fundamental] = [
60    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price
61]
62if len(selected) > self.fundamental_count:
63    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
64for stock in selected:
65    symbol: Symbol = stock.Symbol
66    ticker: str = symbol.Value 
67    self.symbols_by_ticker[ticker] = symbol
68    
69    if symbol in self.data:
70        continue
71    
72    self.data[symbol] = RollingWindow[float](self.long_term_period)
73    history: DataFrame = self.History(symbol, self.long_term_period, Resolution.Daily)
74    if history.empty:
75        continue
76    closes: Series = history.loc[symbol].close
77    for time, close in closes.items():
78        self.data[symbol].Add(close)
79# return newly selected symbols
80return list(map(lambda x: x.Symbol, selected))
81def OnData(self, data: Slice) -> None:
82# execute once a day
83if self.day == self.Time.day:
84    return
85self.day = self.Time.day
86
87# subscribe to new contracts after selection
88if len(self.subscribed_contracts) == 0 and self.selection_flag:
89    for _, symbol in self.symbols_by_ticker.items():
90        if symbol in self.data and self.data[symbol].IsReady:
91            # get all contracts for current stock symbol
92            contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
93            # get current price for etf
94            underlying_price: float = self.data[symbol][0]
95            
96            # get strikes from commodity future contracts
97            strikes: List[float] = [i.ID.StrikePrice for i in contracts]
98            
99            # can't filter contracts, if there isn't any strike price
100            if len(strikes) <= 0 or underlying_price == 0:
101                continue
102            
103            # filter calls and puts contracts with six months expiry
104            long_term_calls, long_term_puts = self.FilterContracts(
105                strikes,
106                contracts,
107                underlying_price,
108                self.long_term_min_expiry,
109                self.long_term_max_expiry
110            )
111            
112            # filter calls and puts contracts with one month expiry
113            short_term_calls, short_term_puts = self.FilterContracts(
114                strikes,
115                contracts,
116                underlying_price,
117                self.short_term_min_expiry,
118                self.short_term_max_expiry
119            )
120            
121            # make sure, there is at least one call and put contract
122            if len(long_term_calls) > 0 and len(long_term_puts) > 0 and len(short_term_calls) > 0 and len(short_term_puts) > 0:
123                # sort by expiry
124                long_term_call: Symbol = sorted(long_term_calls, key = lambda x: x.ID.Date, reverse=True)[0]
125                long_term_put: Symbol = sorted(long_term_puts, key = lambda x: x.ID.Date, reverse=True)[0]
126                
127                short_term_call: Symbol = sorted(short_term_calls, key = lambda x: x.ID.Date, reverse=True)[0]
128                short_term_put: Symbol = sorted(short_term_puts, key = lambda x: x.ID.Date, reverse=True)[0]
129                
130                subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(short_term_call.Underlying)
131                if subscriptions:
132                    # add contracts
133                    for contract in [long_term_call, long_term_put, short_term_call, short_term_put]:
134                        
135                        self.AddContract(contract)
136                    
137                    # retrieve expiry date for contracts
138                    long_term_expiry_date: datetime.date = long_term_call.ID.Date.date()
139                    
140                    short_term_expiry_date: datetime.date = short_term_call.ID.Date.date()
141                    
142                    # store contracts with expiry date under stock's symbol
143                    self.subscribed_contracts[symbol] = Contracts(
144                        long_term_expiry_date,
145                        short_term_expiry_date, 
146                        [long_term_call, long_term_put],
147                        [short_term_call, short_term_put],
148                        underlying_price
149                    )
150    
151# calculate term structure and trade options
152elif len(self.subscribed_contracts) != 0 and data.OptionChains.Count != 0 and self.selection_flag:
153    self.selection_flag = False # this makes sure, there will be no other trades until next selection
154    
155    term_structure_long_term: Dict[Symbol, float] = {} # storing term structures keyed by stock's symbol
156    term_structure_short_term: Dict[Symbol, float] = {}
157    
158    for kvp in data.OptionChains:
159        chain: OptionChain = kvp.Value
160        ticker: str = chain.Underlying.Symbol.Value
161        if ticker in self.symbols_by_ticker:
162            # get stock's symbol
163            symbol: Symbol = self.symbols_by_ticker[ticker]
164            if symbol in self.subscribed_contracts:
165                # get contracts
166                contracts: List[Symbol] = [x for x in chain]
167                
168                # check if there are enough contracts for option and daily prices are ready
169                if len(contracts) < self.min_contracts or not self.data[symbol].IsReady:
170                    continue
171                
172                # expiry dates are needed for finding out which contract is long term and which one is short term when retrieving IV
173                long_term_expiry_date: datetime.date = self.subscribed_contracts[symbol].long_term_expiry_date
174                short_term_expiry_date: datetime.date = self.subscribed_contracts[symbol].short_term_expiry_date
175                
176                # get call and put implied volatility
177                long_term_call_iv, long_term_put_iv, short_term_call_iv, short_term_put_iv = self.GetImpliedVolatilities(
178                    contracts,
179                    long_term_expiry_date,
180                    short_term_expiry_date
181                )
182                
183                if long_term_call_iv and long_term_put_iv and short_term_call_iv and short_term_put_iv:
184                    # make mean from call implied volatility and put implied volatility
185                    long_term_iv: float = (long_term_call_iv + long_term_put_iv) / 2 
186                    short_term_iv: float = (short_term_call_iv + short_term_put_iv) / 2
187                    
188                    # get historical volatility for long term
189                    long_term_hv: float = self.GetHistoricalVolatility(self.data[symbol], self.long_term_period)
190                    short_term_hv: float = self.GetHistoricalVolatility(self.data[symbol], self.short_term_period)
191                    
192                    # store stock's term structure
193                    term_structure_long_term[symbol] = (long_term_iv - long_term_hv) / long_term_hv
194                    term_structure_short_term[symbol] = (short_term_iv - short_term_hv) / short_term_hv
195        
196    # can't perform selection
197    if len(term_structure_long_term) < self.quantile or len(term_structure_short_term) < self.quantile:
198        return
199    
200    # perform quintile selection
201    quantile: int = int(len(term_structure_long_term) / self.quantile)
202    sorted_by_ts_long_term: List[Symbol] = [x[0] for x in sorted(term_structure_long_term.items(), key=lambda item: item[1])]
203    sorted_by_ts_short_term: List[Symbol] = [x[0] for x in sorted(term_structure_short_term.items(), key=lambda item: item[1])]
204    
205    # the strategy sells quintile 5 of 1 month ATM straddle and buys quintile 5 of 6 month ATM straddles.
206    long: List[Symbol] = sorted_by_ts_long_term[:quantile]
207    short: List[Symbol] = sorted_by_ts_short_term[:quantile]
208    
209    # trade execution
210    self.Liquidate()
211    
212    # trade long
213    self.TradeOptions(data, long, True, True) # parameters: symbols, long_flag, long_term_flag
214    # trade short
215    self.TradeOptions(data, short, False, False) # parameters: symbols, long_flag, long_term_flag
216    
217def Selection(self) -> None:
218self.selection_flag = True # perform new selection
219self.Liquidate() # rebalance monthly, so liquidate all holding contracts
220
221# remove contracts from securities
222# for _, contracts_obj in self.subscribed_contracts.items():
223#     for contract in contracts_obj.long_term_contracts + contracts_obj.short_term_contracts:
224#         self.RemoveSecurity(contract)
225        
226# clear dictionary for subscribed contracts, because there will be new selection
227self.subscribed_contracts.clear()
228# clear dictionary of tickers and their symbols, because new stocks will be selected
229self.symbols_by_ticker.clear()
230
231def FilterContracts(self, 
232                strikes: List[float],
233                contracts: List[Symbol], 
234                underlying_price: float, 
235                min_expiry: datetime.date, 
236                max_expiry: datetime.date) -> List[Symbol]:
237''' filter call and put contracts from contracts parameter '''
238''' return call and put contracts '''
239
240# Straddle
241call_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
242put_strike: float = call_strike
243
244calls: List[Symbol] = [] # storing call contracts
245puts: List[Sybol] = [] # storing put contracts
246
247for contract in contracts:
248    # check if contract has one month expiry
249    if min_expiry < (contract.ID.Date - self.Time).days < max_expiry:
250        # check if contract is call
251        if contract.ID.OptionRight == OptionRight.Call and contract.ID.StrikePrice == call_strike:
252            calls.append(contract)
253        # check if contract is put
254        elif contract.ID.OptionRight == OptionRight.Put and contract.ID.StrikePrice == put_strike:
255            puts.append(contract)
256
257# return filtered calls and puts with one month expiry
258return calls, puts
259
260def AddContract(self, contract: Symbol) -> None:
261''' subscribe option contract, set price mondel and normalization mode '''
262option: Option = self.AddOptionContract(contract, Resolution.Daily)
263option.PriceModel = OptionPriceModels.CrankNicolsonFD()
264option.SetDataNormalizationMode(DataNormalizationMode.Raw)
265
266def GetImpliedVolatilities(self, 
267                    contracts: List[Symbol], 
268                    long_term_expiry: datetime.date, 
269                    short_term_expiry: datetime.date) -> float:
270''' retrieve implied volatility of contracts from contracts parameteres '''
271''' returns long term and short term implied volatility for call and put contracts '''
272long_term_call_iv: Union[None, float] = None
273long_term_put_iv: Union[None, float] = None
274
275short_term_call_iv: Union[None, float] = None
276short_term_put_iv: Union[None, float] = None
277        
278# go through option contracts
279for c in contracts:
280    iv: float = c.ImpliedVolatility
281    expiry_date: datetime.date = c.get_Expiry().date()
282    
283    if c.Right == OptionRight.Call:
284        if expiry_date == long_term_expiry:
285            long_term_call_iv = iv
286        elif expiry_date == short_term_expiry:
287            short_term_call_iv = iv
288    else:
289        if expiry_date == long_term_expiry:
290            long_term_put_iv = iv
291        elif expiry_date == short_term_expiry:
292            short_term_put_iv = iv
293    
294return long_term_call_iv, long_term_put_iv, short_term_call_iv, short_term_put_iv
295
296def GetHistoricalVolatility(self, 
297                        rolling_window_prices: RollingWindow, 
298                        period: int) -> float:
299''' calculate historical volatility based on daily prices in rolling_window_prices parameter '''
300prices: np.ndarray = np.array([x for x in rolling_window_prices][:period])
301returns: np.ndarray = (prices[:-1] - prices[1:]) / prices[1:]
302return np.std(returns)
303
304def TradeOptions(self, 
305            data: Slice, 
306            symbols: List[Symbol], 
307            long_flag: bool, 
308            long_term_flag: bool) -> None:
309''' on long signal buy call and put option contract '''
310''' on short signal sell call and put option contract '''
311length: int = len(symbols)
312
313# trade etf's call and put contracts
314for symbol in symbols:
315    if long_term_flag:
316        contracts: List[Symbol] = self.subscribed_contracts[symbol].long_term_contracts
317    else:
318        contracts: List[Symbol] = self.subscribed_contracts[symbol].short_term_contracts
319        
320    # check if contracts are tradebale and don't have 0 price
321    for contract in contracts:
322        if not self.Securities[contract].IsTradable or self.Securities[contract].Price == 0:
323            return
324        
325    # get call and put contract
326    call, put = contracts
327    # get underlying price
328    underlying_price: float = self.subscribed_contracts[symbol].underlying_price
329    
330    options_q: int = int(((self.Portfolio.TotalPortfolioValue * self.percentage_traded) / length) / (underlying_price * 100))
331    
332    self.Securities[put].MarginModel = BuyingPowerModel(2)
333    self.Securities[call].MarginModel = BuyingPowerModel(2)
334    
335    if call in data and data[call] and put in data and data[put]:
336        if long_flag:
337            self.Buy(call, options_q)
338            self.Buy(put, options_q)
339        else:
340            self.Sell(call, options_q)
341            self.Sell(put, options_q)
342
343class Contracts():
344def __init__(self, 
345        long_term_expiry_date: datetime.date, 
346        short_term_expiry_date: datetime.date, 
347        long_term_contracts: List[Symbol], 
348        short_term_contracts: List[Symbol], 
349        underlying_price: float) -> None:
350self.long_term_expiry_date: datetime.date = long_term_expiry_date
351self.long_term_contracts: List[Symbol] = long_term_contracts
352
353self.short_term_expiry_date: datetime.date = short_term_expiry_date
354self.short_term_contracts: List[Symbol] = short_term_contracts
355
356self.underlying_price = underlying_price
357# Custom fee model
358class CustomFeeModel(FeeModel):
359def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
360fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
361return OrderFee(CashAmount(fee, "USD"))