“该策略涉及根据13个异常现象和调整后的账面市值比，将股票分类为价值加权的多空投资组合。选择调整后账面市值比最高的七个异常现象，并每年重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每年 | 市场: 股票 | 关键词: 价值效应
I. 策略概要
该投资范围包括纽约证券交易所、美国证券交易所和纳斯达克的普通股，不包括价格低于1美元的股票和金融股。创建了13种交易策略，根据各种异常现象将股票分类为价值加权的多空投资组合。对于每个异常投资组合，通过从t-1年的账面市值比中减去过去几年（t-6到t-2）的平均账面市值比来计算历史调整后的账面市值比（BM）。该策略投资于过去一年中调整后账面市值比最高的七个异常现象。该策略每年重新平衡，异常现象等权重。
II. 策略合理性
该论文通过学术研究中各种异常策略形成的投资组合，发现了一种强大的异常价值效应。与等权重相比，这种效应显示出增强的性能。该策略在不同的规范下仍然稳健，例如较少的赢家异常或分别处理多头和空头。它与个股或行业价值不同，也不能被它们解释。该论文表明，异常价值与行为理论相关，其中投资者对按异常特征排序的投资组合相关的信息反应不足。通过异常和价值因子进行双重排序，可以获得更盈利的投资组合，而这些投资组合不能仅用价值来解释。
III. 来源论文
Value and Momentum in Anomalies [点击查看论文]
- 德尼兹·安吉纳（Deniz Anginer）、苏加塔·雷（Sugata Ray）、H·内贾特·塞云（H. Nejat Seyhun）和徐路琪（Luqi Xu），西蒙弗雷泽大学（Simon Fraser University, SFU）、阿拉巴马大学经济、金融与法律研究系、密歇根大学斯蒂芬·M·罗斯商学院、查尔斯顿学院（College of Charleston）。
<摘要>
我们发现，当13个著名的股票市场异常现象相对于其历史水平表现出价值倾向时，它们未来的异常回报会更高。我们发现，表现出价值倾向（便宜）的异常现象比表现出增长倾向（昂贵）的异常现象未来每月超额回报约30个基点（bps）。此外，我们发现基于价值和动量综合倾向的有利异常现象比不利异常现象每月超额回报约90个基点，并且夏普比率翻倍以上。或者，当13个异常现象具有负动量和昂贵倾向时，其超过96%的美元回报会消失。
IV. 回测表现
| 年化回报 | 10.27% |
| 波动率 | 9.68% |
| β值 | 0.01 |
| 夏普比率 | 1.06 |
| 索提诺比率 | -0.665 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
import data_tools
from functools import reduce
class ValueInAnomalies(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.data:Dict[Symbol, data_tools.SymbolData] = {}
self.traded_quantity:Dict[Symbol, float] = {}
self.accruals_data:Dict[Symbol, data_tools.StockData] = {}
# Creating long and short portfolio according this list.
self.positive:List[str] = ['PEAD', 'ROA', 'GP', 'BM', 'MOM']
self.anomalies:Dict[str, Dict] = {
"AG": {}, # TotalAssets
"BM": {}, # Book to market = 1 / PBRatio
"GP": {}, # Gross Profitability = (TotalRevenue + CostOfGoods) / TotalAssets
"MOM": {}, # Twelve month momentum with first month skipped
"CEI": {}, # log(ME t/ ME t-5) – Return t-5
"NSI": {}, # Net stock issuance
"ACC": {}, # Accruals
"IVA": {}, # Investments to assets = (GrossPPE + ChangeInInventory) / TotalAssets
"ROA": {}, # Return on assets
"SIZE": {} # Market capitalization
# NOA = {} # Net operating assets. Missing Total Debt in current Liabilities, TotalPreferredStock(capital), TotalCommonStockEquity
# PEAD = {} # PEAD # Too complicated because of periods and missing data.
}
# Storing anomaly BM ratio for each year
self.anomalies_bm_ratio:Dict[str, RollingWindow] = {
"AG": RollingWindow[float](5),
"BM": RollingWindow[float](5),
"GP": RollingWindow[float](5),
"MOM": RollingWindow[float](5),
"CEI": RollingWindow[float](5),
"NSI": RollingWindow[float](5),
"ACC": RollingWindow[float](5),
"IVA": RollingWindow[float](5),
"ROA": RollingWindow[float](5),
"SIZE": RollingWindow[float](5)
}
self.month:int = 0 # Months counter
self.period:int = 12 * 21 # This is period for momentum
self.anomalies_count:int = 7 # Number of selected anomalies for trading
self.book_to_market_period:int = 12 # Each month storing book to market values for one year
self.leverage:int = 5
self.min_share_price:float = 5.
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.financial_statement_names:List[str] = [
'FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths',
'FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths',
'FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths',
'FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths',
'FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths',
'FinancialStatements.BalanceSheet.GrossPPE.TwelveMonths',
'ValuationRatios.CFOPerShare',
'ValuationRatios.TotalAssetPerShare',
'FinancialStatements.BalanceSheet.AccountsReceivable.TwelveMonths',
'FinancialStatements.IncomeStatement.TotalRevenueAsReported.TwelveMonths',
'ValuationRatios.PBRatio',
'MarketCap',
'FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths',
'FinancialStatements.IncomeStatement.CostOfRevenue.TwelveMonths',
'FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths',
]
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag:bool = False
self.rebalance_flag:bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.AddUniverse(self.FundamentalSelectionFunction)
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 RollingWindow each day
for stock in fundamental:
symbol:Symbol = stock.Symbol
if symbol in self.data:
self.data[symbol].update(stock.AdjustedPrice)
# Select each month
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
selected: List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Price >= self.min_share_price and x.Market == 'usa'
and x.SecurityReference.ExchangeId in self.exchange_codes
and all((not np.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]]
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = data_tools.SymbolData(self.period, self.book_to_market_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(close)
if self.data[symbol].is_ready():
self.data[symbol].update_market_caps(stock.MarketCap)
self.data[symbol].update_book_to_market(1 / stock.ValuationRatios.PBRatio)
if self.rebalance_flag:
if symbol not in self.accruals_data:
# Data for previous year.
self.accruals_data[symbol] = None
# Accrual calc.
current_accruals_data:data_tools.StockData = data_tools.StockData(stock.FinancialStatements.BalanceSheet.CurrentAssets.Value, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value,
stock.FinancialStatements.BalanceSheet.CurrentLiabilities.Value, stock.FinancialStatements.BalanceSheet.CurrentDebt.Value, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.Value,
stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value, stock.FinancialStatements.BalanceSheet.TotalAssets.Value,
stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value)
# There is not previous accruals data.
if not self.accruals_data[symbol]:
self.accruals_data[symbol] = current_accruals_data
continue
# Calculate current accruals.
current_accruals = self.CalculateAccruals(current_accruals_data, self.accruals_data[symbol])
# Store current accruals into dictionary for next calculations.
self.accruals_data[symbol] = current_accruals_data
# Start storing values for each anomaly only if book to market values are ready.
if not self.data[symbol].is_ready_book_to_market():
continue
# Store values for each anomaly
self.anomalies['ACC'][symbol] = current_accruals
self.anomalies['AG'][symbol] = stock.FinancialStatements.BalanceSheet.TotalAssets.Value
self.anomalies['BM'][symbol] = 1 / stock.ValuationRatios.PBRatio
self.anomalies['MOM'][symbol] = self.data[symbol].performance()
self.anomalies['CEI'][symbol] = self.data[symbol].cei(5) # Momentum for fifth month
self.anomalies['NSI'][symbol] = stock.FinancialStatements.CashFlowStatement.NetCommonStockIssuance.Value
self.anomalies['ROA'][symbol] = stock.OperationRatios.ROA.Value
self.anomalies['SIZE'][symbol] = stock.MarketCap
# Gross Profitability = (TotalRevenue + CostOfGoods) / TotalAssets
self.anomalies['GP'][symbol] = (stock.FinancialStatements.IncomeStatement.CostOfRevenue.Value + stock.FinancialStatements.IncomeStatement.TotalRevenue.Value) / stock.FinancialStatements.BalanceSheet.TotalAssets.Value
# Investments to assets = (GrossPPE + ChangeInInventory) / TotalAssets
self.anomalies['IVA'][symbol] = (stock.FinancialStatements.BalanceSheet.GrossPPE.Value + stock.FinancialStatements.CashFlowStatement.ChangeInInventory.Value) / stock.FinancialStatements.BalanceSheet.TotalAssets.Value
# Rebalance yearly
if not self.rebalance_flag:
return Universe.Unchanged
# There are no data in dictionaries, because book_to_market isn't warmed up.
if len(self.anomalies['AG']) == 0:
self.ClearAnomalies()
return Universe.Unchanged
# Store BM ratio for each anomaly
self.anomalies_bm_ratio['AG'].Add(self.CalculateAnomalyBMRatio(self.anomalies['AG'], 'AG'))
self.anomalies_bm_ratio['BM'].Add(self.CalculateAnomalyBMRatio(self.anomalies['BM'], 'BM'))
self.anomalies_bm_ratio['MOM'].Add(self.CalculateAnomalyBMRatio(self.anomalies['MOM'], 'MOM'))
self.anomalies_bm_ratio['CEI'].Add(self.CalculateAnomalyBMRatio(self.anomalies['CEI'], 'CEI'))
self.anomalies_bm_ratio['NSI'].Add(self.CalculateAnomalyBMRatio(self.anomalies['NSI'], 'NSI'))
self.anomalies_bm_ratio['ROA'].Add(self.CalculateAnomalyBMRatio(self.anomalies['ROA'], 'ROA'))
self.anomalies_bm_ratio['SIZE'].Add(self.CalculateAnomalyBMRatio(self.anomalies['SIZE'], 'SIZE'))
self.anomalies_bm_ratio['GP'].Add(self.CalculateAnomalyBMRatio(self.anomalies['GP'], 'GP'))
self.anomalies_bm_ratio['IVA'].Add(self.CalculateAnomalyBMRatio(self.anomalies['IVA'], 'IVA'))
self.anomalies_bm_ratio['ACC'].Add(self.CalculateAnomalyBMRatio(self.anomalies['ACC'], 'ACC'))
# Check if anomalies bm ratio values are ready.
if not self.anomalies_bm_ratio['AG'].IsReady:
self.ClearAnomalies() # Function clears each anomaly dictionary.
return Universe.Unchanged
history_adjusted_bm_ratio = {}
for anomaly, bm_ratios in self.anomalies_bm_ratio.items():
history_adjusted_bm_ratio[anomaly] = self.CalculateHistoryAdjustedBMRatio(bm_ratios)
# Sort anomalies based on their history_adjusted_bm_ratio.
sorted_by_hist_adj_bm_ratio = [x[0] for x in sorted(history_adjusted_bm_ratio.items(), key=lambda item: item[1])]
# Select highest anomalies for trading.
selected_anomalies = sorted_by_hist_adj_bm_ratio[-self.anomalies_count:]
# Create trading portfolio for each selected anomaly
for anomaly in selected_anomalies:
self.CreateLongAndShort(self.anomalies[anomaly], anomaly)
self.ClearAnomalies()
return list(self.traded_quantity.keys())
def OnData(self, data: Slice) -> None:
if not self.rebalance_flag:
return
self.rebalance_flag = False
# Trade execution
self.Liquidate()
# Trade stock with MarketOrder based on their quantity.
for symbol, quantity in self.traded_quantity.items():
if symbol in data and data[symbol]:
self.MarketOrder(symbol, quantity)
self.traded_quantity.clear()
def Selection(self) -> None:
self.selection_flag = True
if self.month != 12:
self.rebalance_flag = True
# Reset rebalance period
self.month = 0
self.month += 1
# Function calculates anomaly BM ratio
def CalculateAnomalyBMRatio(self, dictionary, anomaly):
quintile = int(len(dictionary) / 5)
# Sort dictionary by value and create list of symbols.
sorted_by_value = [x[0] for x in sorted(dictionary.items(), key=lambda item: item[1])]
long, short = self.SelectLongAndShort(sorted_by_value, quintile, anomaly)
bm_ratio = 0
total_anomaly_cap = sum([self.anomalies['SIZE'][symbol] for symbol in long + short])
# Sum BM weight for each stock in long and short
for symbol in long + short:
# self.anomalies['BM'][symbol] * market_cap / total_market_cap
bm_ratio += (self.anomalies['BM'][symbol] * (self.anomalies['SIZE'][symbol] / total_anomaly_cap))
return bm_ratio
# Function calculates history adjusted BM ratio for each anomaly.
def CalculateHistoryAdjustedBMRatio(self, bm_ratios_roll_window):
bm_ratios_values = [x for x in bm_ratios_roll_window]
# History adjusted BM ratio = current BM ratio - average from others BM ratios
return bm_ratios_values[0] - np.mean(bm_ratios_values[1:])
# Function creates long and short portfolio for each anomaly.
def CreateLongAndShort(self, dictionary, anomaly):
quintile = int(len(dictionary) / 5)
# Sort dictionary by value and create list of symbols.
sorted_by_value = [x[0] for x in sorted(dictionary.items(), key=lambda item: item[1])]
long, short = self.SelectLongAndShort(sorted_by_value, quintile, anomaly)
long_w = self.Portfolio.TotalPortfolioValue / self.anomalies_count
short_w = self.Portfolio.TotalPortfolioValue / self.anomalies_count
# Create anomaly long portfolio weighted by market cap
total_cap_long = sum([self.anomalies['SIZE'][symbol] for symbol in long])
for symbol in long:
# Each weight needs to be divided by total anomalies in portfolio
self.traded_quantity[symbol] = np.floor((long_w * (self.anomalies['SIZE'][symbol] / total_cap_long)) / self.data[symbol].last_price())
# Create anomaly short portfolio weighted by market cap
total_cap_short = sum([self.anomalies['SIZE'][symbol] for symbol in short])
for symbol in short:
# Each weight needs to be divided by total anomalies in portfolio
self.traded_quantity[symbol] = -np.floor((short_w * (self.anomalies['SIZE'][symbol] / total_cap_short)) / self.data[symbol].last_price())
def SelectLongAndShort(self, sorted_symbols, quintile, anomaly):
long = []
short = []
if anomaly in self.positive:
long = sorted_symbols[-quintile:] # Long top quintile
short = sorted_symbols[:quintile] # Short bottom quintile
else:
short = sorted_symbols[-quintile:] # Short top quintile
long = sorted_symbols[:quintile] # Long bottom quintile
return long, short
# Function clear each anomaly dictionary stored in self.anomalies dictionary.
def ClearAnomalies(self):
for _, dictionary in self.anomalies.items():
dictionary.clear()
# Source: https://papers.ssrn.com/sol3/papers.cfm?abstract_id=3188172
def CalculateAccruals(self, current_accrual_data, prev_accrual_data):
delta_assets = current_accrual_data.CurrentAssets - prev_accrual_data.CurrentAssets
delta_cash = current_accrual_data.CashAndCashEquivalents - prev_accrual_data.CashAndCashEquivalents
delta_liabilities = current_accrual_data.CurrentLiabilities - prev_accrual_data.CurrentLiabilities
delta_debt = current_accrual_data.CurrentDebt - prev_accrual_data.CurrentDebt
dep = current_accrual_data.DepreciationAndAmortization
total_assets_prev_year = prev_accrual_data.TotalAssets
acc = (delta_assets - delta_liabilities - delta_cash + delta_debt - dep) / total_assets_prev_year
return acc
# 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('.'))