“该策略交易纽约证券交易所、美国证券交易所和纳斯达克股票，买入隔夜表现强势的股票，卖空隔夜表现疲软的股票，使用过去的表现或回归模型，持有隔夜头寸，并在开盘时清算。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 隔夜
I. 策略概要
该策略针对纽约证券交易所、美国证券交易所和纳斯达克股票，重点关注隔夜回报模式。每年，通过基于过去年度表现排序（前十分位数）或使用回归模型（等式8）识别易受强或弱隔夜回报影响的股票，来选择具有较高隔夜回报的股票。每日，投资者买入被归类为强隔夜表现的股票，卖空弱隔夜表现的股票。头寸等权重，持有过夜，并在开盘时清算。这种多空方法系统地利用隔夜价格波动，利用历史模式和预测模型来获取目标回报。
II. 策略合理性
学术论文将“隔夜动量”利润的持续性归因于预期隔夜回报的横截面差异。预期隔夜回报较高的股票往往保持在表现最佳的十分位数中，从而推动持续的利润。正如康拉德和考尔（1998）所建议的那样，即使在随机游走模型中，这种效应也可能发生。与风险等与市场定价因素相关的典型回报特征不同，隔夜回报和日内回报与风险呈现相反的关系。隔夜回报和日内回报之间强烈的负相关关系表明，抵消偏差造成的扭曲，呈现出超越标准市场特征的回报动态中的独特场景。
III. 来源论文
夜盘交易：更低风险，更高回报？[点击查看论文]
- Lachance, 圣地亚哥州立大学 – 金融系。
<摘要>
本文证明，隔夜回报存在高度持续的偏差，并在此背景下检验仅隔夜投资的盈利能力。隔夜回报往往超过其日内回报，本文首先通过引入一个考虑重复偏差的模型来调和这些模式。该模型识别出五分之一的股票具有积极且统计上显着的隔夜偏差。来年对这些股票进行隔夜投资，收益是市场回报的两倍，而市场贝塔系数仅为三分之一。结果也对日间投资者有影响，因为这些股票的日内平均回报为负。文中讨论了实施成本和问题。
IV. 回测表现
| 年化回报 | 21.28% |
| 波动率 | 6.6% |
| β值 | -0.126 |
| 夏普比率 | 2.62 |
| 索提诺比率 | 0.479 |
| 最大回撤 | N/A |
| 胜率 | 51% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from typing import Dict, List
class OvernightStockTrading(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.leverage:int = 10
self.quantile:int = 10
self.period:int = 12 * 21
self.min_share_price:int = 5
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.fundamental_count:int = 100
self.selected:List[Symbol] = [] # symbols of selected stocks from fundamentalSelectionFunction
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.data:Dict[Symbol, SymbolData] = {}
self.months_counter:int = 1
self.selection_flag:bool = True
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.EveryDay(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol, 20), self.MarketClose)
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection)
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]:
# updating overnight data and stock price every day
for stock in fundamental:
symbol:Symbol = stock.Symbol
if symbol in self.data:
# update stock price
self.data[symbol].price = stock.AdjustedPrice
# get history data
history:dataframe = self.History(symbol, 1, Resolution.Daily)
# update overnight return and change prev_close_price
self.UpdateOvernightReturns(history, symbol)
# one year rebalance
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
# sort stocks by dollar volume
selected:List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa'
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol in self.data:
continue
# create object of SymbolData class for current stock
self.data[symbol] = SymbolData(self.period)
# get history data
history:dataframe = self.History(symbol, self.period + 1, Resolution.Daily)
# update overnight return and change prev_close_price
self.UpdateOvernightReturns(history, symbol)
# change self.selected list on rebalance
self.selected = [x.Symbol for x in selected]
return self.selected
def MarketClose(self) -> None:
total_performance:Dict[Symbol, float] = {} # storing total overnight returns performance for self.period overnight returns
# calculate total overnight performances
for symbol in self.selected:
if not self.data[symbol].is_overnight_returns_ready():
continue
# calculate and store total overnight performance
total_performance[symbol] = self.data[symbol].total_overnight_performance()
if len(total_performance) >= self.quantile:
# quantile selection
quantile:int = int(len(total_performance) / self.quantile)
sorted_by_total_perf:List[Symbol] = [x[0] for x in sorted(total_performance.items(), key=lambda item: item[1])]
# long top quantile stocks and short bottom quantile stocks
self.long = sorted_by_total_perf[-quantile:]
self.short = sorted_by_total_perf[:quantile]
long_length:int = len(self.long)
short_length:int = len(self.short)
# trade execution
for symbol in self.long:
current_price = self.data[symbol].price
if current_price != 0 and self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
quantity = np.floor((self.Portfolio.TotalPortfolioValue / long_length) / current_price)
self.MarketOnCloseOrder(symbol, quantity)
self.MarketOnOpenOrder(symbol, -quantity)
for symbol in self.short:
current_price = self.data[symbol].price
if current_price != 0 and self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
quantity = np.floor((self.Portfolio.TotalPortfolioValue / short_length) / current_price)
self.MarketOnCloseOrder(symbol, -quantity)
self.MarketOnOpenOrder(symbol, quantity)
def UpdateOvernightReturns(self, history, symbol: Symbol) -> None:
''' update overnight returns for specific stock according to history data '''
# check if history isn't empty and history dataframe has required attributes
if not history.empty and hasattr(history, 'close') and hasattr(history, 'open'):
# get open and close prices from dataframe
opens:Series = history['open']
closes:Series = history['close']
# update overnight return
for (_, open_price), (_, close_price) in zip(opens.items(), closes.items()):
# update overnight return and change prev_close_price
self.data[symbol].update(open_price, close_price)
def Selection(self) -> None:
if self.months_counter % 12 == 0:
self.selection_flag = True
self.months_counter += 1
class SymbolData():
def __init__(self, period: int):
self.overnight_returns:RollingWindow = RollingWindow[float](period)
self.prev_close_price:Union[None, float] = None
self.price:int = 0
def update(self, open_price: float, close_price: float) -> None:
# update overnight returns only if prev_close_price isn't None
if self.prev_close_price:
overnight_return = open_price / self.prev_close_price - 1
self.overnight_returns.Add(overnight_return)
# change previous close price to current close price
self.prev_close_price = close_price
def total_overnight_performance(self) -> float:
return sum(list(self.overnight_returns))
def is_overnight_returns_ready(self) -> bool:
return self.overnight_returns.IsReady
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))