宏观经济公告贝塔反转

“该策略根据预排名的CAPM贝塔买入低贝塔股票并卖空高贝塔CRSP股票，对宏观经济公告做出反应，并在公告后持有头寸两天。”

资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 宏观经济、贝塔、反转

I. 策略概要

该策略侧重于CRSP股票中最高和最低贝塔十分位数，按其过去五年预排名CAPM贝塔进行排名。股票数量根据多元化需求进行调整。在宏观经济公告日，例如就业数据、通胀报告和联邦公开市场委员会会议，该策略买入最低贝塔十分位数的股票，并卖空最高贝塔十分位数的股票。头寸在公告后持有两天。该策略旨在从市场对宏观经济新闻的反应中获利，利用贝塔作为风险因素来指导交易。

II. 策略合理性

该策略基于CAPM资产定价模型，使用线性回归来理解市场行为。在公告日，证券市场线具有显著的正斜率，但在公告后，它会翻转为负斜率，其中高贝塔股票显示负回报，低贝塔股票显示正回报。这种模式在各种稳健性检查和样本期中都是一致的。该策略的功能并非基于风险，因为在公告后没有观察到市场投资组合风险的降低。相反，这种模式可能反映了坏消息的处理速度比好消息慢。股票价格在公告日跳涨，但在公告后，价格会修正，这为交易策略提供了基础。

III. 来源论文

Post Macroeconomic Announcement Reversal [点击查看论文]

• 牛子龙（Zilong Niu）和张天睿（Terry Zhang），西南财经大学金融研究院，澳大利亚国立大学（ANU）。

<摘要>

我们记录了在坏的宏观经济消息发布后的几天，股市继续下跌，并且证券市场线具有显著的负斜率。我们发现好的宏观经济消息发布后，回报持续的证据较弱。这些发现表明市场在公告日对坏消息反应不足。当中介资本稀缺且卖空限制更严格的股票中，反应不足更强，这与套利限制理论一致。新闻初始市场反应的这种不对称性夸大了公告溢价。使用更长的窗口来衡量公告回报会导致公告溢价不显著。

IV. 回测表现

年化回报	7.76%
波动率	9.02%
β值	-0.036
夏普比率	0.86
索提诺比率	-0.119
最大回撤	N/A
胜率	52%

V. 完整的 Python 代码

from AlgorithmImports import *
import numpy as np
from pandas.tseries.offsets import BDay
from scipy import stats
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
#endregion
class MacroeconomicAnnouncementBetaReversal(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2000, 1, 1)
        self.SetCash(100000)
        
        self.market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
        self.period: int = 5 * 12
        self.daily_period: int = 21
        self.quantile: int = 10
        self.leverage: int = 5
        
        self.data: Dict[Symbol, SymbolData] = {}
        self.selected_symbols: List[Symbol] = []
        self.long: List[Symbol] = []
        self.short: List[Symbol] = []
        csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/economic_announcements.csv')
        dates: List[str] = csv_string_file.split('\r\n')
        announcement_dates: List[datetime.date] = [datetime.strptime(x, "%Y-%m-%d") for x in dates]
        sort_dates: List[datetime.date] = [(x + BDay(1)).date() for x in announcement_dates]
        liquidation_dates: List[datetime.date] = [(x + BDay(2)).date() for x in announcement_dates]
        
        self.fundamental_count: int = 500
        self.fundamental_sorting_key = lambda x: x.DollarVolume
        self.selection_flag: bool = False
        self.rebalance_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.MonthEnd(self.market), self.TimeRules.At(0,0), self.Selection)
        self.Schedule.On(self.DateRules.On(sort_dates), self.TimeRules.At(0,0), self.Rebalance)
        self.Schedule.On(self.DateRules.On(liquidation_dates), self.TimeRules.At(0,0), self.Liquidation)
    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
        
        # calculate monthly return
        for stock in fundamental:
            symbol = stock.Symbol
            
            # check if current stock have last month price
            if symbol in self.data and self.data[symbol].last_month_price:
                self.data[symbol].update_monthly_return(stock.AdjustedPrice)
        
        selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData]
        if len(selected) > self.fundamental_count:
            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
        
        # Store monthly return for every stock selected this month.
        for stock in selected + [self.market]:
            if stock == self.market:
                symbol = stock
            else:
                symbol: Symbol = stock.Symbol
            if symbol in self.data:
                continue
            
            self.data[symbol] = SymbolData(self.period)
            history: dataframe = self.History([symbol], self.daily_period * self.period, Resolution.Daily)
            
            if history.empty:
                self.Log(f"Not enough data for {symbol} yet")
                continue
            
            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)
        
        # get stocks, which have ready monthly returns
        self.selected_symbols = [x.Symbol for x in selected if self.data[x.Symbol].is_ready() and x.Symbol != self.market]        
        
        return self.selected_symbols
    
    def OnData(self, data: Slice) -> None:
        if not self.rebalance_flag:
            return
        self.rebalance_flag = False
        # there has to be at least one selected symbol and market returns has to be ready
        if len(self.selected_symbols) == 0 or not self.data[self.market].is_ready():
            return
        
        market_monthly_returns: List[float] = [x for x in self.data[self.market].monthly_returns]
        
        beta: Dict[Symbol, float] = {}
        for symbol in self.selected_symbols:
            stock_monthly_returns: List[float] = [x for x in self.data[symbol].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 data for decile selection    
        if len(beta) < self.quantile:
            self.Liquidate()
            return
          
        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 the lowest beta decile
        self.long: List[Symbol] = sorted_by_beta[:quantile]
        # short the highest beta decile
        self.short: List[Symbol] = sorted_by_beta[-quantile:]
        
        # Trade 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)
    def Rebalance(self) -> None:
        self.rebalance_flag = True
    def Selection(self) -> None:
        self.selection_flag = True
        
    def Liquidation(self) -> None:
        self.Liquidate()
class SymbolData():
    def __init__(self, period: int) -> None:
        self.monthly_returns: RollingWindow = RollingWindow[float](period)
        self.last_month_price = 0
        
    def update_monthly_return(self, price: float) -> None:
        if self.last_month_price != 0:
            monthly_return: float = (price - self.last_month_price) / self.last_month_price
            self.monthly_returns.Add(monthly_return)
        self.last_month_price = price
        
    def is_ready(self) -> bool:
        return self.monthly_returns.IsReady
# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
        fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))