“该策略在宏观经济公告日投资于CRSP股票的最高和最低贝塔十分位数，买入高贝塔股票并卖空低贝塔股票，持仓一天。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 宏观经济、贝塔
I. 策略概要
该策略侧重于CRSP股票中最高和最低贝塔十分位数，按其过去五年预排名CAPM贝塔进行排名。股票数量取决于多元化需求。头寸在宏观经济公告日（例如就业数据、通胀报告和联邦公开市场委员会会议）建立。在公告日，该策略买入最高贝塔十分位数的股票，并卖空最低贝塔十分位数的股票，持仓仅一天。这种方法旨在利用市场对宏观经济新闻的反应，使用贝塔作为风险因素来指导交易。
II. 策略合理性
该策略利用了公告日市场回报显著高于普通日的事实。使用基本线性回归和CAPM模型，证券市场线在公告日显示出显著的正斜率和不显著的截距。这证实了CAPM模型并解释了做多高贝塔股票和做空低贝塔股票的盈利能力。CAPM模型表明回报由贝塔因子驱动，截距（阿尔法）不显著。因此，该策略侧重于市场敞口（贝塔），通过交易高低贝塔股票来利用正斜率。
III. 来源论文
Post Macroeconomic Announcement Reversal [点击查看论文]
- 牛子龙（Zilong Niu）和张天睿（Terry Zhang），西南财经大学金融研究院，澳大利亚国立大学（ANU）
<摘要>
我们记录了在坏的宏观经济消息发布后的几天，股市继续下跌，并且证券市场线具有显著的负斜率。我们发现好的宏观经济消息发布后，回报持续的证据较弱。这些发现表明市场在公告日对坏消息反应不足。当中介资本稀缺且卖空限制更严格的股票中，反应不足更强，这与套利限制理论一致。新闻初始市场反应的这种不对称性夸大了公告溢价。使用更长的窗口来衡量公告回报会导致公告溢价不显著。
IV. 回测表现
| 年化回报 | 10.37% |
| 波动率 | 8.79% |
| β值 | 0.073 |
| 夏普比率 | 1.18 |
| 索提诺比率 | -0.14 |
| 最大回撤 | N/A |
| 胜率 | 49% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
from pandas.tseries.offsets import BDay
from scipy import stats
from data_tools import CustomFeeModel, SymbolData
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
#endregion
class MacroeconomicAnnouncementBeta(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.leverage: int = 5
self.quantile: int = 10
self.period: int = 5 * 12
self.daily_period: int = 21
self.days_offset: int = 1
self.data: Dict[Symbol, SymbolData] = {}
self.selected_symbols: List[Symbol] = []
self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.data[self.market] = SymbolData(self.period)
self.WarmUpSymbolPrices(self.market)
csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/economic_announcements.csv')
dates: List[str] = csv_string_file.split('\r\n')
before_announcement_dates: List[datetime.date] = [(datetime.strptime(x, '%Y-%m-%d') - BDay(self.days_offset)).date() for x in dates]
self.fundamental_count: int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.sort_flag: bool = False
self.selection_flag: bool = False
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.On(before_announcement_dates), self.TimeRules.AfterMarketOpen(self.market), self.DayBeforeAnnouncement)
self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
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]:
# monthly selection
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
selected: List[fundamental] = []
self.selected_symbols.clear()
for stock in fundamental:
symbol: Symbol = stock.Symbol
if symbol in self.data and self.data[symbol].is_last_month_price_ready():
self.data[symbol].update_monthly_return(stock.AdjustedPrice)
if stock.HasFundamentalData:
selected.append(stock)
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] = SymbolData(self.period)
self.WarmUpSymbolPrices(symbol)
if self.data[symbol].is_ready():
self.selected_symbols.append(symbol)
return self.selected_symbols
def OnData(self, data: Slice) -> None:
self.Liquidate()
# sort one day before earnings annoucement
if not self.sort_flag or len(self.selected_symbols) == 0:
return
self.sort_flag = False
# make sure there is n years of SPY monthly returns history
if not self.data[self.market].is_ready():
return
market_monthly_returns: List[float] = self.data[self.market].get_monthly_returns()
beta: Dict[Symbol, float] = {}
for symbol in self.selected_symbols:
if symbol in data and data[symbol]:
stock_monthly_returns: List[float] = self.data[symbol].get_monthly_returns()
# linear regression - X = market returns, Y = stock returns
slope, intercept, r_value, p_value, std_err = stats.linregress(market_monthly_returns, stock_monthly_returns)
beta[symbol] = slope
# check if there are enough stocks for selection
if len(beta) < self.quantile:
self.Liquidate()
return
# beta sorting
quantile: int = int(len(beta) / self.quantile)
sorted_by_beta: List[Symbol] = [x[0] for x in sorted(beta.items(), key=lambda item: item[1])]
long: List[Symbol] = sorted_by_beta[-quantile:]
short: List[Symbol] = sorted_by_beta[:quantile]
# trade execution
targets: List[PortfolioTarget] = []
for i, portfolio in enumerate([long, short]):
for symbol in portfolio:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
def WarmUpSymbolPrices(self, symbol: Symbol) -> None:
history: dataframe = self.History([symbol], self.daily_period * self.period, Resolution.Daily)
if history.empty:
return
closes: Series = history.loc[symbol].close
closes_grouped: Series = closes.groupby(pd.Grouper(freq='M')).last()
for close in closes_grouped:
self.data[symbol].update_monthly_return(close)
def Selection(self) -> None:
self.selection_flag = True
def DayBeforeAnnouncement(self) -> None:
self.sort_flag = True