“该策略涉及对价内期权中最高十分位数做多,最低十分位数做空,并进行Delta对冲,持有至到期。投资组合等权重。”

I. 策略概要

该投资范围包括OptionMetrics Ivy DB数据库中列出的期权,到期日为下个月。对于每只股票,选择最接近价内的看涨期权,考虑价内程度在0.7到1.3之间,排除未平仓合约或交易量为零的期权。然后根据标的股票价格将期权分为十分位数。该策略涉及对最高十分位数做多,对最低十分位数做空,采用等权重投资组合,进行Delta对冲,并持有至期权到期。标的股票数据来源于CRSP/Compustat。

II. 策略合理性

这种异常现象可归因于散户投资者非理性地推高低价股票期权的价格,认为它们便宜且损失最小。这种效应在机构持股比例较低的股票中更为明显,因为这些股票受专业交易员的关注较少。相比之下,专业交易员在股价上涨时更倾向于购买期权。研究还表明,散户投资者购买期权与股价之间存在负相关关系,这进一步证实了散户投资者的行为通过高估这些低价股票期权来推动这种异常现象。

III. 来源论文

Cheap Options Are Expensive [点击查看论文]

<摘要>

我们发现,对标的股票价格(部分)不关注会导致对低价股票期权的需求压力,从而导致此类期权定价过高。从经验上看,我们发现,对低价股票进行Delta对冲的期权表现比对高价股票进行Delta对冲的期权每周低0.63%(看涨期权)和0.36%(看跌期权)。自然实验证实了这一发现;股票拆分后,期权往往变得相对更贵,迷你指数期权相对于其他相同的常规指数期权定价过高。偏度偏好并不能解释我们的结果。

IV. 回测表现

年化回报32.32%
波动率13.32%
β值-0.023
夏普比率2.43
索提诺比率-1.005
最大回撤N/A
胜率28%

V. 完整的 Python 代码

