“该策略使用MACD指标作为趋势信号交易7种期货，并根据波动率阈值调整头寸，采用等权重配置和每日再平衡，以适应不断变化的市场条件。”
资产类别: 差价合约（CFDs）、期货 | 地区: 全球 | 周期:
每日 | 市场: 债券、大宗商品、货币、股票 | 关键词: 时间序列、动量、波动率
I. 策略概要
该策略交易7种期货（欧元兑美元、美国国债、标普500指数、欧元/美元、美元/日元、英镑/美元、铜），使用MACD指标识别趋势并根据波动率调整头寸。MACD参数因资产而异，波动率每日使用“风险度量”方法计算。第一年的数据用于优化波动率阈值，并应用于样本外交易。在低波动率环境中，当短期移动平均线（MA）高于长期移动平均线时，投资者做多；当短期移动平均线低于长期移动平均线时，做空。在高波动率环境中，这一规则相反。期货采用等权重配置，投资组合每日根据更新的信号进行再平衡。
II. 策略合理性
学术研究认为，高波动率时期通常与价格方向变化的时期相关。这些时期对动量/趋势跟踪策略不利，因为这些策略更适合低波动率的趋势环境。因此，建议在高波动率时期采用反转交易规则。
III. 来源论文
资产管理中的波动率过滤：在管理期货中的应用 [点击查看论文]
- Dunis、Miao，利物浦约翰摩尔斯大学银行与金融学教授兼CIBEF主任，CIBEF副研究员，目前正在利物浦约翰摩尔斯大学攻读博士学位。
<摘要>
众所周知，技术交易规则在高波动率时期表现不佳。本文的目的是研究波动率过滤器的加入是否能提高模型的表现。与以往从学术角度研究技术交易规则的文献不同，本文试图与现实世界的业务联系起来：构建了两个与管理期货指数和货币交易者基准指数高度相关的投资组合，以模拟典型的管理期货和管理货币基金的表现。然后，将提出的波动率过滤器直接应用于这两个投资组合，希望所提出的技术既具有学术意义，又具有工业意义。提出了两种波动率过滤器，一种是“不交易”过滤器，即在波动时期关闭所有市场头寸；另一种是“反转”过滤器，即如果市场波动率高于给定阈值，则反转简单移动平均收敛和发散（MACD）的信号。为了评估模型表现的一致性，整个时期（1999年1月4日至2004年12月31日）被分为3个子时期。我们的结果表明，在所有3个子时期中，加入这两种波动率过滤器在年化回报、最大回撤、风险调整后的夏普比率和卡尔马比率方面都为模型表现增加了价值。
IV. 回测表现
| 年化回报 | 5.47% |
| 波动率 | 5.51% |
| β值 | 0.069 |
| 夏普比率 | 0.99 |
| 索提诺比率 | -0.793 |
| 最大回撤 | -4.88% |
| 胜率 | 48% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from collections import deque
class TimeSeriesMomentumVolatilityFilters(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.symbols = [
'ICE_DX1', # US Dollar Index Futures, Continuous Contract #1
'CME_TY1', # 10 Yr Note Futures, Continuous Contract #1
'CME_ES1', # E-mini S&P 500 Futures, Continuous Contract #1
'CME_EC1', # Euro FX Futures, Continuous Contract #1
'CME_BP1', # British Pound Futures, Continuous Contract #1
'CME_HG1' # Copper Futures, Continuous Contract
]
self.period = 12*21
self.vol_period = 21
self.SetWarmUp(self.period)
self.optimalization_count = 100
self.data = {}
self.macd = {}
self.macd_signal = {}
self.volatility = {}
for symbol in self.symbols:
data = self.AddData(QuantpediaFutures, symbol, Resolution.Daily)
data.SetFeeModel(CustomFeeModel())
self.data[symbol] = deque(maxlen=self.period+self.vol_period+1)
self.volatility[symbol] = deque(maxlen=self.period)
self.macd_signal[symbol] = deque(maxlen=self.period)
self.macd['ICE_DX1'] = self.MACD('ICE_DX1', 1, 250, 1)
self.macd['CME_TY1'] = self.MACD('CME_TY1', 1, 250, 1)
self.macd['CME_HG1'] = self.MACD('CME_HG1', 1, 250, 1)
self.macd['CME_EC1'] = self.MACD('CME_EC1', 1, 61, 1)
self.macd['CME_BP1'] = self.MACD('CME_BP1', 1, 61, 1)
self.macd['CME_ES1'] = self.MACD('CME_ES1', 3, 250, 1)
def OnData(self, data):
for symbol in self.data:
symbol_obj = self.Symbol(symbol)
if symbol_obj in data.Keys:
if data[symbol_obj]:
price = data[symbol_obj].Value
if price != 0:
self.data[symbol].append(price)
if self.IsWarmingUp: return
self.Liquidate()
# Optimalization
optimal_treshold = {}
for symbol in self.symbols:
if self.securities[symbol].get_last_data() and self.time.date() > QuantpediaFutures.get_last_update_date()[symbol]:
self.liquidate(symbol)
continue
macd = self.macd[symbol]
closes = np.array([x for x in self.data[symbol]])
if macd.IsReady and len(closes) >= self.vol_period:
# MACD SIGNAL = if short term MA is UNDER long term MA == 1 else -1
self.macd_signal[symbol].append(1 if macd.Slow > macd.Fast else -1)
daily_return = closes[:-1] / closes[1:] - 1
vol = np.std(daily_return) * np.sqrt(252)
self.volatility[symbol].append(vol)
if len(closes) == self.data[symbol].maxlen and len(self.macd_signal[symbol]) == self.macd_signal[symbol].maxlen:
values = np.array(closes)
daily_changes = (values[1:] - values[:-1]) / values[:-1]
vol_min = min(self.volatility[symbol])
vol_max = max(self.volatility[symbol])
vol_range = vol_max - vol_min
vol_step = vol_range / self.optimalization_count
avg_return = {}
treshhold_value = vol_min
while treshhold_value <= vol_max:
vol_vector = np.array([-1 if x <= treshhold_value else 1 for x in self.volatility[symbol]])
returns = vol_vector * self.macd_signal[symbol] * daily_changes[-self.period:]
avg_return[treshhold_value] = np.average([x for x in returns])
treshhold_value += vol_step
optimal_treshold[symbol] = max(avg_return, key=avg_return.get)
if len(optimal_treshold) == 0: return
# Trading
count = len(optimal_treshold)
for symbol, threshold in optimal_treshold.items():
vol = self.volatility[symbol][-1]
if symbol in data and data[symbol]:
# low volatility enviroment
if vol <= threshold:
if self.macd[symbol].Fast > self.macd[symbol].Slow:
self.SetHoldings(symbol, 1/count)
else:
self.SetHoldings(symbol, -1/count)
# high volatility enviroment
else:
if self.macd[symbol].Fast > self.macd[symbol].Slow:
self.SetHoldings(symbol, -1/count)
else:
self.SetHoldings(symbol, 1/count)
# 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"))