“该策略根据12个月的日内动量交易中国股票，对排名前十的股票做多，对排名后十的股票做空，采用价值加权，每月重新平衡日内头寸。”
资产类别: 股票 | 地区: 中国 | 周期: 每月 | 市场: 股票 | 关键词: 动量
I. 策略概要
该策略侧重于CSMAR数据库中的中国股票。日内回报计算为收盘价减去开盘价，再除以开盘价。计算所有股票的12个月累计日内回报（日内动量），然后将其分为十等份。该策略对排名前十的股票（日内动量最高）采取价值加权多头头寸，对排名后十的股票（日内动量最低）采取空头头寸。头寸在日内持有，每月重新平衡，利用日内动量捕捉中国股票市场的短期回报模式。
II. 策略合理性
中国股市由于独特的制度因素和强大的投资者异质性而与其他市场显著不同。主要区别包括用于设定开盘价的10分钟盘前集合竞价，这通过纳入新的隔夜信息促进了价格发现。在盘前交易的投资者是早期知情者，而那些在接近收盘时交易的投资者是后期知情者。此外，散户投资者占据主导地位，贡献了A股交易量的80%以上，这与机构驱动的美国市场不同。
动量策略在中国整体表现不佳。日内（开盘至收盘）和隔夜（收盘至开盘）回报呈负相关，日内强劲上涨的股票往往在隔夜反转，反之亦然。虽然日内（隔夜）赢家在随后的日内（隔夜）期间表现优于输家，但这种持续性在另一个期间会反转，形成一种“拉锯战”效应，从而削弱了动量策略。
与投资者异质性一致，日内强劲上涨的股票往往是规模小、换手率高、高增长的公司，具有高特质风险、有限的分析师覆盖和显著的机构持股。这种回报模式具有高度持续性，并且在控制了众所周知的回报预测因子后仍然稳健，这反映了中国市场的独特动态以及传统基于动量策略的挑战。
III. 来源论文
Investor Heterogeneity and Momentum-based Trading Strategies in China [点击查看论文]
- 高亚（Ya Gaoa）、韩星（Xing Hanb）、李有为（Youwei Lic）和熊雄（Xiong Xiong）。天津大学管理与经济学部、奥克兰大学商学院、赫尔大学商学院、天津大学管理与经济学部及中国社会计算与分析中心
<摘要>
传统的动量策略在中国整体表现不佳，因为股票价格在市场开盘交易和休市时表现截然不同。过去的日内（隔夜）赢家在随后的日内（隔夜）期间持续跑赢过去的日内（隔夜）输家。然而，同样的日内（隔夜）动量策略在随后的隔夜（日内）期间表现却大幅下滑。进一步分析表明，过去的日内（隔夜）赢家往往是日间（夜间）需求量大的更具（或更少）投机性的股票。总的来说，我们的结果与投资者异质性一致，这种持续的拉锯战几乎消除了投资者在中国追求基于动量交易策略的有效性。
IV. 回测表现
| 年化回报 | 36.87% |
| 波动率 | 23.5% |
| β值 | 0.01 |
| 夏普比率 | 1.57 |
| 索提诺比率 | -0.018 |
| 最大回撤 | N/A |
| 胜率 | 47% |
V. 完整的 Python 代码
from AlgorithmImports import *
from numpy import floor
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
#endregion
class TheIntradayMomentumInChina(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2015, 1, 1)
self.SetCash(100_000)
self.tickers_to_ignore: List[str] = ['EVK']
self.period: int = 12 * 21
self.quantile: int = 10
self.sort_chunk: float = 0.3
self.leverage: int = 10
self.traded_portion: float = 0.2
self.data: Dict[Symbol, SymbolData] = {}
self.weight: Dict[Symbol, float] = {}
symbol: Symbol = self.AddEquity("SPY", Resolution.Minute).Symbol
time_offset: int = 16
self.selection_flag: bool = False
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Minute
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
self.Schedule.On(self.DateRules.EveryDay(symbol), self.TimeRules.AfterMarketOpen(symbol, -time_offset), self.BeforeOpen)
self.Schedule.On(self.DateRules.EveryDay(symbol), self.TimeRules.BeforeMarketClose(symbol, time_offset), self.BeforeClose)
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]:
if not self.selection_flag:
return Universe.Unchanged
selected: List[Fundamental] = [
x for x in fundamental
if x.HasFundamentalData
and x.MarketCap != 0
and x.CompanyReference.BusinessCountryID == 'CHN'
and x.Symbol.Value not in self.tickers_to_ignore
]
selected_stocks = [x for x in sorted(selected, key=lambda x:x.MarketCap, reverse=True)][:int(len(selected) * self.sort_chunk)]
market_cap: Dict[Symbol, float] = {}
cum_intraday_returns: Dict[Symbol, float] = {}
for stock in selected_stocks:
symbol: Symbol = stock.Symbol
if symbol not in self.data:
# Collect opens and closes of stock
self.data[symbol] = SymbolData(self.period)
history: dataframe = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet")
continue
closes: Series = history.loc[symbol].close
opens: Series = history.loc[symbol].open
for (_, close_price), (_, open_price) in zip(closes.items(), opens.items()):
self.data[symbol].update(close_price, open_price)
if self.data[symbol].is_ready():
cum_intraday_returns[symbol] = self.data[symbol].cumulative_intraday_returns()
market_cap[symbol] = stock.MarketCap
if len(cum_intraday_returns) < self.quantile:
return Universe.Unchanged
quantile: int = int(len(cum_intraday_returns) / self.quantile)
sorted_by_cum_intraday_returns: List[Symbol] = [x[0] for x in sorted(cum_intraday_returns.items(), key=lambda item: item[1])]
long: List[Symbol] = sorted_by_cum_intraday_returns[-quantile:]
short: List[Symbol] = sorted_by_cum_intraday_returns[:quantile]
# Need to clear last long and short portfolio before creating the new one.
self.weight.clear()
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda symbol: market_cap[symbol], portfolio)))
for symbol in portfolio:
self.weight[symbol] = ((-1)**i) * market_cap[symbol] / mc_sum
# Long and short portfolio is selected only once in a month.
self.selection_flag = False
return long + short
def OnData(self, data: Slice) -> None:
if (self.Time.hour == 0 and self.Time.minute == 0):
# Updating RollingWindow each day with new open and close price.
for symbol in self.data:
history: History = self.History(symbol, 1, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet")
continue
close: Series = history.loc[symbol].close
open: Series = history.loc[symbol].open
for (_, close_price), (_, open_price) in zip(close.items(), open.items()):
self.data[symbol].update(close_price, open_price)
def BeforeOpen(self) -> None:
# open new positions
for symbol, w in self.weight.items():
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
quantity = floor((1 * self.Portfolio.TotalPortfolioValue * w) * self.traded_portion / self.data[symbol].LastPrice)
self.MarketOnOpenOrder(symbol, quantity)
def BeforeClose(self) -> None:
# liquidate
for symbol, w in self.weight.items():
if self.Portfolio[symbol].Invested:
quantity: int = self.Portfolio[symbol].Quantity
self.MarketOnCloseOrder(symbol, -quantity)
def Selection(self):
self.selection_flag = True
class SymbolData():
def __init__(self, period: int) -> None:
self.Closes: RollingWindow = RollingWindow[float](period)
self.Opens: RollingWindow = RollingWindow[float](period)
self.LastPrice: float = 0.
def update(self, close_price: float, open_price: float) -> None:
self.Closes.Add(close_price)
self.Opens.Add(open_price)
self.LastPrice = close_price
def update_opens(self, open_price: float) -> None:
self.Opens.Add(open_price)
def update_closes(self, close_price: float) -> None:
self.Closes.Add(close_price)
def is_ready(self) -> bool:
return self.Closes.IsReady and self.Opens.IsReady
def cumulative_intraday_returns(self) -> float:
closes: List[float] = [x for x in self.Closes]
opens: List[float] = [x for x in self.Opens]
intraday_returns: List[float] = [(close_price - open_price) / open_price for close_price, open_price in zip(closes, opens)]
return sum(intraday_returns)
# 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"))