中国的动量商品溢价

“该策略交易中国商品期货，根据12个月的回报形成多空投资组合，每月重新平衡，采用逐步展期，并排除低交易量合约以确保实际可投资性。”

资产类别: 期货 | 地区: 中国 | 周期: 每月 | 市场: 大宗商品 | 关键词: 动量、中国

I. 策略概要

该策略交易郑商所、大商所和上期所的商品期货，使用第三近月合约过去12个月的复合回报将期货分为两半。通过买入上半部分（回报最高）和卖出下半部分（回报最低）来构建多空投资组合，头寸等权重并每月重新平衡。采用逐步展期方法，持有第三近月合约直到到期前五天，并每天将20%的头寸展期到下一个第三近月合约。月交易量低于10,000手的商品被排除在外，以确保投资组合在实际世界中的可投资性。

II. 策略合理性

横截面动量策略涉及做多过去的赢家和做空过去的输家，这受到行为偏差（锚定和过度反应）和宏观经济风险（流动性、现货溢价和期货溢价周期）的驱动。尽管在发达市场有充分记录，但其在中国历史上封闭市场的应用引人入胜。为了避免非流动性合约，“逐步展期”持有第三近月合约，直到到期前五天，每天展期20%，提供120万人民币的容量，年回报率为8.99%。“动态展期”则转向未平仓合约量最高的合约，将容量提高到9.87亿人民币，但回报降至6.21%。探索了替代权重方案，以实现可定制的绩效和容量。

III. 来源论文

Investable Commodity Premia in China [点击查看论文]

• 罗伯特·J·比安奇、范华和张婷熙。格里菲斯大学；格里菲斯大学会计、金融与经济系。科廷大学

<摘要>

我们研究了中国商品风险溢价的可投资性。此前记录的标准动量、展期和基差动量因子由于中国期货曲线独特的流动性模式而无法投资。然而，动态展期和战略性投资组合权重显著提高了此类溢价的投资容量，而不会损害其统计和经济意义。同时，风格整合带来了增强的绩效和改进的机会集。此外，观察到的可投资溢价对执行滞后、止损、非流动性、子周期规范、季节性和交易成本均具有鲁棒性。它们还为投资者提供了投资组合多样化。最后，中国可投资商品溢价显示出与全球实际经济增长的强大预测能力。

IV. 回测表现

年化回报	8.99%
波动率	12.05%
β值	-0.023
夏普比率	0.74
索提诺比率	N/A
最大回撤	-32.21%
胜率	45%

V. 完整的 Python 代码

from AlgorithmImports import *
#endregion
class MomentumCommodityPremiainChina(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2017, 1, 1)
        self.SetCash(100000)
        
        # NOTE: QC max cap of 100 custom symbols added => 50 commodities x2 contracts
        self.symbols:list[str] = [
            # 'ER','ME','RO','S','TC','WS','WT',    # empty
            
            'CU', 'A', 'AG', 'AL', 'AP', 'AU',
            'B', 'BB', 'BU', 'C', 'CF', 'CS',
            'CY', 'FB', 'FG', 'FU', 'HC', 'I',
            'IC', 'IF', 'IH', 'J', 'JD', 'JM', 
            'JR', 'L', 'LR', 'M', 'MA', 'NI',
            'OI', 'P', 'PB', 'PM', 'PP', 'RB',
            'RI', 'RM', 'RS', 'RU',  'SF', 'SM',
            'SN', 'SR', 'T', 'TF', 'V',
            'WR', 'ZN', 'Y'
            #  'ZC', 'TA', 'WH'
        ]
        
        self.period:int = 12 * 21
        self.SetWarmup(self.period, Resolution.Daily)
        self.price_data:dict = {}
        self.latest_update_date:dict = {}   # latest price data arrival time
        self.quantile:int = 2
        for symbol in self.symbols:
            # futures data
            sym:str = symbol + '3'  # erd contract
            data = self.AddData(QuantpediaChineseFutures, sym, Resolution.Daily)
            data.SetLeverage(5)
            data.SetFeeModel(CustomFeeModel())
            self.price_data[sym] = RollingWindow[float](self.period)
            self.latest_update_date[sym] = None
                
        self.recent_month = -1
    
    def OnData(self, data):
        momentum:dict[Symbol, float] = {}
        # store daily prices
        for near_c_symbol, _ in self.price_data.items():
            # store price data
            if near_c_symbol in data and data[near_c_symbol] and data[near_c_symbol].Value != 0:
                self.price_data[near_c_symbol].Add(data[near_c_symbol].Value)  # 3rd contract price
                self.latest_update_date[near_c_symbol] = self.Time.date()
                if self.IsWarmingUp: continue
                # rebalance date
                if self.Time.month != self.recent_month:
                    # check data arrival time
                    if (self.Time.date() - self.latest_update_date[near_c_symbol]).days > 5:
                        continue
                    # price data for both contracts are ready
                    if self.price_data[near_c_symbol].IsReady:
                        # calculate momentum from forward ratio rolled contracts
                        near_momentum:float = self.price_data[near_c_symbol][0] / self.price_data[near_c_symbol][self.period-1] - 1
                        momentum[near_c_symbol] = near_momentum
        # monthly rebalance
        if self.Time.month == self.recent_month:
            return
        self.recent_month = self.Time.month
        long:list[Symbol] = []
        short:list[Symbol] = []
        if len(momentum) >= self.quantile:
            # sort by momentum
            sorted_by_momentum = sorted(momentum.items(), key=lambda x: x[1], reverse=True)
            quantile:int = int(len(momentum) / self.quantile)
            # buying (selling) the half with the highest (lowest) return
            long = [x[0] for x in sorted_by_momentum][:quantile]
            short = [x[0] for x in sorted_by_momentum][-quantile:]
        # trade execution
        invested = [x.Key for x in self.Portfolio if x.Value.Invested]
        for symbol in invested:
            if symbol not in long + short:
                self.Liquidate(symbol)
        for symbol in long:
            self.SetHoldings(symbol, 1 / len(long))
        
        for symbol in short:
            self.SetHoldings(symbol, -1 / len(short))
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaChineseFutures(PythonData):
    def GetSource(self, config, date, isLiveMode):
        return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/china/forward_ratio_rolled/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
    def Reader(self, config, line, date, isLiveMode):
        data = QuantpediaChineseFutures()
        data.Symbol = config.Symbol
        
        if not line[0].isdigit(): 
            return None
        split = line.split(';')
        
        data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
        data['close'] = float(split[1]) if split[1] != '' else 0 # unadjusted close
        data['adj_close'] = float(split[2]) if split[2] != '' else 0
        data['last_trade_month'] = int(split[3])
        data.Value = float(split[2]) if split[2] != '' else 0
        return data
# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))