
“该策略投资于45个G10货币对,计算股票回报差异。每个货币对分配等权重,每月使用一个月期货币远期合约跟踪业绩。”
资产类别: 差价合约、远期、期货、掉期 | 地区: 全球 | 周期: 每月 | 市场: 外汇 | 关键词: 差异因子
I. 策略概要
该策略投资于G10货币中的45个货币对。它在每个月末计算以当地货币计价的过去12个月股票指数总回报的差异。每个货币对都被翻转以反映正的股票差异。该策略为每个货币对(或具有最大规模差异的子集)分配等权重,并抵消货币敞口以确定最终权重。投资组合的业绩每月记录,通过一个月期货币远期合约实施,并且该过程每月重复。
II. 策略合理性
该策略使用过去12个月的股票指数回报而非利率来确定货币头寸。买入近期股票回报高的国家的货币,卖出回报低的国家的货币。这种“股票差异”策略通过远期合约执行。与高股票回报国家相关的货币往往跑赢与低股票回报国家相关的货币,显示出稳健且一致的结果。如果股票溢价较高的国家也具有较高的固有风险,那么其较高的本地回报可能与较高的货币回报相关,这表明货币风险和国家股票风险是相互关联的,并影响利率。
III. 来源论文
The Equity Differential Factor in Currency Markets [点击查看论文]
- 大卫·特金顿(David Turkington)、阿里雷扎·亚兹达尼(Alireza Yazdani)。CFA,现任马萨诸塞州剑桥市State Street Associates高级董事;副总裁,State Street Associates,剑桥,马萨诸塞州。
<摘要>
我们发现,各国滞后股票市场表现的差异强烈预测了货币回报的横截面。具体而言,在前一年股票回报最强的国家,汇率往往会升值。自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"))