按时间顺序的收益排序

“通过现金回报率（CRO）交易全球大盘股，做多最高CRO十分位，做空最低CRO十分位，使用价值加权投资组合，每月在49个国家重新平衡。”

资产类别: 股票 | 地区: 全球 | 周期: 每月 | 市场: 股票 | 关键词: 时间顺序

I. 策略概要

投资范围包括来自49个国家的全球大盘股。每月计算每日股票回报与月末剩余天数之间的皮尔逊相关系数（CRO）。低CRO值表示当前收益高而未来回报低，而高CRO值则预示着未来回报更高。股票根据CRO值分为十分位，做多最高十分位（最高CRO），做空最低十分位（最低CRO）。该策略采用价值加权，每月重新平衡，利用CRO作为未来股票表现的预测指标。

II. 策略合理性

该策略的功能根植于投资者的近因偏差，即对近期市场信息的高度关注导致错误定价。当前收益较高而过去收益较低的股票被高估，而低估资产则出现相反情况。套利者会随着时间的推移纠正这些错误定价。这种异常在市场崩盘期间尤其强烈，提供更高的回报作为风险溢价。这项研究独特地使用了时间顺序回报排序（CRO），并且是第一个在全球49个国家对其进行分析的研究。稳健性检验表明，这种现象适用于大公司和小型公司，跨越多个行业，并且在考察期间保持一致。

III. 来源论文

Chronological Return Ordering and the Cross-Section of International Stock Returns [点击查看论文]

• Cakici、Zaremba。福特汉姆大学。蒙彼利埃商学院；波兹南经济与商业大学；开普敦大学（UCT）

<摘要>

投资者通常只关注近期信息，低估了遥远过去数据的相关性。因此，历史回报的顺序可靠地预测了未来股票在横截面上的表现。我们使用来自49个国家的数据，全面考察了国际市场中的这种异常现象。根据时间顺序回报排序的全球股票高低十分位之间的平均回报差异每月达到0.91%。这种效应在最大的公司中表现出明显的稳健性，但在国际上表现出显著的异质性。这种错误定价在个人主义程度高和股东保护强的国家普遍存在。此外，它集中在市场下跌和过度波动时期之后。

IV. 回测表现

年化回报	12.02%
波动率	12.69%
β值	0.128
夏普比率	0.94
索提诺比率	0.113
最大回撤	N/A
胜率	48%

V. 完整的 Python 代码

import numpy as np
from AlgorithmImports import *
class ChronologicalReturnOrdering(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2000, 1, 1)
        self.SetCash(100000)
        
        self.period:int = 21
        self.quantile:int = 10
        self.leverage:int = 5
        market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        # daily price data
        self.prices:Dict[Symbol, RollingWindow] = {}
        
        self.long:List[Symbol] = []
        self.short:List[Symbol] = []
        self.fundamental_count:int = 500
        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.MonthStart(market), self.TimeRules.AfterMarketOpen(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 the rolling window every day
        for stock in fundamental:
            symbol:Symbol = stock.Symbol
            # store daily price
            if symbol in self.prices:
                self.prices[symbol].Add(stock.AdjustedPrice)
        if not self.selection_flag:
            return Universe.Unchanged
        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]]
        # warmup price rolling windows
        for stock in selected:
            symbol:Symbol = stock.Symbol
            if symbol in self.prices:
                continue
            self.prices[symbol] = RollingWindow[float](self.period)
            history = self.History(symbol, self.period, Resolution.Daily)
            if history.empty:
                self.Log(f"Not enough data for {symbol} yet.")
                continue
            closes = history.loc[symbol].close
            for time, close in closes.items():
                self.prices[symbol].Add(close)
            
        # cro calc - Pearson product-moment correlation coefficients
        daily_returns:Dict[Symbol, pd.Series] = { x.Symbol : pd.Series([p for p in self.prices[x.Symbol]][::-1]).cumprod().dropna() for x in selected if self.prices[x.Symbol].IsReady }
        CRO:Dict[Symbol, float] = { x : np.corrcoef(daily_returns[x].values, range(len(daily_returns[x])))[0,1] for x in daily_returns }
        
        # cro sorting
        if len(CRO) >= self.quantile:
            sorted_by_cro = sorted(CRO.items(), key = lambda x: x[1], reverse=True)
            quantile:int = int(len(sorted_by_cro) / self.quantile)
            self.long = [x[0] for x in sorted_by_cro[:quantile]]
            self.short = [x[0] for x in sorted_by_cro[-quantile:]]
        
        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:
        self.selection_flag = True
# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))