外汇 · 全球 Free

时机选择套利交易策略

学术论文

Predictability of Currency Carry Trades and Asset Pricing Implications

作者作者：Gurdip Bakshi 和 George Panayotov; 机构：天普大学福克斯商学院（Temple University-Fox School of Business）

机构

?香港科技大学（HKUST）

论文摘要

本文研究了基于远期贴水选择相对于美元买入或卖出的货币套利交易的时间序列可预测性。研究发现，商品指数的变化、货币波动率的变化，以及在较小程度上流动性指标，可以在样本内有效预测动态再平衡套利交易的收益。这一结论通过月度预测回归（预测期长达六个月）的个体和联合p值得到了验证。

预测能力还通过样本外指标得到了进一步支持。基于预测能力的决策规则显著提高了套利交易的夏普比率和收益分布偏度特征。此外，证据表明，预测能力主要来源于套利交易的多头头寸及其货币组成部分。

研究进一步检验了资产定价模型的理论限制，该模型将平均货币收益和货币波动率创新的模拟组合作为风险因子，并将其限制应用于预测回归中的系数。研究结果为货币套利交易的可预测性和基于风险因子的资产定价模型提供了新的实证支持。

论文链接http://papers.ssrn.com/sol3/papers.cfm?abstract_id=1977642

策略概要

该策略涉及8种货币（AUD、CAD、CHF、GBP、JPY、NOK、NZD、SEK），采用套利交易方法，即买入收益率最高的货币，对收益率最低的货币进行卖出（均相对于美元）。策略使用以下三个预测变量：

CRB现货商品指数（原材料类）三个月对数变化

货币市场波动率的标准化日对数变化

TED利差等效指标推导出的整体流动性变量

通过线性回归模型（使用180个月扩展窗口估算），预测一个月的套利交易收益率。仅当模型预测收益为正时才进行交易；如果预测为负，则避免交易。

策略合理性

研究表明，商品市场的投资增加通常与更高的风险偏好一致，从而有利于套利交易。然而，在波动性较高的货币市场中，套利交易的盈利能力往往下降，因此货币波动率变化是一个有用的预测指标。此外，全球市场压力上升（反映在流动性变化中）也对套利交易产生重要影响，其中TED利差被证明是有效的预测变量之一。

通过综合商品市场、货币波动率和流动性因素，该策略动态适应市场环境，仅在正预期回报时进行交易，避免了高风险的负预期场景。

回测表现

年化收益12.6%

波动率10%

贝塔0.099

夏普比率0.86

胜率62%

完整 Python 代码

import data_tools
from AlgorithmImports import *
from collections import deque
import numpy as np
class TimingCarryTrade(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1) 
self.SetCash(100000)

# Source: https://www.quandl.com/data/OECD-Organisation-for-Economic-Co-operation-and-Development
self.symbols = {
                "CME_AD1" : "OECD/KEI_IR3TIB01_AUS_ST_M",   # Australian Dollar Futures, Continuous Contract #1
                "CME_BP1" : "OECD/KEI_IR3TIB01_GBR_ST_M",   # British Pound Futures, Continuous Contract #1
                "CME_CD1" : "OECD/KEI_IR3TIB01_CAN_ST_M",   # Canadian Dollar Futures, Continuous Contract #1
                "CME_EC1" : "OECD/KEI_IR3TIB01_EA19_ST_M",  # Euro FX Futures, Continuous Contract #1
                "CME_JY1" : "OECD/KEI_IR3TIB01_JPN_ST_M",   # Japanese Yen Futures, Continuous Contract #1
                "CME_MP1" : "OECD/KEI_IR3TIB01_MEX_ST_M",   # Mexican Peso Futures, Continuous Contract #1
                "CME_NE1" : "OECD/KEI_IR3TIB01_NZL_ST_M",   # New Zealand Dollar Futures, Continuous Contract #1
                "CME_SF1" : "SNB/ZIMOMA"                    # Swiss Franc Futures, Continuous Contract #1
}

# Daily price data.
self.data = {}
self.period = 3 * 21
self.SetWarmUp(self.period)
# Regression data rolling window.
self.regression_min_period = 60
self.regression_data = deque()

self.commodity_index = self.AddEquity('DBC', Resolution.Daily).Symbol    
self.data[self.commodity_index] = deque(maxlen = self.period)

# Quandl ted spread.
self.ted_spread = self.AddData(data_tools.QuandlValue, 'FRED/TEDRATE', Resolution.Daily).Symbol

# Last selected long and short symbols used to calculate monthly performace of carry trade.
self.long = []
self.short = []

for symbol, rate_symbol in self.symbols.items():
    self.AddData(data_tools.QuandlValue, rate_symbol, Resolution.Daily)
    data = self.AddData(data_tools.QuantpediaFutures, symbol, Resolution.Daily)
    data.SetFeeModel(data_tools.CustomFeeModel())
    data.SetLeverage(5)
    
    self.data[symbol] = deque(maxlen = self.period)

