行业投资组合中的长期反转与动量效应相结合

“该策略交易主要行业的ETF，买入近期表现强劲的长期输家，卖空近期表现疲软的长期赢家，每季度重新平衡，等权重头寸持有三个月。”

资产类别: ETF | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 长期反转，动量

I. 策略概要

该策略交易追踪48个主要股票行业的ETF，采用后期逆向投资方法。每个季度，行业根据120个月的表现分为四分位数。从顶部（赢家）和底部（输家）四分位数中，行业根据12个月的表现进一步排名。该策略买入近期表现强劲的长期输家，卖出近期表现疲软的长期赢家。头寸等权重，持有三个月，并每季度重新平衡，旨在利用长期表现极端情况下的短期趋势驱动的表现反转。

II. 策略合理性

学术论文认为，行业受到消费者偏好、宏观经济变化、技术进步和监管转变等因素的影响是不均衡的。这可能导致一些行业长期表现不佳，而另一些行业则蓬勃发展。然而，这种极端情况是不可持续的，通常会促使结构性调整，例如陷入困境的行业进行合并、收购或退出，从而最终实现整合和改善状况。相比之下，表现强劲的行业可能会面临日益激烈的竞争，从而削弱其前景。投资者可能无法认识到这些结构性变化，导致他们反应不足，从而导致回报反转：过去回报较低的行业未来可能会获得较高的回报，反之亦然。

III. 来源论文

行业长期回报反转 [点击查看论文]

• Bornholt、Gharaibeh、Malin，格里菲斯大学，格里菲斯大学 – 格里菲斯商学院，格里菲斯大学 – 会计、金融与经济学系

<摘要>

鉴于极端的行业回报可能预示着相关行业长期结构性变化，最终可能导致行业命运逆转，我们研究了行业回报长期回报逆转的证据。我们的研究采用了纯粹的逆向策略和后期逆向策略，并包括超长的策略形成期（长达132个月），以便为结构性变化开始留出足够的时间。我们发现了行业长期回报出现逆转的有力证据。这些逆转持续多年（观察到估值效应在开始后长达十年），并且很难用过度反应来解释。

IV. 回测表现

年化回报	9.9%
波动率	19.49%
β值	0.165
夏普比率	0.3
索提诺比率	0.218
最大回撤	N/A
胜率	50%

V. 完整的 Python 代码

