“该策略将股票分为十个分位，交易最好的和最差的分位，使用MA50/MA200信号每月调整仓位，并在每年6月进行组合再平衡。”
资产类别: 股票 | 地区: 全球 | 周期: 每月 | 市场: 股票 | 关键词: 趋势跟随
I. 策略概要
策略每年6月基于前一个财政年度的八个会计变量对所有股票进行排名，并将其分配到十个分位（Deciles）。创建等权重投资组合，设立一个扩展投资组合，在平均回报最高的分位（Decile High）上建立多头仓位，在最低回报的分位（Decile Low）上建立空头仓位。投资组合每年6月重新平衡。每月，投资者计算每只股票的50日和200日移动平均线（MA50和MA200）。如果Decile High的股票MA50低于MA200，则将其剔除；如果Decile Low的股票MA50高于MA200，则将其剔除，从而精细化投资组合。
II. 策略合理性
许多股票异常的一个共同特征是它们基于低频信息形成，这意味着投资组合通常每年进行一次重新平衡。然而，股票价格数据是以更高频率提供的，投资者可以利用这些信息来更新他基于低频信息形成的投资组合观点。
III. 来源论文
Anomalies Enhanced: A Portfolio Rebalancing Approach [点击查看论文]
- Yufeng Han, Dayong Huang 和 Guofu Zhou. 北卡罗来纳大学夏洛特分校 – 金融学系。北卡罗来纳大学格林斯伯勒分校 – 布莱恩商学院与经济学院。圣路易斯华盛顿大学 – 约翰·M·奥林商学院。
<摘要>
许多异常现象是基于公司特征，并且每年重新平衡一次，忽略了年度中的任何信息。在本文中，我们提供了动态交易策略，每月重新平衡异常投资组合。对于八个主要异常现象，我们发现这些动态交易策略显著提升了它们的经济重要性，在Fama和French（2015）五因素风险调整异常收益的改善幅度在每月0.40%到0.75%之间。结果对多种控制变量具有稳健性。我们的研究结果表明，许多著名的异常现象比之前认为的更有利可图，为它们的理论解释带来了新的挑战。
IV. 回测表现
| 年化回报 | 16.18% |
| 波动率 | 15.48% |
| β值 | 0.05 |
| 夏普比率 | 1.05 |
| 索提诺比率 | -0.008 |
| 最大回撤 | N/A |
| 胜率 | 43% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from functools import reduce
from numpy import isnan
from pandas.core.frame import dataframe
class ClassicalEquityAnomaliesCombinedwithTrendfollowingFilter(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.financial_statement_names:List[str] = [
'ValuationRatios.PBRatio',
'FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths',
'FinancialStatements.IncomeStatement.TotalOperatingIncomeAsReported.TwelveMonths',
'OperationRatios.TotalAssetsGrowth.OneYear',
'FinancialStatements.CashFlowStatement.CapitalExpenditure.TwelveMonths',
'EarningReports.BasicAverageShares.TwelveMonths',
# accruals data
'FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths',
'FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths',
'FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths',
'FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths',
'FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths',
'FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths',
'FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths'
]
# Daily sma data.
self.data:Dict[Symbol, SymbolData] = {}
self.short_period:int = 50
self.long_period:int = 200
self.performance_period:int = 12*21
self.quantile:int = 10
self.leverage:int = 5
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.fundamental_count:int = 1000
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag:bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.BeforeMarketClose(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 SMA every day.
for stock in fundamental:
symbol:Symbol = stock.Symbol
# Store monthly price.
if symbol in self.data:
self.data[symbol].update(self.Time, 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 \
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]]
# Ranked stocks.
rank = {}
current_accruals_data = {}
acc = {}
momentum = {}
net_stock_issue = {}
capex = {}
noa = {}
selected_stocks:List[Fundamental] = []
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = SymbolData(self.short_period, self.long_period, self.performance_period)
history:dataframe = self.History(symbol, self.performance_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(time, close)
if self.data[symbol].is_ready():
selected_stocks.append(stock)
rank[symbol] = 0
momentum[symbol] = self.data[symbol].performance()
# Accural calc
current_accruals_data = AccrualsData(stock.FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths,
stock.FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths, stock.FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths,
stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths, stock.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths)
if self.data[symbol].accruals_data:
acc[symbol] = self.CalculateAccruals(current_accruals_data, self.data[symbol].accruals_data)
# Calculate NOA.
noa[symbol] = (stock.FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths - stock.FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths) / self.data[symbol].accruals_data.TotalAssets
# Calculate stock stock issue.
num_of_shares = stock.EarningReports.BasicAverageShares.TwelveMonths
if self.data[symbol].shares_outstanding:
net_stock_issue[symbol] = num_of_shares / self.data[symbol].shares_outstanding - 1
self.data[symbol].shares_outstanding = num_of_shares
# Calculate stock capital expenditure.
cap_expenditures = stock.FinancialStatements.CashFlowStatement.CapitalExpenditure.TwelveMonths
if self.data[symbol].capex:
capex[symbol] = cap_expenditures / self.data[symbol].capex - 1
self.data[symbol].capex = cap_expenditures
# Update accruals data for next year's calculation.
self.data[symbol].accruals_data = current_accruals_data
# Ensure that consecutive accruals and shares data are available.
for symbol in self.data:
if symbol not in list(map(lambda x: x.Symbol, selected_stocks)):
self.data[symbol].accruals_data = None
self.data[symbol].shares_outstanding = None
self.data[symbol].capex = None
# Sort by book to market.
sorted_by_bm = sorted(selected_stocks, key=lambda x: 1 / x.ValuationRatios.PBRatio)
for index, stock in enumerate(sorted_by_bm):
rank[stock.Symbol] += (index+1)
# Sort by gross profit.
sorted_by_gp = sorted(selected_stocks, key=lambda x: x.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths / x.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths)
for index, stock in enumerate(sorted_by_gp):
rank[stock.Symbol] += (index+1)
# Sort by operating profit.
sorted_by_op = sorted(selected_stocks, key=lambda x: x.FinancialStatements.IncomeStatement.TotalOperatingIncomeAsReported.TwelveMonths)
for index, stock in enumerate(sorted_by_op):
rank[stock.Symbol] += (index+1)
# Sort by asset growth.
sorted_by_ag = sorted(selected_stocks, key=lambda x: x.OperationRatios.TotalAssetsGrowth.OneYear, reverse = True)
for index, stock in enumerate(sorted_by_ag):
rank[stock.Symbol] += (index+1)
# Sort by investments.
sorted_by_inv = sorted(capex.items(), key=lambda x: x[1], reverse = True)
for index, symbol_si in enumerate(sorted_by_inv):
rank[symbol_si[0]] += (index+1)
# Sort by net stock issue.
sorted_by_si = sorted(net_stock_issue.items(), key=lambda x: x[1], reverse = True)
for index, symbol_si in enumerate(sorted_by_si):
rank[symbol_si[0]] += (index+1)
# Sort by accruals.
sorted_by_acc = sorted(acc.items(), key=lambda x: x[1])
for index, symbol_acc in enumerate(sorted_by_acc):
rank[symbol_acc[0]] += (index+1)
# Sort by NOA.
sorted_by_noa = sorted(noa.items(), key=lambda x: x[1], reverse = True)
for index, symbol_noa in enumerate(sorted_by_noa):
rank[symbol_noa[0]] += (index+1)
if len(rank) > self.quantile:
# Rank sorting.
sorted_by_rank:List = sorted(rank.items(), key=lambda x: x[1], reverse = True)
quantile:int = int(len(sorted_by_rank) / self.quantile)
top_decile:List[Symbol] = [x[0] for x in sorted_by_rank[:quantile]]
bottom_decile:List[Symbol] = [x[0] for x in sorted_by_rank[-quantile:]]
top_decile_perf:float = np.mean([momentum[x] for x in top_decile])
bottom_decile_perf:float = np.mean([momentum[x] for x in bottom_decile])
if top_decile_perf > bottom_decile_perf:
self.long = top_decile
self.short = bottom_decile
else:
self.long = bottom_decile
self.short = top_decile
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:
# Drop stocks according to trend filter.
symbols_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in symbols_invested:
if self.Portfolio[symbol].IsLong:
if not self.data[symbol].is_in_uptrend():
self.Liquidate(symbol)
elif self.Portfolio[symbol].IsShort:
if self.data[symbol].is_in_uptrend():
self.Liquidate(symbol)
if self.Time.month == 6:
self.selection_flag = True
def CalculateAccruals(self, current_accrual_data, prev_accrual_data) -> float:
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
delta_tax = current_accrual_data.IncomeTaxPayable - prev_accrual_data.IncomeTaxPayable
dep = current_accrual_data.DepreciationAndAmortization
avg_total = (current_accrual_data.TotalAssets + prev_accrual_data.TotalAssets) / 2
bs_acc = ((delta_assets - delta_cash) - (delta_liabilities - delta_debt-delta_tax) - dep) / avg_total
return bs_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('.'))
class AccrualsData():
def __init__(self, current_assets, cash_and_cash_equivalents, current_liabilities, current_debt, income_tax_payable, depreciation_and_amortization, total_assets):
self.CurrentAssets = current_assets
self.CashAndCashEquivalents = cash_and_cash_equivalents
self.CurrentLiabilities = current_liabilities
self.CurrentDebt = current_debt
self.IncomeTaxPayable = income_tax_payable
self.DepreciationAndAmortization = depreciation_and_amortization
self.TotalAssets = total_assets
class SymbolData():
def __init__(self, short_sma_period, long_sma_period, performance_period):
self.price = RollingWindow[float](performance_period)
self.short_sma = SimpleMovingAverage(short_sma_period)
self.long_sma = SimpleMovingAverage(long_sma_period)
self.accruals_data = None
self.shares_outstanding = None
self.capex = None
def update(self, time: datetime, value: float):
self.short_sma.Update(time, value)
self.long_sma.Update(time, value)
self.price.Add(value)
def is_ready(self) -> bool:
return (self.short_sma.IsReady and self.long_sma.IsReady and self.price.IsReady)
def is_in_uptrend(self) -> bool:
return (self.short_sma.Current.Value > self.long_sma.Current.Value)
def performance(self) -> float:
return self.price[0] / 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"))