from AlgorithmImports import *
from typing import List, Dict, Tuple
#endregion
class CheapOptionsAreExpensive(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2016, 1, 1)
        self.SetCash(1000000)
        
        self.min_expiry: int = 35
        self.max_expiry: int = 60
        self.leverage: int = 5
        self.min_share_price: int = 5
        self.subscribed_contracts_treshold: int = 10
        self.quantile: int = 10
        
        self.subscribed_contracts: Dict[str, Symbol] = {}
        self.tickers_symbols: Dict[str, Symbol] = {}
        
        self.day: int = -1
        self.fundamental_count: int = 100
        self.fundamental_sorting_key = lambda x: x.DollarVolume
        self.opened_position_with_expiry: List[Tuple[Symbol, Symbol, datetime.date]] = []   # stock symbol, contract symbol, option expiry
        self.settings.daily_precise_end_time = False
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.UniverseSettings.Resolution = Resolution.Minute
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
        self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
            security.SetLeverage(self.leverage)
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        # rebalance on contract expirations
        if len(self.tickers_symbols) != 0:
            return Universe.Unchanged
        
        # select top n stocks by dollar volume with price higher than 5
        selected: List[Fundamental] = [
            x for x in fundamental 
            if x.HasFundamentalData 
            and x.Market == 'usa' 
            and x.Price > self.min_share_price
        ]
        if len(selected) > self.fundamental_count:
            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
        self.tickers_symbols = {x.Symbol.Value: x.Symbol for x in selected}
        
        # return symbols of selected stocks
        return [x.Symbol for x in selected]
        
    def OnData(self, data: Slice) -> None:
        # on next DataBar(1 minute after subscription) after contracts subscription trade selected contracts and their stocks
        if len(self.subscribed_contracts) >= self.subscribed_contracts_treshold:
            stock_prices: Dict[str, float] = {} # storing stock prices keyed by atm call contracts
            
            for ticker, contract_symbol in self.subscribed_contracts.items():
                if ticker in self.tickers_symbols:
                    stock_symbol: Symbol = self.tickers_symbols[ticker]
                    # make sure, there are DataBars for stock and atm contract
                    if stock_symbol in data and data[stock_symbol] and contract_symbol in data and data[contract_symbol]:
                        underlying_price: float = data[stock_symbol].Value
                        # store stock underlying price keyed by ticker
                        stock_prices[ticker] = underlying_price
            traded: bool = False
            # make sure, there are enough data for quantile selection
            if len(stock_prices) >= self.quantile:
                # sorting by underlying stock price
                quantile: int = int(len(stock_prices) / self.quantile)
                sorted_by_price = sorted(stock_prices.items(), key = lambda x: x[1], reverse=True)
                long: List[Symbol] = sorted_by_price[:quantile]
                short: List[Symbol] = sorted_by_price[-quantile:]
                
                long_w: float = 1 / len(long)
                short_w: float = 1 / len(short)
                
                for ticker, price in long:
                    # retrieve atm contract symbol based on ticker
                    contract_symbol: Symbol = self.subscribed_contracts[ticker]
                    # retrieve stock symbol based on ticker
                    stock_symbol: Symbol = self.tickers_symbols[ticker]
                    
                    equity: float = self.Portfolio.TotalPortfolioValue * long_w
                    options_q: int = int(equity / (price * 100))
                    # buy contract
                    self.Securities[contract_symbol].MarginModel = BuyingPowerModel(2)
                    if contract_symbol in data and data[contract_symbol] and stock_symbol in data and data[stock_symbol]:
                        self.Buy(contract_symbol, options_q)
                        self.Sell(stock_symbol,options_q * 50)  # initial delta hedge
                    
                        self.opened_position_with_expiry.append((stock_symbol, contract_symbol, contract_symbol.ID.Date.date()))
                        traded = True
    
                for ticker, price in short:
                    # retrieve atm contract symbol based on ticker
                    contract_symbol = self.subscribed_contracts[ticker]
                    # retrieve stock symbol based on ticker
                    stock_symbol = self.tickers_symbols[ticker]
                    
                    equity = self.Portfolio.TotalPortfolioValue * short_w
                    options_q = int(equity / (price * 100))
                    # sell contract
                    self.Securities[contract_symbol].MarginModel = BuyingPowerModel(2)
                    if contract_symbol in data and data[contract_symbol] and stock_symbol in data and data[stock_symbol]:
                        self.Sell(contract_symbol, options_q)
                        self.Buy(stock_symbol, options_q * 50)  # initial delta hedge
                        
                        self.opened_position_with_expiry.append((stock_symbol, contract_symbol, contract_symbol.ID.Date.date()))
                        traded = True
            
            if traded:
                # clear dictionary and wait for next selection and contracts subscription    
                self.subscribed_contracts.clear()
        
        # check if contracts expiries once in a day
        if self.day == self.Time.day:
            return
        self.day = self.Time.day
        
        # positions are opened
        if len(self.opened_position_with_expiry) != 0:
            positions_to_remove: List[Tuple[Symbol, Symbol, datetime.date]] = []
            for opened_position_with_expiry in self.opened_position_with_expiry:
                stock_symbol: Symbol = opened_position_with_expiry[0]
                contract_symbol: Symbol = opened_position_with_expiry[1]
                exp: datetime.date = opened_position_with_expiry[2]
                if exp <= self.Time.date():
                    self.Liquidate(contract_symbol)     # liquidate contract
                    self.Liquidate(stock_symbol)        # liquidate hedge
                    positions_to_remove.append(opened_position_with_expiry)
            
            for pos_to_remove in positions_to_remove:
                self.opened_position_with_expiry.remove(pos_to_remove)
                
            if len(self.opened_position_with_expiry) == 0:
                # perform next selection of stock
                self.tickers_symbols.clear()
                self.subscribed_contracts.clear()
                return
        
        # subscribe to new contracts, when last one expiries
        if not self.Portfolio.Invested:
            for _, symbol in self.tickers_symbols.items():
                if self.Securities[symbol].IsDelisted:
                    continue
                # subscribe to contract
                contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
                # get current price for stock
                underlying_price: float = self.Securities[symbol].Price
                
                # get strikes from stock contracts
                strikes: List[float] = [i.ID.StrikePrice for i in contracts]
                
                # check if there is at least one strike    
                if len(strikes) <= 0:
                    continue
            
                # at the money
                atm_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
                
                # filtred contracts based on option rights and strikes
                atm_calls: List[Symbol] = [i for i in contracts if i.ID.OptionRight == OptionRight.Call and 
                                                     i.ID.StrikePrice == atm_strike and 
                                                     self.min_expiry <= (i.ID.Date - self.Time).days <= self.max_expiry]
                
                # make sure there are enough contracts
                if len(atm_calls) > 0:
                    # sort by expiry
                    atm_call: Symbol = sorted(atm_calls, key = lambda item: item.ID.Date, reverse=True)[0]
                    
                    # add contract
                    option: Option = self.AddOptionContract(atm_call, Resolution.Minute)
                    option.SetDataNormalizationMode(DataNormalizationMode.Raw)
                    
                    # store subscribed atm call contract keyed by it's ticker
                    self.subscribed_contracts[atm_call.Underlying.Value] = atm_call
        else:
            if len(self.opened_position_with_expiry) == 0:
                self.Liquidate()
                
# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
        fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))

发表评论

了解 Quant Buffet 的更多信息

立即订阅以继续阅读并访问完整档案。

继续阅读