股票 · 美国 Free

主动领口策略

学术论文

作者Szado

机构

?Schneeweis, 普罗维登斯学院，马萨诸塞大学阿姆赫斯特分校 - 伊森堡管理学院

论文摘要

这项研究是对Szado和Schneeweis[2010]的更新。原始研究涵盖了1999年3月到2009年5月的时期，而这项更新的研究将分析期延长至2010年9月。信贷危机及其引发的股市下跌重新引起了人们对基于期权的股市对冲策略和保护性策略的兴趣。本文考虑了QQQ ETF以及一个小盘股股票共同基金样本的被动和主动实施对冲策略的表现。如预期所示，分析结果表明，在下跌市场中，相较于持有基础资产的长期头寸，被动对冲策略最为有效，而在上涨市场中则效果较差。本研究还考虑了对冲策略的更为主动的实施。与被动对冲策略的固定规则不同，在主动对冲调整策略中，我们应用一套规则，依据不同的经济和市场条件调整对冲策略。这种方法类似于将一套战术资产配置规则应用于一组投资。显然，有无限多的条件因素可以用于确定策略实施。在本文中，为了便于呈现，我们结合了学术文献中建议的三种条件因素（动量、波动性和复合宏观经济因素（失业率和商业周期））来生成一个动态对冲调整交易策略。在分析期内，主动对冲调整策略的表现通常优于被动对冲策略，且这一结果在样本内和样本外均为如此。至于被动和主动对冲策略的具体优势，当然取决于个人投资者的风险承受能力。

论文链接https://papers.ssrn.com/sol3/papers.cfm?abstract_id=1507991

策略概要

该策略涉及100%纳斯达克指数头寸，以QQQ ETF为例。每个月，投资者卖出1个月期看涨期权，并利用所得权利金购买6个月期看跌期权。看涨期权、看跌期权与QQQ股票的比例，以及期权的实值程度，由三大信号决定：动量、波动率和宏观经济趋势。

动量：纳斯达克-100指数的移动平均交叉（1/50，5/150，1/200 SMA组合）用于识别趋势，买入信号时扩大保护区间，卖出信号时收紧保护区间。

波动性：VIX的日度收盘水平与其移动平均（50、150、250）比较，根据波动性调整看涨期权的写入强度（如果VIX > MA的标准差1倍，则每个头寸写入0.75份看涨期权，如果VIX低于MA，则写入1.25份）。

宏观经济趋势：初请失业金人数和NBER商业周期数据帮助调整期权的实值程度。在经济扩张期间，失业金人数上升时发出看涨信号（ATM看跌期权，OTM看涨期权），而在经济收缩期间，失业金人数上升时会反向调整行权价。

这些信号结合起来将期权的实值程度设置在ATM和5% OTM之间，并根据市场状况进行月度调整。

策略合理性

跨式策略通过交换上行参与和下行保护来调整风险回报结构——该策略将标的资产的回报分布转化为具有更有利特征的新分布。通过使用系统性因素（如动量、波动性和宏观经济形势），该策略提升了风险/回报特征。

回测表现

波动率11.34%

夏普比率0.75

索提诺比率0.444

最大回撤-21.5%

胜率53%

完整 Python 代码

