“该策略投资于纽约证券交易所、美国证券交易所和纳斯达克的大盘股，结合了52周高点接近度和Piotroski FSCORE，做多排名靠前的股票，做空排名靠后的股票，每月重新排序并持有六个月。”
资产类别: 股票 | 地区: 美国 | 周期: 6个月 | 市场: 股票 | 关键词: 基本面
I. 策略概要
投资范围包括在纽约证券交易所、美国证券交易所和纳斯达克上市的普通股，不包括金融公司和股价低于5美元的股票。只考虑最大的公司。股票根据其与52周高点的接近程度（收盘价相对于过去12个月最高价）和Piotroski的FSCORE（衡量基本面实力的季度指标）进行评估。股票通过双向独立排序分为15个投资组合：52周接近度分为五分位，FSCORE分为三个组（低：0-3，中：4-6，高：7-9）。对两项指标均排名靠前的股票建立多头头寸，对两项指标均排名靠后的股票建立空头头寸。头寸持有六个月，每月重新排序，创建重叠的等权重投资组合。该策略整合了动量和基本面实力，以识别潜在的跑赢者和跑输者。
II. 策略合理性
52周高点异常是由受锚定偏差影响的非成熟投资者驱动的。将52周高点接近度与Piotroski的FSCORE相结合有助于识别对基本面消息反应不足的公司，为投资者创造机会。尽管q因子模型提供了一个有前景的基于风险的解释，但其有效性对动量崩溃期间的异常值敏感。接近52周高点的股票可能反映非基本面因素，这使得FSCORE成为一个重要的补充衡量指标，它来源于最近的财务报表。成熟投资者，如机构和卖空者，对FSCORE信号迅速做出反应，避免在接近52周高点时反应不足。研究结果与表明反应不足集中在非成熟投资者中的研究一致。
III. 来源论文
Fundamental Strength and the 52-Week High Anomaly [点击查看论文]
- 朱昭博（Zhu Zhaobo）、孙立成（Sun Licheng）和陈敏（Chen Min），深圳大学；奥登西亚商学院；老自治大学；旧金山州立大学会计系。
<摘要>
当股票在其52周高点附近交易时，投资者往往对其未来回报抱有较低的预期。我们将这种预期与公司的基本面实力进行对比。对于基本面强劲的公司，我们证实投资者的预期过低，这与52周高点作为心理锚点的假设一致。我们报告称，基本面实力增强的52周高点交易策略的平均回报几乎是无条件策略的两倍，显著优于后者。此外，我们提供了有趣的证据，表明这种异常效应在投资者情绪高涨时最为明显，但在更成熟的机构和卖空者中则不存在。
IV. 回测表现
| 年化回报 | 12.68% |
| 波动率 | 21.06% |
| β值 | -0.159 |
| 夏普比率 | 0.6 |
| 索提诺比率 | 0.179 |
| 最大回撤 | N/A |
| 胜率 | 48% |
V. 完整的 Python 代码
from numpy import floor, isnan
from AlgorithmImports import *
from pandas.core.frame import dataframe
import data_tools
from functools import reduce
class FundamentalStrength52WeekHigh(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.period:int = 52 * 5
self.holding_period:int = 6
self.quantile:int = 5
self.leverage:int = 5
self.min_share_price:float = 5.
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.data:Dict[Symbol, data_tools.SymbolData] = {}
self.last_fine:List[Symbol] = []
self.managed_queue:List[data_tools.RebalanceQueueItem] = []
self.financial_statement_names:List[str] = [
'EarningReports.BasicAverageShares.ThreeMonths',
'EarningReports.BasicEPS.TwelveMonths',
'OperationRatios.ROA.ThreeMonths',
'OperationRatios.GrossMargin.ThreeMonths',
'FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.ThreeMonths',
'FinancialStatements.IncomeStatement.NormalizedIncome.ThreeMonths',
'FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths',
'FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths',
'FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths',
'FinancialStatements.IncomeStatement.TotalRevenueAsReported.ThreeMonths',
'ValuationRatios.PERatio',
'OperationRatios.CurrentRatio.ThreeMonths',
]
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_count:int = 3000
self.fundamental_sorting_key = lambda x: x.MarketCap
self.selection_flag = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthStart(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(data_tools.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 x.Price > self.min_share_price and \
all((not isnan(self.rgetattr(x, statement_name)) and self.rgetattr(x, statement_name) != 0) for statement_name in self.financial_statement_names)
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
score = {}
nearness = {}
# warmup price rolling windows
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = data_tools.SymbolData(symbol, self.period)
history = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update_price(close)
if self.data[symbol].is_ready():
nearness[symbol] = self.data[symbol].nearness()
# FSCORE calc
roa:float = stock.OperationRatios.ROA.ThreeMonths
cfo:float = stock.FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.ThreeMonths
leverage:float = stock.FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths / stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
liquidity:float = stock.OperationRatios.CurrentRatio.ThreeMonths
equity_offering:float = stock.FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths
gross_margin:float = stock.OperationRatios.GrossMargin.ThreeMonths
turnover:float = stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.ThreeMonths / stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
symbol_data = self.data[symbol]
# check if data has previous year's data ready and their values are consecutive
if (not symbol_data.data_is_set()) or (symbol not in self.last_fine):
symbol_data.update_data(roa, leverage, liquidity, equity_offering, gross_margin, turnover)
continue
score[symbol] = 0
if roa > 0:
score[symbol] += 1
if cfo > 0:
score[symbol] += 1
if roa > symbol_data.ROA: # ROA change is positive
score[symbol] += 1
if cfo > roa:
score[symbol] += 1
if leverage < symbol_data.Leverage:
score[symbol] += 1
if liquidity > symbol_data.Liquidity:
score[symbol] += 1
if equity_offering < symbol_data.Equity_offering:
score[symbol] += 1
if gross_margin > symbol_data.Gross_margin:
score[symbol] += 1
if turnover > symbol_data.Turnover:
score[symbol] += 1
# assing new (this year's) data
symbol_data.update_data(roa, leverage, liquidity, equity_offering, gross_margin, turnover)
long:List[Symbol] = []
short:List[Symbol] = []
if len(score) != 0 and len(nearness) >= self.quantile:
# nearness sorting and F score sorting
sorted_by_nearness:List[Symbol] = sorted(nearness, key = nearness.get, reverse = True)
quantile:int = int(len(sorted_by_nearness) / self.quantile)
high_by_nearness = sorted_by_nearness[:quantile]
low_by_nearness = sorted_by_nearness[-quantile:]
long = [x[0] for x in score.items() if x[1] >= 7 and x[0] in high_by_nearness]
short = [x[0] for x in score.items() if x[1] <= 3 and x[0] in low_by_nearness]
if len(long) != 0 and len(short) != 0:
long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
# symbol/quantity collection
long_symbol_q:List[Tuple] = [(x, floor(long_w / self.data[x].LastPrice)) for x in long]
short_symbol_q:List[Tuple] = [(x, -floor(short_w / self.data[x].LastPrice)) for x in short]
self.managed_queue.append(data_tools.RebalanceQueueItem(long_symbol_q + short_symbol_q))
self.last_fine = [x.Symbol for x in selected]
return long + short
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
remove_item = None
# rebalancing portfolio
for item in self.managed_queue:
if item.holding_period == self.holding_period:
# liquidate
for symbol, quantity in item.symbol_q:
self.MarketOrder(symbol, -quantity)
remove_item = item
# trade execution
if item.holding_period == 0:
open_symbol_q = []
for symbol, quantity in item.symbol_q:
if data.ContainsKey(symbol):
self.MarketOrder(symbol, quantity)
open_symbol_q.append((symbol, quantity))
# only opened orders will be closed
item.symbol_q = open_symbol_q
item.holding_period += 1
# we need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue
if remove_item:
self.managed_queue.remove(remove_item)
def Selection(self) -> None:
self.selection_flag = True
# https://gist.github.com/wonderbeyond/d293e7a2af1de4873f2d757edd580288
def rgetattr(self, obj, attr, *args):
def _getattr(obj, attr):
return getattr(obj, attr, *args)
return reduce(_getattr, [obj] + attr.split('.'))