“该策略使用来自五种异常现象的标准化信号交易40种商品期货，根据综合信号构建等权重、多空投资组合，每月重新平衡以实现系统性多元化。”
资产类别: 期货 | 地区: 中国 | 周期: 每月 | 市场: 大宗商品 | 关键词: 期货
I. 策略概要
该策略交易40种商品期货，使用五种信号：展期收益、动量、价值、投机压力和偏度。信号进行横截面标准化（减去均值并除以标准差），并在异常现象之间等权重。综合投资组合分配计算为标准化信号和等权重的矩阵乘积，从而得出每种商品的平均信号。基于这些综合信号构建多空风险溢价投资组合，并每月重新平衡，利用多种异常现象的综合洞察力进行多元化和系统化的商品交易。
II. 策略合理性
风格整合通过结合方法论和使用过滤后的交易信号的结果来提升绩效。该策略采用三种投资组合分配方法：等权重、基于优化的和波动率择时。研究（例如，Fernandez-Perez et al., 2019）表明，等权重投资组合可确保稳健的样本外绩效。结果在跨度测试后仍然可靠，即使在仅多头设置中也成立，这使得该策略对小众投资者具有吸引力。这种方法在不同的分配方法中表现出弹性和适应性，增强了其在实际投资场景中的适用性和有效性。
III. 来源论文
The strategic allocation to style-integrated portfolios of commodity futures [点击查看论文]
- Rad, Hossein and Low, Rand Kwong Yew and Miffre, Joelle and Faff, Robert W.，昆士兰大学商学院，昆士兰大学；邦德大学，南特高等商学院，布达佩斯考文纽斯大学；昆士兰大学；邦德大学
<摘要>
我们通过将商品期货的多元化收益与风格整合文献相结合，为该领域做出了贡献。我们的工作通过一个整合了对商品期货定价重要的风格的多空投资组合，增强了投资者传统的资产组合。与迄今为止考虑的替代方案相比，将商品风格整合投资组合作为投资者战略组合的一部分，被发现可以提高样本外表现并降低崩溃风险。该结论适用于传统资产组合、投资组合分配方法、整合策略和子时期。尽管较低，但风格整合的多元化收益在仅多头设置中也持续存在。
IV. 回测表现
| 年化回报 | 7.71% |
| 波动率 | 8% |
| β值 | -0.015 |
| 夏普比率 | 0.96 |
| 索提诺比率 | -0.07 |
| 最大回撤 | -14% |
| 胜率 | 51% |
V. 完整的 Python 代码
from AlgorithmImports import *
import data_tools
#endregion
class StyleIntegratedPortfoliosofCommodityFutures(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.tickers:dict[str, str] = {
"CME_S1" : Futures.Grains.Soybeans, # Soybean Futures, Continuous Contract
"CME_W1" : Futures.Grains.Wheat, # Wheat Futures, Continuous Contract
"CME_SM1" : Futures.Grains.SoybeanMeal, # Soybean Meal Futures, Continuous Contract
"CME_BO1" : Futures.Grains.SoybeanOil, # Soybean Oil Futures, Continuous Contract
"CME_C1" : Futures.Grains.Corn, # Corn Futures, Continuous Contract
"CME_O1" : Futures.Grains.Oats, # Oats Futures, Continuous Contract
"CME_LC1" : Futures.Meats.LiveCattle, # Live Cattle Futures, Continuous Contract
"CME_FC1" : Futures.Meats.FeederCattle, # Feeder Cattle Futures, Continuous Contract
"CME_LN1" : Futures.Meats.LeanHogs, # Lean Hog Futures, Continuous Contract
"CME_GC1" : Futures.Metals.Gold, # Gold Futures, Continuous Contract
"CME_SI1" : Futures.Metals.Silver, # Silver Futures, Continuous Contract
"CME_PL1" : Futures.Metals.Platinum, # Platinum Futures, Continuous Contract
"CME_HG1" : Futures.Metals.Copper, # Copper Futures, Continuous Contract
"CME_LB1" : Futures.Forestry.RandomLengthLumber, # Random Length Lumber Futures, Continuous Contract
# "CME_NG1", # Natural Gas (Henry Hub) Physical Futures, Continuous Contract #1
"CME_PA1" : Futures.Metals.Palladium, # Palladium Futures, Continuous Contract
# "CME_RR1", # Rough Rice Futures, Continuous Contract #1
# "CME_CU1", # Chicago Ethanol (Platts) Futures
"CME_DA1" : Futures.Dairy.ClassIIIMilk, # Class III Milk Futures
"ICE_CC1" : Futures.Softs.Cocoa, # Cocoa Futures, Continuous Contract
"ICE_CT1" : Futures.Softs.Cotton2, # Cotton No. 2 Futures, Continuous Contract #1
"ICE_KC1": Futures.Softs.Coffee, # Coffee C Futures, Continuous Contract #1
"ICE_O1" : Futures.Energies.HeatingOil, # Heating Oil Futures, Continuous Contract
"ICE_OJ1": Futures.Softs.OrangeJuice, # Orange Juice Futures, Continuous Contract #1
"ICE_SB1" : Futures.Softs.Sugar11CME, # Sugar No. 11 Futures, Continuous Contract
}
self.k:int = 5 # k signals
self.min_expiration_days:int = 0
self.max_expiration_days:int = 360
self.leverage:int = 5
self.COT_data_period:int = 52 # n weeks
self.momentum_period:int = 12 * 21 # m days
self.value_period:int = 5.5 * 12 * 21 # l days
self.data:dict[Symbol, data_tools.SymbolData] = {}
self.futures_data:dict[str, data_tools.FuturesData] = {}
for qp_ticker, qc_ticker in self.tickers.items():
# subscribe Quantpedia data
security:Security = self.AddData(data_tools.QuantpediaFutures, qp_ticker, Resolution.Daily)
security.SetFeeModel(data_tools.CustomFeeModel())
security.SetLeverage(self.leverage)
qp_symbol:Symbol = security.Symbol
# subscribe COT data
COT_ticker:str = 'Q' + qp_ticker.split('_')[1][:-1]
COT_symbol:Symbol = self.AddData(data_tools.CommitmentsOfTraders, COT_ticker, Resolution.Daily).Symbol
# QC futures
future:Future = self.AddFuture(qc_ticker, Resolution.Daily, dataNormalizationMode=DataNormalizationMode.Raw)
future.SetFilter(timedelta(days=self.min_expiration_days), timedelta(days=self.max_expiration_days))
future_ticker:str = future.Symbol.Value
self.futures_data[future_ticker] = data_tools.FuturesData()
self.data[qp_symbol] = data_tools.SymbolData(COT_symbol, future_ticker,
self.COT_data_period, self.value_period)
self.recent_month:int = -1
self.selection_flag:bool = False
self.settings.daily_precise_end_time = False
self.settings.minimum_order_margin_portfolio_percentage = 0.
def FindAndUpdateContracts(self, futures_chain, ticker:str) -> None:
near_contract:FuturesContract = None
dist_contract:FuturesContract = None
if ticker in futures_chain:
contracts:list[:FuturesContract] = [contract for contract in futures_chain[ticker] if contract.Expiry.date() > self.Time.date()]
if len(contracts) >= 2:
contracts:list[:FuturesContract] = sorted(contracts, key=lambda x: x.Expiry, reverse=False)
near_contract = contracts[0]
dist_contract = contracts[1]
self.futures_data[ticker].update_contracts(near_contract, dist_contract)
def OnData(self, data):
curr_date:datetime.date = self.Time.date()
# rebalance monthly
if self.recent_month != curr_date.month:
self.selection_flag = True
self.recent_month = curr_date.month
momentum:dict[Symbol, float] = {}
value:dict[Symbol, float] = {}
skewness:dict[Symbol, float] = {}
roll_yield:dict[Symbol, float] = {}
speculative_pressure:dict[Symbol, float] = {}
# daily update qc future data
if data.FutureChains.Count > 0:
for ticker, future_obj in self.futures_data.items():
# check if near contract is expired or is not initialized
if not future_obj.is_initialized() or \
(future_obj.is_initialized() and future_obj.near_contract.Expiry.date() == curr_date):
self.FindAndUpdateContracts(data.FutureChains, ticker)
# update QC futures rolling return
if future_obj.is_initialized():
near_c:FuturesContract = future_obj.near_contract
dist_c:FuturesContract = future_obj.distant_contract
if near_c.Symbol in data and data[near_c.Symbol] and dist_c.Symbol in data and data[dist_c.Symbol]:
near_price:float = data[near_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
dist_price:float = data[dist_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
if near_price != 0 and dist_price != 0:
roll_yield:float = np.log(near_price) - np.log(dist_price)
future_obj.set_roll_yield(roll_yield)
for qp_symbol, symbol_obj in self.data.items():
# update Quantpedia prices
if qp_symbol in data and data[qp_symbol]:
# firstly make sure Quantpedia data still comming
if self.securities[qp_symbol].get_last_data() and self.time.date() > data_tools.LastDateHandler.get_last_update_date()[qp_symbol]:
symbol_obj_reset_prices()
price:float = data[qp_symbol].Value
symbol_obj.update_prices(price)
# retreive COT data (weekly)
COT_symbol:Symbol = symbol_obj.COT_symbol
if self.securities[COT_symbol].get_last_data() and self.time.date() > data_tools.LastDateHandler.get_last_update_date()[COT_symbol]:
symbol_obj.reset_speculative_net()
if COT_symbol in data and data[COT_symbol]:
speculator_long_count:float = data[COT_symbol].Large_Speculator_Long
speculator_short_count:float = data[COT_symbol].Large_Speculator_Short
# calculate net position of large speculators relative to their total position over previous week
speculative_net:float = 0 if speculator_long_count == 0 or speculator_short_count == 0 else \
self.SpeculativeNetCal(speculator_long_count, speculator_short_count)
symbol_obj.update_speculative_net(speculative_net)
future_ticker:str = symbol_obj.future_ticker
# prices for momentum and value and skewness calulation is ready and data for average speculative net position is ready
if self.selection_flag and symbol_obj.prices_ready() and symbol_obj.speculative_net_ready() and self.futures_data[future_ticker].is_ready():
speculative_pressure[qp_symbol] = symbol_obj.get_speculative_pressure()
momentum[qp_symbol] = symbol_obj.get_momentum(self.momentum_period)
value[qp_symbol] = symbol_obj.get_value(self.momentum_period)
skewness[qp_symbol] = symbol_obj.get_skewness(self.momentum_period)
roll_yield[qp_symbol] = self.futures_data[future_ticker].get_roll_yield()
# COT weekly data arrived and monthly selection should be made
if self.selection_flag and len(momentum) != 0:
self.selection_flag = False
# normalize every signal
speculative_pressure_values = list(self.NormalizeSignalDict(speculative_pressure).values())
momentum_values = list(self.NormalizeSignalDict(momentum).values())
value_values = list(self.NormalizeSignalDict(value).values())
skewness_values = list(self.NormalizeSignalDict(skewness).values())
roll_yield_values = list(self.NormalizeSignalDict(roll_yield).values())
# calculate the integrated portfolio allocation
signal_matrix:np.ndarray = np.array([speculative_pressure_values, momentum_values, value_values, skewness_values, roll_yield_values]).T
weight_matrix:np.ndarray = np.array([[1 / self.k] * self.k]).T
# matrix multiplication
allocation = np.dot(signal_matrix, weight_matrix).T[0]
# normalized allocation
alloc_sum = sum([np.abs(x) for x in allocation])
allocation_normalized = [x/alloc_sum for x in allocation]
# pair symbols with its portfolio allocation
alloc_with_symbol = { symbol : allocation_normalized[i] for i,symbol in enumerate(momentum.keys()) }
# trade execution
portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in alloc_with_symbol.items() if data.contains_key(symbol) and data[symbol]]
self.SetHoldings(portfolio, True)
def SpeculativeNetCal(self, speculator_long_count:float, speculator_short_count:float) -> float:
return (speculator_long_count - speculator_short_count) / (speculator_long_count + speculator_short_count)
def NormalizeSignalDict(self, signal_dict:dict) -> dict:
signal_values:list = list(signal_dict.values())
signal_mean:float = np.mean(signal_values)
signal_std:float = np.std(signal_values)
# new adjusted signal dict
result:dict[Symbol, float] = { symbol : (signal-signal_mean)/signal_std for symbol, signal in signal_dict.items() }
return result