“该策略投资于45个G10货币对,计算股票回报差异。每个货币对分配等权重,每月使用一个月期货币远期合约跟踪业绩。”

I. 策略概要

该策略投资于G10货币中的45个货币对。它在每个月末计算以当地货币计价的过去12个月股票指数总回报的差异。每个货币对都被翻转以反映正的股票差异。该策略为每个货币对(或具有最大规模差异的子集)分配等权重,并抵消货币敞口以确定最终权重。投资组合的业绩每月记录,通过一个月期货币远期合约实施,并且该过程每月重复。

II. 策略合理性

该策略使用过去12个月的股票指数回报而非利率来确定货币头寸。买入近期股票回报高的国家的货币,卖出回报低的国家的货币。这种“股票差异”策略通过远期合约执行。与高股票回报国家相关的货币往往跑赢与低股票回报国家相关的货币,显示出稳健且一致的结果。如果股票溢价较高的国家也具有较高的固有风险,那么其较高的本地回报可能与较高的货币回报相关,这表明货币风险和国家股票风险是相互关联的,并影响利率。

III. 来源论文

The Equity Differential Factor in Currency Markets [点击查看论文]

<摘要>

我们发现,各国滞后股票市场表现的差异强烈预测了货币回报的横截面。具体而言,在前一年股票回报最强的国家,汇率往往会升值。自1990年以来,基于该因子构建的投资组合表现优于基于传统套利、趋势和估值因子的货币投资组合。股票差异因子无法用这些传统因子解释,并且产生了统计上显著的超额阿尔法。其表现非常一致且对不同公式具有鲁棒性。我们提供的证据表明,投资者对跑赢股票市场的需求可能促成了这种效应。

IV. 回测表现

年化回报1.9%
波动率1.3%
β值-0.011
夏普比率0.61
索提诺比率-1.354
最大回撤N/A
胜率24%

V. 完整的 Python 代码

from AlgorithmImports import *
class EquityDifferentialCurrencies(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2004, 1, 1)
        self.SetCash(100000)
        
        self.symbols = {
            'CME_AD1' : 'ASX_YAP1',
            'CME_BP1' : 'LIFFE_Z1',
            'CME_CD1' : 'LIFFE_FCE1',
            'CME_EC1' : 'EUREX_FSTX1',
            'CME_JY1' : 'SGX_NK1',
            'CME_SF1' : 'EUREX_FSMI1'
        }
        self.data = {}
        self.period = 12*21
        self.SetWarmUp(self.period)
        self.leverage = 5
        
        for symbol in self.symbols:
            index = self.symbols[symbol]
            
            data = self.AddData(QuantpediaFutures, symbol, Resolution.Daily)
            data.SetLeverage(self.leverage)
            data.SetFeeModel(CustomFeeModel())
            
            self.AddData(QuantpediaFutures, index, Resolution.Daily)
            self.data[index] = RollingWindow[float](self.period)
            
        first_key = [x for x in self.symbols.keys()][0]
        symbol = self.Symbol(self.symbols[first_key])
        self.rebalance_flag: bool = False
        self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.At(0, 0), self.Rebalance)
    def OnData(self, data):
        for symbol in self.symbols:
            index:str = self.symbols[symbol]
            if index in data and data[index]:
                price:float = data[index].Value
                self.data[index].Add(price)
        if not self.rebalance_flag:
            return
        self.rebalance_flag = False
        if self.IsWarmingUp: return
        index_return = {}
        for symbol in self.symbols:
            index = self.symbols[symbol]
            if all([self.Securities[x].GetLastData() and self.Time.date() < QuantpediaFutures.get_last_update_date()[x] for x in [symbol, index]]):
                if self.data[index].IsReady:
                    index_return[index] = self.data[index][0] / self.data[index][self.period-1] - 1
        self.Liquidate()
        if len(index_return) == 0: return
        
        len_ = len(self.symbols)
        count = (len_ * (len_-1)) * 2
        
        for i in range(0, len(self.symbols)):
            key_i = [x for x in self.symbols.keys()][i]
            for j in range(i+1, len(self.symbols)):
                key_j = [x for x in self.symbols.keys()][j]
                if all([data.contains_key(symbol) and data[symbol] for symbol in [key_i, key_j]]):
                    eq_index1 = self.symbols[key_i]
                    eq_index2 = self.symbols[key_j]
                    if eq_index1 in index_return and eq_index2 in index_return:
                        if index_return[eq_index1] > index_return[eq_index2]:
                            self.SetHoldings(key_i, self.leverage*(1/count))
                            self.SetHoldings(key_j, -self.leverage*(1/count))
                        else:
                            self.SetHoldings(key_i, -self.leverage*(1/count))
                            self.SetHoldings(key_j, self.leverage*(1/count))
    def Rebalance(self):
        self.rebalance_flag = True
# 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"))

发表评论

了解 Quant Buffet 的更多信息

立即订阅以继续阅读并访问完整档案。

继续阅读