投资范围包括上述国家的行业指数。投资组合按市值分为等权重五分位,计算“范围”变量(每日最大值减去最小值)。在小市值组合中,做多“范围”值高的股票,做空“范围”值低的股票;在大市值组合中则反向操作。该策略每月再平衡。

策略概述

投资范围包括上述国家的行业指数。将投资组合按市值分为等权重五分位。计算排序变量“范围”,即上个月的每日最大值减去每日最小值。在每个五分位中,按“范围”对股票进行排序。在小市值组合中,做多“范围”值较高的股票,做空“范围”值较低的股票。在大市值组合中,反向操作——做多“范围”值较低的股票,做空“范围”值较高的股票。每月再平衡。

策略合理性

已有研究表明,上个月的每日最大和最小收益值对股票层面的后续收益有显著的预测能力。作者通过将它们结合成一个单一指标“收益范围”,发现这一指标与标准差高度相关。因此,收益范围可以作为总波动率的代理。他们发现,做多高收益范围的小盘股并做空低收益范围的小盘股,同时对大盘股反向操作的策略是有利可图的,而这一效应主要由小市值指数驱动。

论文来源

Return range and the cross-section of expected index returns in international stock markets (September, 2020) [点击浏览原文]

<摘要>

本研究首次探讨了收益范围与未来收益之间的横截面关系。我们发现,收益范围可以作为总波动率的非常实用的度量工具,因为它与标准差高度相关且具有很强的预测能力。范围、标准差和特质波动率与小市值指数的未来收益在横截面上有关,而收益价格比和净股票发行量分别预测中盘和大盘指数的回报。最大和最小收益效应以及动量效应在所有规模的指数回报中普遍存在,但在小市值指数中更为强烈。

回测表现

年化收益率61.4%
波动率40.17%
Beta-0.481
夏普比率1.53
索提诺比率-0.076
最大回撤N/A
胜率53%

完整python代码

from AlgorithmImports import *
from functools import reduce
#endregion