import numpy as np
from AlgorithmImports import *
from dateutil.relativedelta import relativedelta
class ActiveCollarStrategy(QCAlgorithm):
def Initialize(self):
 self.SetStartDate(2015, 1, 1)
 self.SetCash(100000)
 
 # collar settings
 self.targets = np.array([0.95, 1.05])   # initial target
 self.vix_signal = 0
 self.sma_signal_set = False
 self.vix_signal_set = False 
 self.macro_signal_set = False        
 
 option = self.AddOption("QQQ", Resolution.Minute)
 option.SetFilter(-60, +60, timedelta(0), timedelta(35))
 
 # index and sma
 data = self.AddEquity("QQQ", Resolution.Minute)
 data.SetLeverage(10)
 self.symbol = data.Symbol
 
 self.sma_5 = self.SMA(self.symbol, 5, Resolution.Daily)
 self.sma_50 = self.SMA(self.symbol, 50, Resolution.Daily)
 self.sma_150 = self.SMA(self.symbol, 150, Resolution.Daily)
 self.sma_200 = self.SMA(self.symbol, 200, Resolution.Daily)
 
 self.index_smas = [
     (None, self.sma_50),
     (self.sma_5, self.sma_150),
     (None, self.sma_200)
     ]
     
 # vix and SMAs
 self.vix = self.AddData(CBOE, 'VIX', Resolution.Daily).Symbol
 self.vix_sma_5 = self.SMA(self.vix, 5, Resolution.Daily)
 self.vix_std_5 = self.STD(self.vix, 5, Resolution.Daily)
 self.vix_sma_150 = self.SMA(self.vix, 150, Resolution.Daily)
 self.vix_std_150 = self.STD(self.vix, 150, Resolution.Daily)
 self.vix_sma_250 = self.SMA(self.vix, 250, Resolution.Daily)
 self.vix_std_250 = self.STD(self.vix, 250, Resolution.Daily)
 
 self.vix_sma_std = [
     (self.vix_sma_5, self.vix_std_5),
     (self.vix_sma_150, self.vix_std_150),
     (self.vix_sma_250, self.vix_std_250)
     ]
 # recession indicator, claims data and SMAs
 self.us_recession = self.AddData(FREDData, 'USREC', Resolution.Daily).Symbol     # monthly data  
 self.initial_claims = self.AddData(FREDData, 'ICSA', Resolution.Daily).Symbol    # weekly data
 self.claims_sma_10 = self.SMA(self.initial_claims, 10, Resolution.Daily)
 self.claims_sma_30 = self.SMA(self.initial_claims, 30, Resolution.Daily)
 self.claims_sma_40 = self.SMA(self.initial_claims, 40, Resolution.Daily)
 self.claims_smas = [
     self.claims_sma_10,
     self.claims_sma_30,
     self.claims_sma_40
     ]        
 
 self.SetWarmUp(250, Resolution.Daily)
 
 # Next expiry date.
 self.expiry_date = None
 self.recession_signal_lagged = RollingWindow[float](2)
 self.recession_signal_lagged.Add(-1)
 
 self.last_day = -1
 
