“通过产出缺口交易27种货币，产出缺口由工业生产残差计算得出，做多最高五分位，做空最低五分位，投资组合等权重，每月重新平衡。”
资产类别: 差价合约、远期、期货 | 地区: 全球 | 周期: 每月 | 市场: 外汇 | 关键词: 工业产出
I. 策略概要
投资范围包括27个国家兑美元的货币对，不包括欧元区成员国在欧元采用后的货币。产出缺口使用OECD数据库的工业生产数据计算。对于每个国家，使用汉密尔顿（2018）的线性投影方法进行月度回归，其中对数工业生产对24个月和额外的11个滞后进行回归。产出缺口是该回归的残差，表示与拟合值的偏差。货币根据产出缺口分为五分位，做多最高五分位，做空最低五分位。投资组合等权重，每月重新平衡。
II. 策略合理性
该交易策略得到了商业周期是货币回报强有力预测因素的发现支持。经济强劲国家的货币表现始终优于经济疲软国家的货币，这种模式在横截面和时间序列方法中都有观察到，尽管两者之间的相关性较低。该策略主要反映宏观经济风险，并且与利差、动量或价值等既定货币策略不同。即使投资组合根据这些异常现象进行排序，产出缺口仍然是货币定价中的一个重要因素，突显了其在预测横截面货币回报中的独特作用。
III. 来源论文
Business Cycles and Currency Returns [点击查看论文]
- Riccardo Colacito, Steven J. Riddiough, and Lucio Sarno
<摘要>
我们发现货币超额回报与商业周期的相对强度之间存在密切联系。买入经济强劲国家的货币并卖出经济疲软国家的货币，无论是在横截面还是在时间序列上，都能产生高回报。这些回报主要源于即期汇率的可预测性，与常见的货币投资策略无关，并且在无条件或有条件的资产定价测试中都无法用传统的货币风险因素来理解。我们还表明，我们的结果所隐含的商业周期因素在广泛的货币横截面中得到了定价。
IV. 回测表现
| 年化回报 | 4.92% |
| 波动率 | 6.83% |
| β值 | -0.007 |
| 夏普比率 | 0.72 |
| 索提诺比率 | N/A |
| 最大回撤 | -6.88% |
| 胜率 | 54% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
import statsmodels.api as sm
#endregion
class OutputGapPredictsFXReturns(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.symbols = [
("CME_AD1", "AUS"), # Australian Dollar Futures, "AUS"
("CME_BP1", "GBR"), # British Pound Futures, "GBR"
("CME_CD1", "CAN"), # Canadian Dollar Futures, "CAN"
("CME_EC1", "EA19"), # Euro FX Futures, "EA19"
("CME_JY1", "JPN"), # Japanese Yen Futures, "JPN"
("CME_MP1", "MEX"), # Mexican Peso Futures, "MEX"
("CME_NE1", "NZL"), # New Zealand Dollar Futures, "NZL"
("CME_SF1", "CHE") # Swiss Franc Futures "CHE"
]
self.data = {}
self.length = 12 # Lenght of each variable in regression
self.regression_period = 48 # Need 48 monthly data for regression x
self.quantile = 5
self.max_missing_days = 5
# These are countries with quarterly data, which were adjusted to monthly data
self.not_complete_monthly_data = ['AUS', 'NZL', 'CHE']
# Takes only countries with complete monthly data, when flag is True
self.monthly_flag = False
for currency, industrial in self.symbols:
# Check if strategy works with all data
if not self.monthly_flag or industrial not in self.not_complete_monthly_data:
# Subscribe to future
data = self.AddData(QuantpediaFutures, currency, Resolution.Daily)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(5)
# Subscribe to country industrial
self.AddData(QuantpediaIndustrial, industrial, Resolution.Daily)
# Create SymbolData for industrial, with future symbol
self.data[industrial] = SymbolData(self.regression_period, data.Symbol)
self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.recent_month = -1
def OnData(self, data):
# Update industrial values
for symbol in self.data:
if symbol in data and data[symbol]:
value = data[symbol].Value
self.data[symbol].update(value, self.Time.date())
if self.recent_month == self.Time.month:
return
self.recent_month = self.Time.month
output_gaps = {}
# Go through each country industrial and make regression
for industrial_symbol, symbol_data in self.data.items():
if not self.Securities[industrial_symbol].GetLastData() or not self.Securities[symbol_data.future_symbol].GetLastData():
continue
if (self.Time.date() - self.Securities[industrial_symbol].GetLastData().Time.date()).days > self.max_missing_days or \
(self.Time.date() - self.Securities[symbol_data.future_symbol].GetLastData().Time.date()).days > self.max_missing_days:
continue
# Check if regression data are ready
if not symbol_data.is_ready():
continue
# Change flag to prevent including country in next rebalance, if new value won't come in OnData
symbol_data.new_value_flag = False
# Get future symbol for this country
future_symbol = symbol_data.future_symbol
# Create regression variables (y, x)
regression_y, regression_x = symbol_data.create_regression_variables(self.length)
regression_model = self.MultipleLinearRegression(regression_x, regression_y)
# Store output gap under future symbol of this country
output_gaps[future_symbol] = regression_model.resid[-1]
# Continue only if there is enough data for quintile selection
if len(output_gaps) < self.quantile:
self.Liquidate()
return
# Sort by output gap
quantile = int(len(output_gaps) / self.quantile)
sorted_by_output_gap = [x[0] for x in sorted(output_gaps.items(), key=lambda item: item[1])]
# Go long the top quintile, and short the bottom quintile
long = sorted_by_output_gap[-quantile:]
short = sorted_by_output_gap[:quantile]
# Trade execution
invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long + short:
self.Liquidate(symbol)
long_length = len(long)
short_length = len(short)
for symbol in long:
self.SetHoldings(symbol, 1 / long_length)
for symbol in short:
self.SetHoldings(symbol, -1 / short_length)
def MultipleLinearRegression(self, x, y):
x = np.array(x).T
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result
class SymbolData():
def __init__(self, period, future_symbol):
self.values = RollingWindow[float](period)
self.future_symbol = future_symbol
self.new_value_flag = False
def update(self, value, update_date:datetime.date):
self.values.Add(value)
self.new_value_flag = True
def is_ready(self):
return self.values.IsReady and self.new_value_flag
def create_regression_variables(self, length):
values = [x for x in self.values]
# regression y
y = values[:length][::-1]
# regression x
x = []
for i in range(24, 36):
x.append(values[i:i+length][::-1])
return y, x
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaFutures(PythonData):
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])
return data
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaIndustrial(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/economic/industrial_production/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaIndustrial()
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['value'] = float(split[1])
data.Value = float(split[1])
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"))