“该策略选择流动性好的纽约证券交易所和美国证券交易所股票，结合动量和投资与资产比率，做多高动量、低投资的股票，做空低动量、高投资的股票，并每半年重新平衡一次。”
资产类别: 股票 | 地区: 美国 | 周期: 6个月 | 市场: 股票 | 关键词: 投资、动量
I. 策略概要
该策略侧重于价格高于5美元的纽约证券交易所和美国证券交易所普通股（股票代码10和11），选择买卖价差最低的流动性最佳的股票。通过结合六个月的过去回报（动量）和投资与资产（I/A）比率，构建了一个复合投资-动量（InvMom）策略。I/A的计算方法是毛财产、厂房、设备和存货的年度变化除以滞后资产账面价值。
股票每月根据动量和I/A双重排序为五分位，创建5×5的投资组合。该策略做多高动量、低投资的股票，做空低动量、高投资的股票，持仓六个月（第2个月至第7个月）。投资组合等权重并每半年重新平衡一次，利用流动性、动量和投资指标来提高业绩。
II. 策略合理性
动量和投资是金融文献中已确立的因子。动量源于羊群效应、过度反应、反应不足和确认偏误等行为偏差。投资因子表明，保守型投资组合的表现优于激进型投资组合，因为激进型投资往往无法改善近期回报。这项研究强调利用市场效率低下的多个维度，而不是用新的基本面来增强动量策略。通过结合动量和投资因子，该策略产生比单一因子更强劲、更持久的结果，即使在个别策略表现不佳的时期（例如2000-2015年）也是如此。该组合策略在买卖价差低的股票中表现稳健，突出了其实用性和有效性。
III. 来源论文
Investment-Momentum: A Two-Dimensional Behavioral Strategy [点击查看论文]
- 徐方明（Fangming Xu）、赵淮南（Huainan Zhao）和郑立懿（Liyi Zheng），布里斯托大学，拉夫堡大学商学院，布里斯托大学。
<摘要>
我们提出了一种投资-动量策略，即买入过去表现良好且投资较低的股票，卖出过去表现不差且投资较高的股票，该策略同时利用了市场效率低下的两个维度。1965-2015年，新策略产生的月回报是价格动量或投资策略的两倍（1.44% vs. 0.75%或0.61%）。尽管近几十年来异常现象逐渐减少，但投资-动量策略仍然保持持久。基于错误定价的策略在投资者情绪高涨或套利限制较高的股票中表现更好，这与我们的预期一致。总的来说，我们表明，除了“基本面”增强的动量策略之外，还可以同时利用多维度低效率来实现卓越表现。
IV. 回测表现
| 年化回报 | 8.99% |
| 波动率 | 11.66% |
| β值 | -0.121 |
| 夏普比率 | 0.77 |
| 索提诺比率 | 0.059 |
| 最大回撤 | N/A |
| 胜率 | 49% |
V. 完整的 Python 代码
from AlgorithmImports import *
from pandas.core.frame import dataframe
from numpy import isnan
class InvestmentMomentumStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))
market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.data:Dict[Symbol, SymbolData] = {}
self.period:int = 6 * 21
self.leverage:int = 5
self.quantile:int = 5
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.months:int = 0
self.selection_flag = True
self.UniverseSettings.Resolution = Resolution.Daily
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
self.settings.daily_precise_end_time = False
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]:
# Update the rolling window every day.
for stock in fundamental:
symbol:Symbol = stock.Symbol
# Store monthly price.
if symbol in self.data:
self.data[symbol].update_price(stock.AdjustedPrice)
if not self.selection_flag:
return Universe.Unchanged
selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
not isnan(x.FinancialStatements.BalanceSheet.GrossPPE.TwelveMonths) and x.FinancialStatements.BalanceSheet.GrossPPE.TwelveMonths != 0 and \
not isnan(x.FinancialStatements.BalanceSheet.Inventory.TwelveMonths) and x.FinancialStatements.BalanceSheet.Inventory.TwelveMonths != 0 and \
not isnan(x.FinancialStatements.BalanceSheet.TangibleBookValue.ThreeMonths) and x.FinancialStatements.BalanceSheet.TangibleBookValue.ThreeMonths != 0
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
selected_symbols:List[Symbol] = [x.Symbol for x in selected]
momentum:Dict[Symbol, float] = {}
ia_ratio:Dict[Symbol, float] = {}
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
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:pd.Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update_price(close)
if self.data[symbol].is_ready():
momentum[symbol] = self.data[symbol].performance()
ppe:float = stock.FinancialStatements.BalanceSheet.GrossPPE.TwelveMonths
inv:float = stock.FinancialStatements.BalanceSheet.Inventory.TwelveMonths
book_value:float = stock.FinancialStatements.BalanceSheet.TangibleBookValue.ThreeMonths
if self.data[symbol]._inventory == 0 or self.data[symbol]._PPE == 0 or self.data[symbol]._book_value == 0:
self.data[symbol].update_data(ppe, inv, book_value)
continue
ppe_change:float = ppe - self.data[symbol]._PPE
inv_change:float = inv - self.data[symbol]._inventory
# IA calc
ia_ratio[symbol] = (ppe_change + inv_change) / book_value
self.data[symbol].update_data(ppe, inv, book_value)
# Reset not updated symbols.
for symbol in self.data:
if symbol not in selected_symbols:
self.data[symbol].update_data(0,0,0)
if len(momentum) >= self.quantile:
# Momentum sorting
sorted_by_mom:List[Symbol] = sorted(momentum, key = momentum.get, reverse = True)
quantile:int = int(len(sorted_by_mom) / self.quantile)
high_by_mom:List[Symbol] = sorted_by_mom[:quantile]
low_by_mom:List[Symbol] = sorted_by_mom[-quantile:]
# IA sorting
sorted_by_ia:List[Symbol] = sorted(ia_ratio, key = ia_ratio.get, reverse = True)
quantile = int(len(sorted_by_ia) / self.quantile)
high_by_ia:List[Symbol] = sorted_by_ia[:quantile]
low_by_ia:List[Symbol] = sorted_by_ia[-quantile:]
self.long = [x for x in high_by_mom if x in low_by_ia]
self.short = [x for x in low_by_mom if x in high_by_ia]
return self.long + self.short
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# order execution
targets:List[PortfolioTarget] = []
for i, portfolio in enumerate([self.long, self.short]):
for symbol in portfolio:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
self.long.clear()
self.short.clear()
def Selection(self) -> None:
if self.months == 0 or self.months == 5:
self.selection_flag = True
self.months += 1
if self.months == 6:
self.months = 0
class SymbolData():
def __init__(self, period:int):
self._price:RollingWindow = RollingWindow[float](period)
self._PPE:float = 0.
self._inventory:float = 0.
self._book_value:float = 0.
def update_data(self, ppe:float, inv:float, book_value:float) -> None:
self._PPE = ppe
self._inventory = inv
self._book_value = book_value
def update_price(self, price:float) -> None:
self._price.Add(price)
def is_ready(self) -> bool:
return self._price.IsReady
def performance(self, values_to_skip = 0) -> float:
return self._price[values_to_skip] / self._price[self._price.Count - 1] - 1
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))