def OnData(self, slice: Slice) -> None:
 # Open new trades only on market close.
 if not (self.Time.hour == 15 and self.Time.minute == 59):
     return
 
 last_update_date:Dict[str, datetime.date] = FREDData.get_last_update_date()
 # data stopped comming in
 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] ):
     self.Liquidate()
     return
 # on option roll date
 if self.expiry_date:
     if self.Time.date() < self.expiry_date.date():
         return
     else:
         # update lagged recession signal
         if self.Securities.ContainsKey(self.us_recession):
             recession_signal = self.Securities[self.us_recession].Price
             self.recession_signal_lagged.Add(recession_signal)
 
         # SMA signal calculation - widened or tightened collar
         for sma_pair in self.index_smas:
             if sma_pair[1].IsReady:
                 self.sma_signal_set = True
                 
                 long_sma = sma_pair[1].Current.Value
                 
                 if sma_pair[0] is not None:
                     if sma_pair[0].IsReady:
                         short_sma = sma_pair[0].Current.Value
                         
                         if short_sma > long_sma:
                             self.targets += np.array([-0.01, 0.01])
                             # sma_signal += 1
                         else:
                             self.targets += np.array([+0.01, -0.01])
                             # sma_signal -= 1
                 else:
                     price = self.Securities[self.symbol].Price
                     if price > long_sma:
                         self.targets += np.array([-0.01, 0.01])
                         # sma_signal += 1
                     else:
                         # sma_signal -= 1
                         self.targets += np.array([+0.01, -0.01])
 
         # VIX signal calculation - quantity
         for sma_std in self.vix_sma_std:
             if sma_std[0].IsReady and sma_std[1].IsReady:
                 sma = sma_std[0].Current.Value
                 std = sma_std[1].Current.Value
                 current_vix = self.Securities[self.vix].Price
                 self.vix_signal_set = True
                 if current_vix > sma + 1*std:
                     self.vix_signal += 0.75
                 elif current_vix < sma - 1*std:
                     self.vix_signal += 1.25
                 
         # macroeconomic signal - colar shift
         if self.Securities.ContainsKey(self.initial_claims) and self.recession_signal_lagged.IsReady:
             recession_signal = self.recession_signal_lagged[1]
             if recession_signal != -1:
                 claims_value = self.Securities[self.initial_claims].Price
                 for claims_sma in self.claims_smas:
                     if claims_sma.IsReady:
                         self.macro_signal_set = True
                         if claims_value > claims_sma.Current.Value:
                             if recession_signal == 1:
                                 self.targets += np.array([0.01])
                             else:
                                 self.targets -= np.array([0.01])                
 
 for i in slice.OptionChains:
     chains = i.Value
     if not self.Portfolio.Invested:
         calls = list(filter(lambda x: x.Right == OptionRight.Call, chains))
         puts = list(filter(lambda x: x.Right == OptionRight.Put, chains))
         if not calls or not puts: return
     
         underlying_price = self.Securities[self.symbol].Price
         call_expiries = [i.Expiry for i in calls]
         call_strikes = [i.Strike for i in calls]
         # 1-month to expiration call
         call_expiry = min(call_expiries, key=lambda x: abs((x.date() - self.Time.date()).days - 30))
         put_expiries = [i.Expiry for i in puts]
         put_strikes = [i.Strike for i in puts]
         # 6-months to expiration put
         # put_expiry = min(put_expiries, key=lambda x: abs((x.date()-self.Time.date()).days-180))
         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
         # determine strikes
         put_strike = min(put_strikes, key = lambda x:abs(x - float(self.targets[0]) * underlying_price))    # changed by macro
         call_strike = min(call_strikes, key = lambda x:abs(x - float(self.targets[1]) * underlying_price))    # changed by macro
         
         put = [i for i in puts if i.Expiry == put_expiry and i.Strike == put_strike]
         call = [i for i in calls if i.Expiry == call_expiry and i.Strike == call_strike]
         
         if call_expiry:
             self.expiry_date = call_expiry   # store shorter expiry date
         
         if put and call:
             # All three signals were set to trade.
             if self.sma_signal_set and self.vix_signal_set and self.macro_signal_set:
                 options_q:int = int(self.Portfolio.TotalPortfolioValue / (underlying_price * 100))  # changed by vix
                 if options_q >= 1:
                     # buy index.
                     self.SetHoldings(self.symbol, 1)
                     
                     # sell call
                     self.Sell(call[0].Symbol, self.vix_signal)
                     
                     # buy put
                     self.Buy(put[0].Symbol, options_q)
                 
                 # monthly signal reset
                 self.vix_signal = 0
                 self.sma_signal_set = False
                 self.vix_signal_set = False 
                 self.macro_signal_set = False
                 self.targets = np.array([0.95, 1.05])
         else:
             pass
 invested = [x.Key for x in self.Portfolio if x.Value.Invested]
 if len(invested) == 1:
     self.Liquidate(self.symbol)
# Source: https://fred.stlouisfed.org/series/T10Y3M
class FREDData(PythonData):
def GetSource(self, config:SubscriptionDataConfig, date:datetime, isLiveMode:bool) -> SubscriptionDataSource:
 return SubscriptionDataSource(f'data.quantpedia.com/backtesting_data/economic/{config.Symbol.Value}.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
_last_update_date:Dict[str, datetime.date] = {}
@staticmethod
def get_last_update_date() -> Dict[str, datetime.date]:
return FREDData._last_update_date
def Reader(self, config:SubscriptionDataConfig, line:str, date:datetime, isLiveMode:bool) -> BaseData:
 data = FREDData()
 data.Symbol = config.Symbol
 if not line[0].isdigit(): return None
 split = line.split(';')
 
 # Parse the CSV file's columns into the custom data class
 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + relativedelta(months=1)
 if split[1] != '.':
     data.Value = float(split[1])
 # store last update date
 if config.Symbol.Value not in FREDData._last_update_date:
     FREDData._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
 if data.Time.date() > FREDData._last_update_date[config.Symbol.Value]:
     FREDData._last_update_date[config.Symbol.Value] = data.Time.date()
 
 return data