外汇市场中的时间序列动量与波动率过滤相结合

““该策略使用3倍MACD指标作为趋势信号交易9种加权货币对，并根据波动率阈值调整头寸，每日进行再平衡以适应市场条件。”

资产类别: 差价合约（CFDs）、期货 | 地区: 全球 | 周期: 每日 | 市场: 外汇 | 关键词: 时间序列、动量、波动率

I. 策略概要

该策略交易9种按国际清算银行（BIS）交易量加权的货币对，包括美元/日元、英镑/美元、欧元/美元等。它使用参数为（1/32, 1/61, 1/117）的3倍MACD指标来确定趋势。波动率每日使用“风险度量”方法计算。第一年的数据用于优化波动率阈值，并应用于样本外交易。在低波动率环境中，当短期移动平均线（MA）高于长期移动平均线时，投资者做多；当短期移动平均线低于长期移动平均线时，做空。在高波动率条件下，这一规则相反。头寸每日进行再平衡，确保投资组合适应不断变化的市场动态。

II. 策略合理性

学术研究认为，高波动率时期通常与价格方向变化的时期相关。这些时期对动量/趋势跟踪策略不利，因为这些策略更适合低波动率的趋势环境。因此，建议在高波动率时期采用反转交易规则。

III. 来源论文

资产管理中的波动率过滤：在管理期货中的应用 [点击查看论文]

• Dunis、Miao，利物浦约翰摩尔斯大学银行与金融学教授兼CIBEF主任，CIBEF副研究员，目前正在利物浦约翰摩尔斯大学攻读博士学位。

<摘要>

众所周知，技术交易规则在高波动率时期表现不佳。本文的目的是研究波动率过滤器的加入是否能提高模型的表现。与以往从学术角度研究技术交易规则的文献不同，本文试图与现实世界的业务联系起来：构建了两个与管理期货指数和货币交易者基准指数高度相关的投资组合，以模拟典型的管理期货和管理货币基金的表现。然后，将提出的波动率过滤器直接应用于这两个投资组合，希望所提出的技术既具有学术意义，又具有工业意义。提出了两种波动率过滤器，一种是“不交易”过滤器，即在波动时期关闭所有市场头寸；另一种是“反转”过滤器，即如果市场波动率高于给定阈值，则反转简单移动平均收敛和发散（MACD）的信号。为了评估模型表现的一致性，整个时期（1999年1月4日至2004年12月31日）被分为3个子时期。我们的结果表明，在所有3个子时期中，加入这两种波动率过滤器在年化回报、最大回撤、风险调整后的夏普比率和卡尔马比率方面都为模型表现增加了价值。

IV. 回测表现

年化回报	4.63%
波动率	5%
β值	0.004
夏普比率	0.93
索提诺比率	N/A
最大回撤	-9.3%
胜率	31%

V. 完整的 Python 代码