self.Schedule.On(self.DateRules.MonthStart(self.commodity_index), self.TimeRules.AfterMarketOpen(self.commodity_index), self.Rebalance)
def OnData(self, data):
for symbol in self.data:
    symbol_obj = self.Symbol(symbol)
    if symbol_obj in data and data[symbol_obj]:
        price = data[symbol_obj].Value
        self.data[symbol].append(price)
def Rebalance(self):
# make sure data is still comming in
if self.Securities[self.commodity_index].GetLastData() and (self.Time.date() - self.Securities[self.commodity_index].GetLastData().Time.date()).days > 5:
    self.Liquidate()
    return
if self.Securities[self.ted_spread].GetLastData() and (self.Time.date() - self.Securities[self.ted_spread].GetLastData().Time.date()).days > 5:
    self.Liquidate()
    return

# Regression data.
carry_trade_performance = None
delta_crb = None
delta_vol = None
ted_spread = None

# Calculate carry trade last month's performance.
if len(self.long) != 0 and len(self.short) != 0:
    carry_trade_perf_long = np.sum([data_tools.Return([y for y in self.data[x]][-21:]) for x in self.long if x in self.data and len(self.data[x]) == self.data[x].maxlen])
    carry_trade_perf_short = np.sum([-1 * data_tools.Return([y for y in self.data[x]][-21:]) for x in self.short if x in self.data and len(self.data[x]) == self.data[x].maxlen])
    carry_trade_performance = (carry_trade_perf_long + carry_trade_perf_short) / len(self.long + self.short)
    self.long.clear()
    self.short.clear()
    
# Interbank rate sorting.
sorted_by_rate = sorted([y for y in self.symbols if self.Securities[y].GetLastData() and (self.Time.date() - self.Securities[y].GetLastData().Time.date()).days <= 5 and \
                        self.Securities[self.symbols[y]].GetLastData() and (self.Time.date() - self.Securities[self.symbols[y]].GetLastData().Time.date()).days <= 31],   \
    key = lambda x: self.Securities[self.symbols[x]].Price, reverse = True)
traded_count = 3
if len(sorted_by_rate) > traded_count*2:
    self.long = [x for x in sorted_by_rate[:traded_count]]
    self.short = [x for x in sorted_by_rate[-traded_count:]]

# Commodity index log change.
if len(self.data[self.commodity_index]) == self.data[self.commodity_index].maxlen:
    delta_crb = np.log(self.data[self.commodity_index][-1] / self.data[self.commodity_index][0]) / 3
    
# Average volatility.
# t and t-3 volatility tuples.
t_t_3_vol = [(data_tools.Volatility([y for y in self.data[x]][-21:]), data_tools.Volatility([y for y in self.data[x]][:21])) for x in self.symbols if x in self.data and len(self.data[x]) == self.data[x].maxlen]
if len(t_t_3_vol) != 0:
    avg_t_volatility = np.average([x[0] for x in t_t_3_vol])
    avg_t_3_volatility = np.average([x[1] for x in t_t_3_vol])
    if avg_t_volatility != 0 and avg_t_3_volatility != 0:
        delta_vol = np.log(avg_t_volatility / avg_t_3_volatility) / 3
# TED spread.
if self.Securities[self.ted_spread].Price != 0:
    ted_spread = self.Securities[self.ted_spread].Price

if delta_crb and delta_vol and ted_spread:
    self.regression_data.append((carry_trade_performance, delta_crb, delta_vol, ted_spread))

# Regression data is ready.
if len(self.regression_data) >= self.regression_min_period:
    carry_trade_performances = [float(x[0]) for x in self.regression_data]
    crb_deltas = [float(x[1]) for x in self.regression_data]
    vol_deltas = [float(x[2]) for x in self.regression_data]
    ted_spreads = [float(x[3]) for x in self.regression_data]

    # Regression.
    x = [crb_deltas[:-1], vol_deltas[:-1], ted_spreads[:-1]]
    regression_model = data_tools.MultipleLinearRegression(x, carry_trade_performances[1:])
    # Predicted carry return.
    alpha = regression_model.params[0]
    prediction_x = [crb_deltas[-1], vol_deltas[-1], ted_spreads[-1]]
    betas = np.array(regression_model.params[1:])
    carry_return_predicted = alpha + sum(np.multiply(betas, prediction_x))
    
    if carry_return_predicted > 0:
        # Trade execution.
        invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
        for symbol in invested:
            if symbol not in self.long + self.short:
                self.Liquidate(symbol)
        
        for symbol in self.long:
            self.SetHoldings(symbol, 1 / len(self.long))
        for symbol in self.short:
            self.SetHoldings(symbol, -1 / len(self.short))