“通过前一日回报交易纽约证券交易所、纳斯达克和美国证券交易所的股票，做多回报最低的十分位，做空回报最高的十分位，持有价值加权投资组合直至市场收盘。”
资产类别: 股票 | 地区: 美国 | 周期: 日内 | 市场: 股票 | 关键词: 反转
I. 策略概要
投资范围包括股票代码为“10”或“11”的纽约证券交易所、纳斯达克和美国证券交易所股票。股票价格必须高于5美元，在过去一个月内至少有10个有效的日回报，并且至少有六个月的上市历史。每月应用筛选器，每日排除派发股息或进行股票拆分的股票。
对于每只股票，计算前一日回报为上午10:00的价格除以前一日收盘价。下午3:30，做多回报最低的十分位，做空回报最高的十分位。排除上午10:00至下午3:30之间绝对回报超过5%或在此期间没有交易的股票。投资组合按价值加权，持有至当日市场收盘（30分钟）。
II. 策略合理性
研究表明，美国股票市场的短期反转效应是由非流动性驱动的，补偿投资者提供流动性和承担风险。这种效应在交易的最后30分钟最为显著，这可能归因于机构投资者在早期和晚期交易时段占据主导地位（Gao et al., 2018）。此外，即使考虑了传统的风险因素，包括规模、价值、动量、波动性和交易量，反转效应仍然保持稳健。这突显了机构交易模式和市场微观结构在强化美国股票市场短期反转现象中的作用。
III. 来源论文
What Drives Intraday Reversal? Illiquidity or Liquidity Oversupply? [点击查看论文]
- 康俊清、林申、熊雄。中山大学岭南（大学）学院。天津大学管理与经济学院；清华大学五道口金融学院。管理与经济学院和中国社会计算与分析中心
<摘要>
先前对美国市场的研究将短期反转视为流动性提供的补偿。然而，我们发现日内反转与中国市场的股票流动性没有显著依赖关系。因此，基于一个程式化框架，我们提出了另一种解释：非理性的非知情流动性提供者，由于生理锚定而低估了均衡价格中的信息成分，他们逆着先前的价格变动进行交易，从而产生相反的价格压力。实证结果证实了这种流动性过剩（来自非理性的非知情流动性提供者）的解释。一旦我们延长持有期，中国市场日内回报与未来回报之间的负相关关系就会反转。这表明反转是由于非知情散户交易者过度提供流动性而导致的定价错误，而不是由于缺乏流动性而导致的临时价格让步所引起的价格修正。
IV. 回测表现
| 年化回报 | 22.42% |
| 波动率 | 5.41% |
| β值 | -0.005 |
| 夏普比率 | 4.15 |
| 索提诺比率 | -1.858 |
| 最大回撤 | N/A |
| 胜率 | 46% |
V. 完整的 Python 代码
from AlgorithmImports import *
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
#endregion
class IntradayReversalInUS(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2010, 1, 1)
self.SetCash(100_000)
self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
self.quantile: int = 10
self.leverage: int = 5
self.min_share_price: int = 5
self.percentage_threshold: float = 0.05
self.data: Dict[Symbol, SymbolData] = {} # Storing important data for each selected stock in this strategy
self.selected_symbols: List[Symbol] = []
self.fundamental_count: int = 100
self.fundamental_sorting_key = lambda x: x.DollarVolume
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.
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# Each day select stocks for trading
selected: List[Fundamental] = [
x for x in fundamental
if x.HasFundamentalData
and x.Market == 'usa'
and x.Price > self.min_share_price
and x.MarketCap != 0
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]]
# Create list of selected symbols and create SymbolData object for each of these stocks
for stock in selected:
symbol: Symbol = stock.Symbol
# Create SymbolData object and store stock's market capitalization
self.data[symbol] = SymbolData(stock.MarketCap)
self.selected_symbols.append(symbol)
return self.selected_symbols
def OnData(self, data: Slice) -> None:
selected: List[Symbol] = []
# Calculate overnight return
if self.Time.hour == 10 and self.Time.minute == 0:
# Select only stocks, which return in absolute value isn't greater than 5%
for symbol in self.selected_symbols:
if symbol in data and data[symbol]:
# Get current price from data object
current_price: float = data[symbol].Value
# Store current price, for next calculation
self.data[symbol].open_price = current_price
# Use history to get last close
history: dataframe = self.History(symbol, 1, Resolution.Daily)
if history.empty:
continue
# Get last close from history
closes: Series = history.loc[symbol].close
close: float = closes[0]
# Calculate over night return
over_night_return: float = (current_price - close) / close
# Select current stock, if stock's absolute value of over night return is equal or smaller than 5%
if abs(over_night_return) <= self.percentage_threshold:
self.data[symbol].over_night_return = over_night_return
selected.append(symbol)
self.selected_symbols = selected
# Trade 30 minutes before market close
if self.Time.hour == 15 and self.Time.minute == 30:
over_night_returns: Dict[Symbol, float] = {}
# Select stocks, which don't have return in absolute value greater than 5% since 10:00 a.m. to 3:30 p.m.
for symbol in self.selected_symbols:
if symbol in data and data[symbol]:
current_price: float = data[symbol].Value
open_price: float = self.data[symbol].open_price
# Calculate return since since 10:00 a.m. to 3:30 p.m
daily_return: float = (current_price - open_price) / open_price
# Select current stock, if stock's absolute value of daily return is equal or smaller than 5%
if abs(daily_return) <= self.percentage_threshold:
# self.data[symbol].daily_return = daily_return
over_night_returns[symbol] = self.data[symbol].over_night_return
# Check if we have enough stocks for trade
if len(over_night_returns) < self.quantile:
return
# Sort stocks based on overnight return
quantile: int = int(len(over_night_returns) / self.quantile)
sorted_by_overnight_ret: List[Symbol] = [x[0] for x in sorted(over_night_returns.items(), key=lambda item: item[1])]
# Go long(short) on the decile of stocks with the lowest(highest) measure
long: List[Symbol] = sorted_by_overnight_ret[:quantile]
short: List[Symbol] = sorted_by_overnight_ret[-quantile:]
# Trade execution
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda symbol: self.data[symbol].market_cap, portfolio)))
for symbol in portfolio:
self.SetHoldings(symbol, ((-1)**i) * self.data[symbol].market_cap / mc_sum)
# Liquidate one minute before market close
if self.Time.hour == 15 and self.Time.minute == 59:
# Clear dictionaries for next intra day trading
self.selected_symbols.clear()
self.data.clear()
self.Liquidate() # liquidate all stocks
class SymbolData():
def __init__(self, market_cap: float) -> None:
self.market_cap: float = market_cap
self.open_price: Union[None, float] = None
self.over_night_return: Union[None, float] = None
self.daily_return: Union[None, float] = None
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0
return OrderFee(CashAmount(fee, "USD"))