“通过流动性增长交易美国股票，做多高流动性增长（LIQG）股票，做空低流动性增长（LIQG）股票（流动性强和流动性弱的），使用美元交易量加权，每月重新平衡的投资组合。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 流动性、增长因子
I. 策略概要
投资范围包括CRSP数据库中的美国股票（纽约证券交易所、美国证券交易所、纳斯达克）。流动性计算为每月交易股数乘以月末价格。流动性增长（LIQG）通过比较当月（t月）与去年同月（t-12月）的流动性得出。对于非负累计回报的股票，LIQG是流动性的相对差异；对于负累计回报的股票，则是负绝对差异。
股票被分为2×3组：流动性强/流动性弱和低/中性/高LIQG。该策略做多高LIQG股票（包括流动性强和流动性弱的），做空低LIQG股票。投资组合按美元交易量加权，每月重新平衡。
II. 策略合理性
该策略的功能源于流动性作为股票回报因素的重要性及其与长期以交易量为重点的投资者的相关性。低流动性增长的股票，通常不被注意，预计会产生更高的回报，而高流动性增长的股票，通常交易过度，应该产生更低的回报。然而，研究并未完全支持这种行为解释。相反，流动性增长被认为是风险因素，反映了投资者对流动性增长风险的担忧。此外，流动性增长因子与动量密切相关，表明存在与流动性增长相关的潜在风险溢价，进一步增强了其在解释股票回报中的作用。
III. 来源论文
Earnings and Liquidity Factors [点击查看论文]
- Robert Snigaroff and David Wroblewski, Denali Advisors, Denali Advisors
<摘要>
一个包含盈利、流动性、各自增长以及市场因素的模型可以提供一个具有低定价误差的消费理由。它还包含了为期一年的动量和剔除反转后的动量，即通常所说的“动量”因子。这些盈利和流动性因子都非常重要，并且结合起来形成了一个没有因子冗余的模型。受投资者建立头寸能力的启发，我们基于交易量构建投资组合，并将流动性整合到补充基于公司因子的简化形式的基于特征的因子模型中。
IV. 回测表现
| 年化回报 | 7.18% |
| 波动率 | 11.29% |
| β值 | 0 |
| 夏普比率 | 0.63 |
| 索提诺比率 | -94.825 |
| 最大回撤 | N/A |
| 胜率 | 44% |
V. 完整的 Python 代码
from AlgorithmImports import *
import pandas as pd
from collections import deque
import data_tools
#endregion
class LiquidityGrowthFactor(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2002, 1, 1)
self.SetCash(100_000)
market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.fundamental_count: int = 1_000
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.data: Dict[Symbol, data_tools.SymbolData] = {}
self.weight: Dict[Symbol, float] = {}
self.monthly_period: int = 12
self.daily_period: int = 30
self.volume_period: int = 21
self.leverage: int = 5
self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
self.selection_flag: bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.settings.daily_precise_end_time = False
self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(data_tools.CustomFeeModel())
security.SetLeverage(self.leverage)
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
for stock in fundamental:
symbol: Symbol = stock.Symbol
if symbol in self.data:
# store price and volume and dollar volume every day
self.data[symbol].update_price(stock.AdjustedPrice)
self.data[symbol].update_volume(stock.Volume)
self.data[symbol].update_dollar_volume(stock.DollarVolume)
if self.selection_flag and self.data[symbol].volume_is_ready() and self.data[symbol].price_is_ready():
# store monthly liquidity
liquidity: float = self.data[symbol].monthly_liquidity()
if liquidity != 0:
self.data[symbol].update_liquidity(liquidity, self.Time)
if not self.selection_flag:
return Universe.Unchanged
selected: List[Fundamental] = [
f for f in fundamental if f.HasFundamentalData
and f.Market == 'usa'
and f.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]]
liquidity: Dict[Symbol, float] = {}
liquidity_growth: Dict[Symbol, float] = {}
# warmup price rolling windows
for stock in selected:
symbol: Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = data_tools.SymbolData(self.monthly_period+1, self.monthly_period+1, self.volume_period)
history: dataframe = self.History(symbol, self.monthly_period*self.daily_period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
if 'close' in history and 'volume' in history:
closes: Series = history.loc[symbol]['close']
volumes: Series = history.loc[symbol]['volume']
# find monthly closes and update rolling window
closes_len: int = len(closes.keys())
for index, time_close in enumerate(closes.items()):
# index out of bounds check
if index + 1 < closes_len:
date_month: int = time_close[0].date().month
next_date_month: int = closes.keys()[index + 1].month
# found last day of month
if date_month != next_date_month:
self.data[symbol].update_price(time_close[1])
# monthly grouped data
shares_traded_grouped = volumes.groupby(pd.Grouper(freq='M')).sum()
closes_grouped = closes.groupby(pd.Grouper(freq='M')).last()
liquidity_grouped = shares_traded_grouped * closes_grouped
for i, liq in enumerate(liquidity_grouped):
self.data[symbol].update_liquidity(liq, closes_grouped.index[i])
# liq and liq growth calc
if self.data[symbol].liquidity_is_ready() \
and self.data[symbol].price_is_ready() \
and self.data[symbol].dollar_volume_is_ready():
lg: float = self.data[stock.Symbol].liquidity_growth()
if lg != -1:
liquidity[symbol] = self.data[stock.Symbol].recent_liquidity()
liquidity_growth[symbol] = lg
long: List[Symbol] = []
short: List[Symbol] = []
# sorting
min_stock_cnt: int = 6
if len(liquidity) >= min_stock_cnt and len(liquidity_growth) >= min_stock_cnt:
sorted_by_liq = sorted(liquidity, key = liquidity.get, reverse = True)
half: int = int(len(sorted_by_liq) / 2)
liquid: List[Symbol] = sorted_by_liq[:half]
iliquid: List[Symbol] = sorted_by_liq[-half:]
percentile: int = int(len(liquid) * 0.3)
liquid_by_growth: List[Symbol] = sorted(liquid, key = lambda x: liquidity_growth[x], reverse = True)
high_liquid: List[Symbol] = liquid_by_growth[:percentile]
low_liquid: List[Symbol] = liquid_by_growth[-percentile:]
iliquid_by_growth: List[Symbol] = sorted(iliquid, key = lambda x: liquidity_growth[x], reverse = True)
high_iliquid: List[Symbol] = iliquid_by_growth[:percentile]
low_iliquid: List[Symbol] = iliquid_by_growth[-percentile:]
long = high_liquid + high_iliquid
short = low_liquid + low_iliquid
# dollar volume weighting
for i, portfolio in enumerate([long, short]):
total_dollar_vol: float = sum([self.data[x].monthly_dollar_volume() for x in portfolio])
for stock in portfolio:
self.weight[symbol] = ((-1) ** i) * (self.data[symbol].monthly_dollar_volume() / total_dollar_vol)
return list(self.weight.keys())
def OnData(self, slice: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# trade execution
portfolio: List[PortfolioTarget] = [
PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if slice.contains_key(symbol) and slice[symbol]
]
self.SetHoldings(portfolio, True)
self.weight.clear()
def Selection(self):
self.selection_flag = True