“该策略涉及按隔夜回报对股票进行排序，做多顶部十分位数（赢家），做空底部十分位数（输家）。持仓隔夜，每月重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 日内 | 市场: 股票 | 关键词: 隔夜、动量
I. 策略概要
投资范围包括来自纽约证券交易所、美国证券交易所和纳斯达克的股票，以及来自CRSP和TAQ数据库的价格和成交量数据。投资者使用指定的公式计算隔夜回报，并根据上个月的隔夜回报将股票分为十分位数。在每个月末，投资者做多顶部十分位数（赢家股票），做空底部十分位数（输家股票）。持仓仅隔夜，在收盘时建仓，在开盘时平仓。投资组合中的股票按价值加权，并每月重新平衡。
II. 策略合理性
股票中的动量效应源于投资者的非理性和对新闻的反应不足。作者认为，由于机构交易主要在日内进行，并且经常逆势交易，因此隔夜动量更强。
III. 来源论文
A Tug of War: Overnight Versus Intraday Expected Returns [点击查看论文]
- 董楼、克里斯托弗·波尔克、斯皮罗斯·斯科拉斯。英国伦敦经济学院金融系，伦敦 WC2A 2AE，英国 及 CEP。英国伦敦经济学院金融系，伦敦 WC2A 2AE，英国 及 CEPR。雅典经济与商业大学。
<摘要>
我们发现，动量利润完全在隔夜产生，而所研究的所有其他交易策略的利润则完全在日内产生。实际上，对于四因子异常现象，日内回报特别高，因为存在符号相反的隔夜溢价，部分抵消了日内回报。我们将动量预期回报分解中的横截面和时间序列变化与机构动量交易的变化联系起来，从而产生大约每月2%的隔夜减日内动量回报变化。对九个非美国市场动量回报的隔夜/日内分解与我们的美国研究结果一致。最后，我们记录了强劲且持续的隔夜动量、日内动量和跨期反转效应。
IV. 回测表现
| 年化回报 | 50.58% |
| 波动率 | 11.24% |
| β值 | -0.629 |
| 夏普比率 | 4.04 |
| 索提诺比率 | -0.2 |
| 最大回撤 | N/A |
| 胜率 | 48% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
from typing import List, Dict
from pandas.core.frame import dataframe
#endregion
class OvernightMomentumStrategy(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2015, 1, 1)
self.SetCash(100_000)
self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
self.period: int = 21 # need n of ovenight returns
market: Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
self.data: Dict[Symbol, SymbolData] = {} # storing objects of SymbolData under stocks symbols
self.quantile: int = 10
self.leverage: int = 20
self.min_share_price: int = 5
self.traded_quantity: Dict[Symbol, float] = {}
self.fundamental_count: int = 100
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag: bool = False
self.UniverseSettings.Leverage = self.leverage
self.UniverseSettings.Resolution = Resolution.Minute
self.AddUniverse(self.FundamentalSelectionFunction)
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market, 1), self.Selection)
self.Schedule.On(self.DateRules.EveryDay(market), self.TimeRules.BeforeMarketClose(market, 20), self.MarketClose)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# update overnight prices on daily basis
for stock in fundamental:
symbol: Symbol = stock.Symbol
if symbol in self.data:
# store current stock price
self.data[symbol].current_price = stock.AdjustedPrice
# get history prices
history: dataframe = self.History(symbol, 1, Resolution.Daily)
# update overnight returns based on history prices
self.UpdateOvernightReturns(symbol, history)
# monthly rebalance
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
selected: List[Fundamental] = [
x for x in fundamental
if x.HasFundamentalData
and x.Market == 'usa'
and x.MarketCap != 0
and x.Price > self.min_share_price
and x.SecurityReference.ExchangeId in self.exchange_codes
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
# warm up overnight returns
for stock in selected:
symbol: Symbol = stock.Symbol
if symbol in self.data and self.data[symbol].is_overnight_returns_ready():
# get overnight returns from RollingWindow object and reverse it's list for simplier calculation of returns accumulation
overnight_returns: List[float] = [x for x in self.data[symbol].overnight_returns]
overnight_returns.reverse()
# calculate accumulated returns
accumulated_returns = np.prod([(1 + x) for x in overnight_returns]) - 1
# update returns accumulated for last month
self.data[symbol].returns_accumulated_last_month = accumulated_returns
# go to next iteration, because there is no need for warm up overnight returns
continue
# initialize SymbolData object for current symbol
self.data[symbol] = SymbolData(self.period)
# get history of n + 1 days
history: dataframe = self.History(symbol, self.period + 1, Resolution.Daily)
# update overnight returns based on history prices
self.UpdateOvernightReturns(symbol, history)
market_cap: Dict[Symbol, float] = {} # storing stocks market capitalization
last_accumulated_returns: Dict[Symbol, float] = {} # storing stocks last accumuldated returns
for stock in selected:
symbol = stock.Symbol
if not self.data[symbol].is_ready():
continue
# store stock's market capitalization
market_cap[symbol] = stock.MarketCap
# store stock's last accumulated returns
last_accumulated_returns[symbol] = self.data[symbol].returns_accumulated_last_month
# not enough data for decile selection
if len(last_accumulated_returns) < self.quantile:
return Universe.Unchanged
# overnight returns sorting
quantile: int = int(len(last_accumulated_returns) / self.quantile)
sorted_by_last_acc_ret: List[Symbol] = [x[0] for x in sorted(last_accumulated_returns.items(), key=lambda item: item[1])]
# long winners
long: List[Symbol] = sorted_by_last_acc_ret[-quantile:]
# short losers
short: List[Symbol] = sorted_by_last_acc_ret[:quantile]
# market cap weighting
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda x: market_cap[x], portfolio)))
for symbol in portfolio:
if self.data[symbol].current_price != 0:
current_price: float = self.data[symbol].current_price
w: float = market_cap[symbol] / mc_sum
quantity: int = ((-1)**i) * np.floor((self.Portfolio.TotalPortfolioValue * w) / current_price)
self.traded_quantity[symbol] = quantity
return list(self.traded_quantity.keys())
def MarketClose(self) -> None:
# send market on open and on close orders before market closes
for symbol, q in self.traded_quantity.items():
self.MarketOnCloseOrder(symbol, q)
self.MarketOnOpenOrder(symbol, -q)
def UpdateOvernightReturns(self, symbol: Symbol, history: dataframe) -> None:
# calculate overnight returns only if history isn't empty
if history.empty:
return
# get open and close prices
opens = history.loc[symbol].open
closes = history.loc[symbol].close
# calculate overnight return for each day
for (_, close_price), (_, open_price) in zip(closes.items(), opens.items()):
# check if previous close price isn't None
if self.data[symbol].prev_close_price:
# calculate overnight return
overnight_return = (open_price / self.data[symbol].prev_close_price) - 1
# store overnight return
self.data[symbol].update(overnight_return)
# change value of prev close price for next calculation
self.data[symbol].prev_close_price = close_price
def Selection(self) -> None:
self.selection_flag = True
self.traded_quantity.clear()
class SymbolData():
def __init__(self, period: int) -> None:
self.overnight_returns: RollingWindow = RollingWindow[float](period)
self.returns_accumulated_last_month: Union[None, float] = None
self.prev_close_price: Union[None, float] = None
self.current_price: float = 0.
def update(self, overnight_return: float) -> None:
self.overnight_returns.Add(overnight_return)
def is_ready(self) -> bool:
return self.returns_accumulated_last_month
def is_overnight_returns_ready(self) -> bool:
return self.overnight_returns.IsReady
# 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"))