“该策略做多12个月回报为正且不在最低季节性十分位数内的商品，做空回报为负且不在最高季节性五分之一位数内的商品，每月重新平衡。”
资产类别: 差价合约、期货 | 地区: 全球 | 周期: 每月 | 市场: 大宗商品 | 关键词: 季节性、趋势跟踪
I. 策略概要
投资范围包括具有可用每日近月期货价格数据的商品。投资者做多过去12个月回报为正且不在最低季节性十分位数内的商品，做空过去12个月回报为负且不在最高季节性五分之一位数内的商品。季节性基于10年的回顾期，不包括过去一年的同月回报以避免相关性。投资组合每月重新平衡，资产等权重。该策略结合了动量和季节性信号，以识别商品期货市场的机会。
II. 策略合理性
商品中的季节性效应是由自然需求模式驱动的，例如冬季取暖能源需求增加或农产品在收获期前后的价格波动。趋势跟踪策略利用投资者的行为偏差，包括羊群效应。通过结合这些策略，如果交易成本保持在一定阈值以下，预计整体表现将得到改善。该策略旨在利用可预测的季节性价格波动，同时受益于动量和市场情绪，最终提高回报。
III. 来源论文
Multi-Asset Seasonality and Trend-Following Strategies [点击查看论文]
- 尼克·巴尔塔斯，帝国理工商学院；高盛国际
<摘要>
本文研究了各种资产类别中的季节性模式。我们发现，一种买入过去平均回报（最多十年）在相同日历月中最高的资产，并卖出过去平均回报在相同日历月中最低的资产的策略，在商品和股票指数范围内获得了统计和经济上的显著溢价。由于策略的高换手率和相关成本，直接利用这些溢价在实践中似乎很困难。因此，我们提出了一种将季节性信号积极纳入趋势跟踪策略的方法，即在相应的季节性信号另有说明时，关闭多头和空头头寸。季节性调整后的趋势跟踪策略对商品和股票指数的原始策略进行了显著改进。增加的换手率可能会影响业绩提升，但流动性期货合约相对较低的交易成本以及优化头寸平滑的方法修正可以使这种改进变得真实。
IV. 回测表现
| 年化回报 | 7.52% |
| 波动率 | 9.32% |
| β值 | -0.055 |
| 夏普比率 | 0.38 |
| 索提诺比率 | -0.05 |
| 最大回撤 | N/A |
| 胜率 | 56% |
V. 完整的 Python 代码
import pandas as pd
from AlgorithmImports import *
import numpy as np
class SeasonalityWithinTrendFollowing(QCAlgorithm):
def Initialize(self):
self.SetStartDate(1990, 1, 1)
self.SetCash(100000)
self.symbols = ["CME_S1", # Soybean Futures, Continuous Contract
"CME_W1", # Wheat Futures, Continuous Contract
"CME_SM1", # Soybean Meal Futures, Continuous Contract
"CME_BO1", # Soybean Oil Futures, Continuous Contract
"CME_C1", # Corn Futures, Continuous Contract
"CME_O1", # Oats Futures, Continuous Contract
"CME_LC1", # Live Cattle Futures, Continuous Contract
"CME_FC1", # Feeder Cattle Futures, Continuous Contract
"CME_LN1", # Lean Hog Futures, Continuous Contract
"CME_GC1", # Gold Futures, Continuous Contract
"CME_SI1", # Silver Futures, Continuous Contract
"CME_PL1", # Platinum Futures, Continuous Contract
"CME_CL1", # Crude Oil Futures, Continuous Contract
"CME_HG1", # Copper Futures, Continuous Contract
"CME_LB1", # Random Length Lumber Futures, Continuous Contract
# "CME_NG1", # Natural Gas (Henry Hub) Physical Futures, Continuous Contract
"CME_PA1", # Palladium Futures, Continuous Contract
"CME_RR1", # Rough Rice Futures, Continuous Contract
"ICE_RS1", # Canola Futures, Continuous Contract
"ICE_GO1", # Gas Oil Futures, Continuous Contract
"CME_RB2", # Gasoline Futures, Continuous Contract
"CME_KW2", # Wheat Kansas, Continuous Contract
"ICE_WT1", # WTI Crude Futures, Continuous Contract
"ICE_CC1", # Cocoa Futures, Continuous Contract
"ICE_CT1", # Cotton No. 2 Futures, Continuous Contract
"ICE_KC1", # Coffee C Futures, Continuous Contract
"ICE_O1", # Heating Oil Futures, Continuous Contract
"ICE_OJ1", # Orange Juice Futures, Continuous Contract
"ICE_SB1" # Sugar No. 11 Futures, Continuous Contract
]
self.period = 12 * 21
self.quantile = 5
self.SetWarmup(self.period)
self.data = {}
for symbol in self.symbols:
data = self.AddData(QuantpediaFutures, symbol, Resolution.Daily)
data.SetLeverage(5)
data.SetFeeModel(CustomFeeModel())
self.data[symbol] = SymbolData(self.period, self)
self.Schedule.On(self.DateRules.MonthStart(self.symbols[0]), self.TimeRules.At(0, 0), self.Rebalance)
def OnData(self, data):
# Update RollingWindows in SymbolData object
for symbol in self.symbols:
if symbol in data and data[symbol]:
price = data[symbol].Value
self.data[symbol].update(price,self.Time.date())
def Rebalance(self):
if self.IsWarmingUp: return
month = self.Time.month
# Thanks to this condition we make sure,
# that montly return is stored under right month in SymbolData.months dictionary
if month == 1:
month = 12
else:
month -= 1
positive_return = [] # Store futures with positive yearly return
negative_return = [] # Store futures with negative yearly return
for symbol in self.symbols:
# If prices for whole year are ready, we can calculate yearly return
if self.data[symbol].is_ready():
# Firstly calculate, then store current monthly return in proper list
self.MonthlyReturnStoring(symbol, month)
# Based on yearly return store future symbol in proper list
if self.data[symbol].yearly_return() > 0:
positive_return.append(symbol)
else:
negative_return.append(symbol)
elif self.data[symbol].are_monthly_prices_ready():
# Firstly calculate, then store current monthly return in proper list
self.MonthlyReturnStoring(symbol, month)
# Seasonality sorting
return_avg = {}
custom_data_last_update_date: Dict[Symbol, datetime.date] = QuantpediaFutures.get_last_update_date()
for symbol in self.symbols:
if self.Securities[symbol].GetLastData() and self.Time.date() < custom_data_last_update_date[symbol]:
# Check if there is enough data for current symbol of future
if self.data[symbol].are_monthly_returns_ready(month):
# Calculate average from yield returns in the same calendar month in history.
# Past year's same-month return is to be excluded from the seasonality calculation
return_avg[symbol] = self.data[symbol].get_avg_of_yield_returns(month)
long = []
short = []
# Prorgam will not continue if there aren't enough future symbols for quintile selection
if len(return_avg) >= self.quantile:
# Create top and bottom list based on avg sort
quintile = int(len(return_avg) / self.quantile)
sorted_by_avg_return = [x[0] for x in sorted(return_avg.items(), key = lambda item: item[1], reverse = True)]
top = sorted_by_avg_return[:quintile]
bottom = sorted_by_avg_return[-quintile:]
# Create long and short portfolio based on strategy description
long = [x for x in positive_return if x not in bottom]
short = [x for x in negative_return if x not in top]
# Trade execution
targets: List[PortfolioTarget] = []
for i, portfolio in enumerate([long, short]):
for symbol in portfolio:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
def MonthlyReturnStoring(self, symbol, month):
# Calculate monthly return for current month
monthly_return = self.data[symbol].monthly_return()
# Store blank list under current monnth number, if it doesn't exists
# Thanks to this list we are making history of monthly returns for each year
if month not in self.data[symbol].months:
self.data[symbol].months[month] = []
# Add current monthly return into list of all montly returns in this month
self.data[symbol].months[month].append(monthly_return)
class SymbolData():
def __init__(self, period, algorithm):
self.monthly_prices = RollingWindow[float](21) # Storing prices for monthly return calculation
self.prices = RollingWindow[float](period) # Storing daily prices of futures
self.months = {} # Storing monthly returns in list for each month separately
self.last_update_date = None
self.algorithm = algorithm
def update(self, price, date):
self.monthly_prices.Add(price)
self.prices.Add(price)
self.last_update_date = date
def is_ready(self):
return self.prices.IsReady and ((self.algorithm.Time.date() - self.last_update_date).days <= 4)
def are_monthly_prices_ready(self):
return self.monthly_prices.IsReady
def are_monthly_returns_ready(self, month):
# Check if there are at least 6 monhtly returns in specific month
if month in self.months and len(self.months[month]) > 5:
return True
else:
return False
def monthly_return(self):
prices = [x for x in self.monthly_prices]
return (prices[0] - prices[-1]) / prices[-1]
def yearly_return(self):
prices = [x for x in self.prices]
return (prices[0] - prices[-1]) / prices[-1]
def get_avg_of_yield_returns(self, month):
# Get all monthly returns for specific month
monthly_returns = self.months[month]
# We get average from max 10 values of monthly returns
if len(monthly_returns) > 10:
monthly_returns = monthly_returns[-11:]
# The past year's same-month return is to be excluded from the seasonality calculation to avoid any correlation between the two signals
monthly_returns = monthly_returns[:-1]
# Return average of monthly returns in specific month
return np.mean(monthly_returns)
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaFutures(PythonData):
_last_update_date:Dict[Symbol, datetime.date] = {}
@staticmethod
def get_last_update_date() -> Dict[Symbol, datetime.date]:
return QuantpediaFutures._last_update_date
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaFutures()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split = line.split(';')
data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
data['back_adjusted'] = float(split[1])
data['spliced'] = float(split[2])
data.Value = float(split[1])
if config.Symbol.Value not in QuantpediaFutures._last_update_date:
QuantpediaFutures._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
if data.Time.date() > QuantpediaFutures._last_update_date[config.Symbol.Value]:
QuantpediaFutures._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"))