class ReturnRangePredictsStockReturns(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2010, 1, 1)
        self.SetCash(100000)
        
        self.period:int = 22        # need n daily prices
        self.cap_quantile:int = 5
        self.range_quantile:int = 5
        self.leverage:int = 5
        self.min_share_price:float = 5.

        self.prices:Dict[Symbol, RollingWindow] = {}
        self.weight:Dict[Symbol, float] = {}

        self.countries_ISO:List[str] = [
            'AUS', 'AUT', 'BEL', 'CAN', 'DNK', 'FIN', 'FRA', 'DEU', 'GRC', 'HKG', 'IRL', 'ITA',
            'JPN', 'NLD', 'NZL', 'NOR', 'PRT', 'SGP', 'ESP', 'SWE', 'CHE', 'GBR', 'USA', 'ARG',
            'BRA', 'CHL', 'CHN', 'IND', 'KOR', 'MYS', 'MEX', 'PHL', 'POL', 'ZAF', 'TWN', 'THA', 'TUR'
        ]

        self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol

        self.fundamental_count:int = 3000
        self.fundamental_sorting_key = lambda x: x.MarketCap

        self.selection_flag:bool = False
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0
        self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), 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]) -> None:
        # daily update stock prices
        for stock in fundamental:
            symbol:Symbol = stock.Symbol

            if symbol in self.prices:
                self.prices[symbol].update(stock.AdjustedPrice)

        if not self.selection_flag:
            return Universe.Unchanged

        selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and \
            x.AdjustedPrice >= self.min_share_price and x.CompanyReference.BusinessCountryID in self.countries_ISO]
        if len(selected) > self.fundamental_count:
            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
            
        stocks_bucket:Dict[str, List[Fundamental]] = {}

        for stock in selected:
            country_ISO:str = stock.CompanyReference.BusinessCountryID 
            sector:str = str(stock.AssetClassification.MorningstarSectorCode)
            bucket_identificator:str = country_ISO + '+' + sector

            if bucket_identificator not in stocks_bucket:
                stocks_bucket[bucket_identificator] = []

            stocks_bucket[bucket_identificator].append(stock)

        # make sure there are enough buckets
        if len(stocks_bucket) < self.cap_quantile: return Universe.Unchanged

        quantile:int = int(len(stocks_bucket) / self.cap_quantile)
        stocks_bucket:List[List[Symbol]] = list(stocks_bucket.values())

        sorted_by_bucket_cap:List[List[Symbol]] = sorted(stocks_bucket,
            key=lambda stocks: np.average([stock.MarketCap for stock in stocks]))

        small_cap:List[Symbol] = reduce(lambda x,y: x + y, sorted_by_bucket_cap[:quantile])
        large_cap:List[Symbol] = reduce(lambda x,y: x + y, sorted_by_bucket_cap[-quantile:])

        # SelectLowAndHighRangeStocks returns two empty list, if there aren't enough stocks for range selection
        small_cap_low_range, small_cap_high_range = self.SelectLowAndHighRangeStocks(small_cap)
        large_cap_low_range, large_cap_high_range = self.SelectLowAndHighRangeStocks(large_cap)

        # in the large-cap portfolio do the converse – long low ‘range’ stocks, short high ‘range’ stocks
        # in the small-cap portfolio, long the stocks with high ‘range’ value, short the stocks with low ‘range’ value
        long_part:List[Symbol] = small_cap_high_range + large_cap_low_range
        short_part:List[Symbol] = small_cap_low_range + large_cap_high_range

        # calc weights for each portfolio part
        for i, portfolio in enumerate([long_part, short_part]):
            for symbol in portfolio:
                self.weight[symbol] = ((-1) ** i) / len(portfolio)

        return list(self.weight.keys())
        
    def OnData(self, data: Slice) -> None:
        # rebalance monthly
        if not self.selection_flag:
            return
        self.selection_flag = False
        
        # trade execution
        portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
        self.SetHoldings(portfolio, True)

        self.weight.clear()

    def SelectLowAndHighRangeStocks(self, stocks:List) -> (List, List):
        stocks_ranges:Dict[Symbol, float] = {}

        for stock in stocks:
            symbol:Symbol = stock.Symbol

            if symbol not in self.prices:
                history = self.History(symbol, self.period, Resolution.Daily)
                if history.empty or len(history) < self.period:
                    continue
                
                # init stock's RollingWindow
                self.prices[symbol] = SymbolData(self.period)

                closes:List[float] = list(history.loc[symbol, 'close'])

                for close in closes:
                    self.prices[symbol].update(close)

            range_value:float = self.prices[symbol].get_range()

            if range_value != 0:
                stocks_ranges[symbol] = range_value

        # return empty lists in case of not enough stocks with range value
        if len(stocks_ranges) < self.range_quantile:
            return [], []

        quantile:int = int(len(stocks_ranges) / self.range_quantile)
        sorted_by_range:List[Symbol] = [x[0] for x in sorted(stocks_ranges.items(), key=lambda item: item[1])]

        low_range:List[Symbol] = sorted_by_range[:quantile]
        high_range:List[Symbol] = sorted_by_range[-quantile:]

        return low_range, high_range

    def Selection(self) -> None:
        self.selection_flag = True

class SymbolData:
    def __init__(self, period: int):
        self.prices:RollingWindow = RollingWindow[float](period)

    def update(self, price: float) -> None:
        self.prices.Add(price)

    def get_range(self) -> float:
        daily_prices:np.array = np.array([x for x in self.prices])
        daily_returns:List[float] = list((daily_prices[:-1] - daily_prices[1:]) / daily_prices[1:])

        daily_max_return:float = max(daily_returns)
        daily_min_return:float = min(daily_returns)

        return daily_max_return - daily_min_return

    def is_ready(self) -> bool:
        return self.prices.IsReady

# Custom fee model
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))

Leave a Reply

Discover more from Quant Buffet

Subscribe now to keep reading and get access to the full archive.

Continue reading