“该策略使用CFTC持仓报告（COT）数据交易纽约证券交易所、美国证券交易所和纳斯达克的商品相关股票，根据11种商品的交易者头寸增长信号，每周构建多空投资组合。”
资产类别: 股票 | 地区: 美国 | 周期: 每周 | 市场: 股票 | 关键词: 股票
I. 策略概要
该策略利用商品期货交易委员会（CFTC）的交易商持仓报告 (COT) 数据，交易与11种商品（例如，金属、能源和软商品）相关的纽约证券交易所、美国证券交易所和纳斯达克普通股。对于每个与四位SIC代码相关联的商品，识别出具有匹配代码的上市公司，从而形成11个商品特定投资组合。
分类COT（DCOT）报告提供每周交易商持仓数据。多头比例增长指标计算为管理资金（MM）交易商多头头寸增长率除以总头寸（MMlong/(MMlong+MMshort+2MMspreading)）。股票每周根据滞后信号变量被分类为多空投资组合。对于具有正信号增长率的股票采取多头头寸，而空头头寸则针对负信号增长率的股票。
该策略使用等权重投资组合，但也允许其他加权方案，例如价值加权或度加权组合。投资组合每周根据更新的COT信号重新平衡，利用交易商头寸趋势来预测商品相关股票的表现。
II. 策略合理性
DCOT报告显示，管理资金（MM）交易者在商品期货中的头寸为股票回报提供了有价值的预测信号。MM交易者以其投机策略、杠杆和市场洞察力而闻名，比生产商（PM）更能有效地预测股票回报的横截面，后者头寸缺乏预测能力。MM头寸反映了对未来商品价格的看法，这与生产商股票的价格相关。该策略从MM多头/空头头寸中产生了经济和统计上显著的阿尔法。这些结果在各种衡量标准、加权方案、时间滞后和商业周期中均保持稳健。此外，多变量回归证实，MM信号独立于市场敞口、规模或动量等传统预测因子，能够预测股票回报。
III. 来源论文
Is There Smart Money? How Information in the Futures Market Is Priced into the Cross-Section of Stock Returns with Delay [点击查看论文]
- Steven Wei Ho (何伟) 和 Alexandre R. Lauwers. 内华达大学拉斯维加斯分校；哥伦比亚大学文理研究生院经济学系。日内瓦大学 – 日内瓦高级国际关系及发展学院 (IHEID)
<摘要>
我们记录了一个新的经验现象：商品期货市场中老练的投机者——资金管理人（MM）的头寸（由CFTC分类交易商持仓（DCOT）报告披露）可以预测下一周商品生产商股票的横截面回报。我们采用横截面方法，包括单次排序、詹森阿尔法分析、双次排序和Fama-Macbeth回归，以证实其可预测性结果。在信息不对称程度较高的公司中，这种结果更为显著，信息不对称程度通过分析师分歧和历史波动率来衡量。因此，我们为有关成本高昂的信息处理导致市场分割和信息在资产市场中逐渐扩散的文献提供了更多的经验证据，正如领先-滞后关系所证明的那样。
IV. 回测表现
| 年化回报 | 19.21% |
| 波动率 | 28.57% |
| β值 | -0.047 |
| 夏普比率 | 0.67 |
| 索提诺比率 | 0.042 |
| 最大回撤 | N/A |
| 胜率 | 52% |
V. 完整的 Python 代码
from AlgorithmImports import *
from functools import reduce
from typing import List, Dict, Tuple
from numpy import isnan
class CrossSectionOfStockReturnsPredictedByCommitmentOfTradersInformation(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2005, 1, 1)
self.SetCash(100000)
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.min_share_price:int = 5
self.leverage:int = 5
self.SIC_stocks = {} # storing list of stocks symbols keyed by SIC code
self.COT_tickers_SICs:List[Tuple[List]] = [
(['QHG'], [1020, 1021, 3331]), # Copper
(['QGC'], [1040, 1041]), # Gold
(['QSI'], [1044]), # Silver
(['QLB'], [2400]), # Lumber
(['QGO', 'QCL'], [1310, 1311]), # Gas, Oil
(['QPL', 'QPA'], [3449, 3491, 3492, 3493, 3494, 3495, 3496, 3497, 3498, 3499]), # Platinum, Palladium
]
# create 1D list from SIC codes
self.SIC_universe:List[int] = map(lambda x: x[1], self.COT_tickers_SICs)
self.SIC_universe:List[int] = reduce(lambda x,y: x+y , self.SIC_universe)
self.last_long_prop:Dict[str, None] = {
'QHG': None,
'QGC': None,
'QSI': None,
'QLB': None,
'QGO': None,
'QCL': None,
'QPL': None,
'QPA': None
}
# subscribe to COT data
for cot_ticker, _ in self.last_long_prop.items():
data = self.AddData(CommitmentsOfTraders, cot_ticker, Resolution.Daily)
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.selection_flag:bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol, 0), 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]:
# selection on monthly basis
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
# filter all symbol of stocks
selected:List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price \
and not isnan(x.AssetClassification.SIC != 0) and (x.AssetClassification.SIC != 0) \
and (x.AssetClassification.SIC in self.SIC_universe) and x.SecurityReference.ExchangeId in self.exchange_codes
]
selected_symbols:List[Symbol] = []
# firstly clear stocks from old selection
self.SIC_stocks.clear()
# store relevant stocks symbols into their basket according to SIC code
for stock in selected:
symbol = stock.Symbol
SIC_code = stock.AssetClassification.SIC
# make sure list for stocks is initialized
if SIC_code not in self.SIC_stocks:
self.SIC_stocks[SIC_code] = []
# add stock's symbol to it's basket based on SIC code
self.SIC_stocks[SIC_code].append(symbol)
selected_symbols.append(symbol)
return selected_symbols
def OnData(self, data: Slice) -> None:
COT_data_last_update_date:Dict[Symbol, datetime.date] = CommitmentsOfTraders.get_last_update_date()
# storing tuples (SIC_list, long_proportion_growth_value)
long_proportion_growth:List[Tuple[List, float]] = []
rebalance_flag:bool = False
for COT_ticker_list, SIC_list in self.COT_tickers_SICs:
long_proportion_growth_values:List[float] = []
for COT_ticker in COT_ticker_list:
if self.Securities[COT_ticker].GetLastData() and self.Time.date() < COT_data_last_update_date[COT_ticker]:
if COT_ticker in data and data[COT_ticker]:
rebalance_flag = True
# retrieve needed values from data object
large_spec_long:int = data[COT_ticker].get_Item('LARGE_SPECULATOR_LONG')
large_spec_short:int = data[COT_ticker].get_Item('LARGE_SPECULATOR_SHORT')
if large_spec_long == 0 or large_spec_short == 0:
continue
if not self.last_long_prop[COT_ticker]:
value:float = large_spec_long / (large_spec_short + large_spec_long + 0)
self.last_long_prop[COT_ticker] = value
continue
curr_long_proportion:float = large_spec_long / (large_spec_short + large_spec_long + 0)
growth_value:float = (curr_long_proportion - self.last_long_prop[COT_ticker]) / self.last_long_prop[COT_ticker]
# append long proportion growth value for current COT data
long_proportion_growth_values.append(growth_value)
# update last long proporiton value
self.last_long_prop[COT_ticker] = curr_long_proportion
if len(long_proportion_growth_values) != 0:
# storing tuples (SIC_list, long_proportion_growth_value)
long_proportion_growth.append( (SIC_list, np.mean(long_proportion_growth_values)) )
# rebalance weekly
if len(long_proportion_growth) != 0 and rebalance_flag:
# long stocks with positive signal growth rates and short stocks with negative signal growth.
long, short = self.CreateLongShortPortfolio(long_proportion_growth)
# order execution
targets:List[PortfolioTarget] = []
for i, portfolio in enumerate([long, short]):
for symbol in portfolio:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
elif len(long_proportion_growth) == 0 and rebalance_flag:
self.Liquidate()
def CreateLongShortPortfolio(self, long_proportion_growth:Tuple):
long:List[Symbol] = []
short:List[Symbol] = []
# long stocks with positive signal growth rates and short stocks with negative signal growth.
for SIC_list, value in long_proportion_growth:
for SIC in SIC_list:
# make sure SIC code has stocks
if SIC not in self.SIC_stocks:
continue
if value > 0:
long += self.SIC_stocks[SIC]
else:
short += self.SIC_stocks[SIC]
return long, short
def Selection(self) -> None:
self.selection_flag = True
# Commitments of Traders data.
# NOTE: IMPORTANT: Data order must be ascending (datewise).
# Data source: https://commitmentsoftraders.org/cot-data/
# Data description: https://commitmentsoftraders.org/wp-content/uploads/Static/CoTData/file_key.html
class CommitmentsOfTraders(PythonData):
_last_update_date:Dict[Symbol, datetime.date] = {}
@staticmethod
def get_last_update_date() -> Dict[Symbol, datetime.date]:
return CommitmentsOfTraders._last_update_date
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/cot/{0}.PRN".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
# File example.
# DATE OPEN HIGH LOW CLOSE VOLUME OI
# ---- ---- ---- --- ----- ------ --
# DATE LARGE SPECULATOR COMMERCIAL HEDGER SMALL TRADER
# LONG SHORT LONG SHORT LONG SHORT
def Reader(self, config, line, date, isLiveMode):
data = CommitmentsOfTraders()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split = line.split(',')
# Prevent lookahead bias.
data.Time = datetime.strptime(split[0], "%Y%m%d") + timedelta(days=1)
data['LARGE_SPECULATOR_LONG'] = int(split[1])
data['LARGE_SPECULATOR_SHORT'] = int(split[2])
data['COMMERCIAL_HEDGER_LONG'] = int(split[3])
data['COMMERCIAL_HEDGER_SHORT'] = int(split[4])
data['SMALL_TRADER_LONG'] = int(split[5])
data['SMALL_TRADER_SHORT'] = int(split[6])
data.Value = int(split[1])
if config.Symbol.Value not in CommitmentsOfTraders._last_update_date:
CommitmentsOfTraders._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
if data.Time.date() > CommitmentsOfTraders._last_update_date[config.Symbol.Value]:
CommitmentsOfTraders._last_update_date[config.Symbol.Value] = data.Time.date()
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"))