from AlgorithmImports import *
import datetime
import numpy as np
#endregion
class TimeSeriesMomentumCombinedwithVolatilityFiltersinFOREX(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2015, 1, 1)
        self.SetCash(100000)
        self.period = 12*21
        self.SetWarmUp(self.period)
        
        self.current_date = -1
        
        self.symbols = ["USDJPY", "GBPUSD", "EURUSD", "USDCHF", "USDCAD", "AUDUSD", "EURGBP", "EURJPY", "EURCHF"]
        self.data = {}
                       
        for symbol in self.symbols:
            data = self.AddForex(symbol, Resolution.Minute, Market.FXCM)
            data.SetFeeModel(CustomFeeModel())
        
            if symbol not in self.data:
                ma32 = SimpleMovingAverage(symbol, 32)
                ma61 = SimpleMovingAverage(symbol, 61)
                ma117 = SimpleMovingAverage(symbol, 117)
                
                self.data[symbol] = SymbolData(symbol, self.period, ma32, ma61, ma117, self.GetWeight(symbol))
    def OnData(self, data):
        if self.Time.time() == datetime.time(0,0,0):
            # Updating last day close.
            for symbol in self.symbols:
                if symbol in data and data[symbol]:
                    close = data[symbol].Value
                    self.data[symbol].update(self.Time, close)
        
        # Trading one minute before day ends.
        if (self.Time + timedelta(minutes=1)).date() ==  self.current_date: 
            return
                
        self.current_date = (self.Time + timedelta(minutes=1)).date()
        
        # Wait until data are warmed up.
        if self.IsWarmingUp: return
    
        # Trade execution
        self.Liquidate()
        
        for symbol in self.symbols:
            if self.data[symbol].is_ready():
                yearly_vol = self.data[symbol].volatility(self.period)
                monthly_vol = self.data[symbol].volatility(21)
    
                price = self.data[symbol].Price
                ma32 = self.data[symbol].ma32()
                ma61 = self.data[symbol].ma61()
                ma117 = self.data[symbol].ma117()
    
                traded_weight = 0
                weight = self.data[symbol].Weight
                    
                # Trend startegy
                if monthly_vol < yearly_vol:
                    #Long
                    if ma32 > ma61 and ma61 > ma117:
                        if price > ma32 and price > ma61 and price > ma117:
                            traded_weight = weight
                        elif price > ma61 and price > ma117:
                            traded_weight = weight * 2/3
                        elif price > ma117:
                            traded_weight = weight * 1/3
                        else:
                            continue
                    #Short        
                    elif ma32 < ma61 and ma61 < ma117:
                        if price < ma32 and price < ma61 and price < ma117:
                            traded_weight = -weight
                        elif price < ma61 and price < ma117:
                            traded_weight = -weight * 2/3
                        elif price < ma117:
                            traded_weight = -weight * 1/3
                        else:
                            continue
                    
                    self.SetHoldings(symbol, traded_weight)
    
                # Counter-Trend startegy    
                elif monthly_vol > yearly_vol:
                    #Long        
                    if ma32 < ma61 and ma61 < ma117:
                        if price < ma32 and price < ma61 and price < ma117:
                            traded_weight = weight
                        elif price < ma61 and price < ma117:
                            traded_weight = weight * 2/3
                        elif price < ma117:
                            traded_weight = weight * 1/3
                        else:
                            continue
                    #Short
                    elif ma32 > ma61 and ma61 > ma117:
                        if price > ma32 and price > ma61 and price > ma117:
                            traded_weight = -weight
                        elif price > ma61 and price > ma117:
                            traded_weight = -weight * 2/3
                        elif price > ma117:
                            traded_weight = -weight * 1/3
                        else:
                            continue
                            
                    self.SetHoldings(symbol, traded_weight)
                        
    def GetWeight(self, argument):
        switcher = {
            "USDJPY": 0.2113,
            "GBPUSD": 0.1749,
            "EURUSD": 0.3576,
            "USDCHF": 0.0557,
            "USDCAD": 0.0507,
            "AUDUSD": 0.0642,
            "EURGBP": 0.0307,
            "EURJPY": 0.0364,
            "EURCHF": 0.0186,
        }
        return switcher.get(argument, "0.0")
class SymbolData:
    def __init__(self, symbol, lookback, ma32, ma61, ma117, weight):
        self.Symbol = symbol
        self.Price = None
        self.MA32 = ma32
        self.MA61 = ma61
        self.MA117 = ma117
        self.Weight = weight
        self.History = RollingWindow[float](lookback)
    def update(self, time, value):
        self.Price = value
        self.History.Add(value)
        self.MA32.Update(time, value)
        self.MA61.Update(time, value)
        self.MA117.Update(time, value)
        
    def is_ready(self):
        return self.MA32.IsReady and self.MA61.IsReady and self.MA32.IsReady and self.History.IsReady
    
    def ma32(self):
        return self.MA32.Current.Value
    def ma61(self):
        return self.MA61.Current.Value
    def ma117(self):
        return self.MA117.Current.Value
        
    def volatility(self, period):
        prices = np.array([x for x in self.History])[:period]
        returns = (prices[:-1]-prices[1:])/prices[1:]
        return np.std(returns) * np.sqrt(period)
# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))