from AlgorithmImports import *
class LongTermReversalCombinedWithMomentumEffectAlgorithm(QCAlgorithm):
    def Initialize(self):
        
        self.SetStartDate(2010, 1, 1) 
        self.SetCash(100000)      
        self.tickers = [
            "XLY", # Consumer Discretionary Select Sector SPDR Fund
            "PBS", # Invesco Dynamic Media ETF
            "PEJ", # Invesco Dynamic Leisure and Entertainment ETF
            "PMR", # Invesco Dynamic Retail ETF
                
            "XLP", # Consumer Staples Select Sector SPDR Fund
            "PBJ", # Invesco Dynamic Food & Beverage ETF
                
            "XLE", # Energy Select Sector SPDR Fund
            "PBW", # Invesco WilderHill Clean Energy ETF
            "PXE", # Invesco Dynamic Energy Exploration & Production ETF
            "NLR", # VanEck Vectors Uranium+Nuclear Energy ETF
            "AMJ", # JPMorgan Alerian MLP Index ETN
                
            "XLF", # Financial Select Sector SPDR Fund
            "KBE", # SPDR S&P Bank ETF
            "KIE", # SPDR S&P Insurance ETF
            "KRE", # SPDR S&P Regional Banking ETF
            "PSP", # Invesco Global Listed Private Equity ETF
                
            "XLV", # Health Care Select Sector SPDR Fund
            "IBB", # iShares Nasdaq Biotechnology ETF
            "IHF", # iShares U.S. Healthcare Providers ETF
            "IHE", # iShares U.S. Pharmaceuticals ETF
                
            "XLI", # Industrial Select Sector SPDR Fund
            "ITA", # iShares U.S. Aerospace & Defense ETF
            "IYT", # iShares Transportation Average ETF
            "PHI", # Invesco Water Resources ETF
                
            "XLB", # Materials Select Sector SPDR ETF
            "MOO", # VanEck Vectors Agribusiness ETF
            "GDX", # VanEck Vectors Gold Miners ETF
            "XHB", # SPDR S&P Homebuilders ETF
            "IGE", # iShares North American Natural Resources ETF
                
            "XLK", # Technology Select Sector SPDR Fund
            "FDN", # First Trust Dow Jones Internet Index
            "SOXX", # iShares PHLX Semiconductor ETF
            "IGV", # iShares Expanded Tech-Software Sector ET
            "IYZ", # iShares U.S. Telecommunications ETF
                
            "XLU", # Utilities Select Sector SPDR Fund
            "IGF", # iShares Global Infrastructure ETF
        ]
        
        self.data = {} # Storing closes
        self.symbols = []
        
        self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        
        self.current_month = 0
        self.needed_months = 120
        self.months_for_short_perf = 12
        self.period = self.needed_months * 21
        self.SetWarmUp(self.period)
        
        for ticker in self.tickers:
            security = self.AddEquity(ticker, Resolution.Daily)
            security.SetFeeModel(CustomFeeModel())
            security.SetLeverage(5)
            
            self.data[security.Symbol] = SymbolData(self.period)
            self.symbols.append(security.Symbol)
         
        self.selection_flag = False    
        self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
        
    def OnData(self, data):
        for symbol in self.symbols:
            if symbol in data:
                if data[symbol]:
                    price = data[symbol].Value
                    self.data[symbol].update(price)
        
        if self.IsWarmingUp: return
        
        if not self.selection_flag:
            return
        self.selection_flag = False
        
        performances = {}
        for symbol in self.symbols:
            if self.data[symbol].is_ready():
                performances[symbol] = self.data[symbol].performance()
            
        sorted_by_performance = [x[0] for x in sorted(performances.items(), key=lambda item: item[1])]
        
        quartile = int(len(sorted_by_performance) / 4)
        winners = sorted_by_performance[-quartile:]
        losers = sorted_by_performance[:quartile]
                
        short_term_perf_losers = {}
        short_term_perf_winners = {}
        
        for symbol in losers:
            short_term_perf_losers[symbol] = self.data[symbol].short_term_performance(self.months_for_short_perf)
        
        for symbol in winners:
            short_term_perf_winners[symbol] = self.data[symbol].short_term_performance(self.months_for_short_perf)
            
        short_term_losers_sort_by_perf = [x[0] for x in sorted(short_term_perf_losers.items(), key=lambda item: item[1])]
        short_term_winners_sort_by_perf = [x[0] for x in sorted(short_term_perf_winners.items(), key=lambda item: item[1])]
        
        quartile = int(len(short_term_winners_sort_by_perf) / 4)
        short = short_term_winners_sort_by_perf[:quartile]
        
        quartile = int(len(short_term_losers_sort_by_perf) / 4)
        long = short_term_losers_sort_by_perf[-quartile:]
        
        # Trade Execution
        stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
        for symbol in stocks_invested:
            if symbol not in long + short:
                self.Liquidate(symbol)
        
        long_length = len(long)
        short_length = len(short)
        
        for symbol in long:
            if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
                self.SetHoldings(symbol, 1 / long_length)
        
        for symbol in short:
            if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
                self.SetHoldings(symbol, -1 / short_length)
        
    def Selection(self):
        self.current_month += 1
        
        if self.current_month % 3 == 0: # Selection each quarter
            self.selection_flag = True
        
class SymbolData():
    def __init__(self, period):
        self.Closes = RollingWindow[float](period)
        
    def update(self, close):
        self.Closes.Add(close)
        
    def is_ready(self):
        return self.Closes.IsReady
        
    def performance(self):
        closes = [x for x in self.Closes]
        return (closes[0] - closes[-1]) / closes[-1]
        
    def short_term_performance(self, months):
        closes = [x for x in self.Closes][:21 * months] # Takes last months
        return (closes[0] - closes[-1]) / closes[-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"))