货币中的经济动能

“该策略交易19种货币兑美元，使用8个基本变量和趋势指标（变化、时间趋势），跨越1-60个月的回溯期。货币按趋势强度排名，形成按逆波动率加权的投资组合。一个综合投资组合结合了跨变量和周期的策略，按过去波动率的5%进行缩放。投资组合每月重新平衡，以实现一致的风险调整后回报。”

资产类别: 差价合约、远期、期货、掉期 | 地区: 全球 | 周期: 每月 | 市场: 外汇 | 关键词: 动量，货币

I. 策略概要

该策略以美元为基准，目标是19种货币，使用8个基本变量：一个月银行同业拆借利率、收益率差、十年期利率、通货膨胀、贸易余额、工业生产、零售额和失业率。从经合组织和其他来源收集的基本数据经过处理，以创建贸易余额、生产指数和收益率差等经济指标。

趋势指标包括变化/对数变化和线性时间趋势。使用1-60个月的回溯期，低频变量使用24-60个月。对于短期利率，动量是通过按波动率标准化的变化构建的，通过指数加权移动平均线（衰减参数：0.94）计算。对于其他变量，趋势强度通过线性时间趋势回归的t统计量来衡量。

货币每月按每个变量和回溯期的趋势信号强度进行排名。投资组合基于横截面排名形成，子策略的权重与其过去三年的波动率成反比。综合投资组合结合了变量、趋势指标和回溯期。权重缩放至总和为一，最终投资组合缩放为5%除以综合策略的过去已实现波动率，确保一致的风险敞口。

该策略整合了跨多个维度的趋势跟踪和基本面分析，以利用经济和市场信号进行优化货币交易。投资组合每月重新平衡。

II. 策略合理性

学术研究表明，货币中存在经济动量，因为过去的宏观趋势反映了对未来基本面的预期。该策略的回报涵盖了套利交易的阿尔法，表明按宏观趋势排序包括了套利排序。超过一半的经济动量回报无法用标准策略解释，表明它提供了额外的、独特的收益来源。虽然该论文探讨了潜在的解释，但这些回报的确切来源仍然不确定，突显了经济动量作为货币市场中一项有价值且独特的投资机会。

III. 来源论文

经济动量与货币回报 [点击查看论文]

• Dahlquist, Hasseltoft

<摘要>

广泛的基本变量的过去趋势可以预测货币回报。我们记录了一个交易策略，该策略做多经济动量强劲国家的货币，做空经济动量疲软国家的货币，其年化夏普比率约为1，并且在控制标准套利交易、动量和价值策略后，能产生显著的阿尔法。经济动量策略涵盖了套利交易的阿尔法，表明国家间套利交易的差异被过去经济趋势的差异所捕获。此外，我们研究了投资者对基本变量的预期，发现预期是外推性的，但与投资组合权重呈负相关，而投资组合权重对各国的经济趋势进行排名。

IV. 回测表现

年化回报	6.15%
波动率	5.6%
β值	0.003
夏普比率	1.1
索提诺比率	N/A
最大回撤	-12.39%
胜率	28%

V. 完整的 Python 代码

from AlgorithmImports import *
from io import StringIO
import data_tools
from dateutil.relativedelta import relativedelta
import numpy as np
import pandas as pd
from collections import deque
from pandas.core.frame import dataframe
# endregion
class EconomicMomentuminCurrencies(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2005, 1, 1)
        self.SetCash(100000)
        self.leverage:int = 5
        self.period:int = 60
        self.volatility_period:int = 36
        self.rolling_period:int = 2
        self.max_missing_days:int = 32
        self.min_lookback_period:int = 24
        self.data:Dict[Symbol, data_tools.SymbolData] = {}
        self.custom_data_df:Dict[str, DataFrame] = {}
        self.measures:Dict[Symbol, Tuple[int, float]] = {}
        self.custom_data_index:List[str] = ['IND_PRO_countries', 'goods_export_countries', 'goods_import_countries']
        self.custom_data_perc:List[str] = ['CPI_countries', 'LTIR_countries', 'STIR_countries', 'UNEMPLOYMENT_countries']
        # create dataframes from custom data
        for variable in self.custom_data_index + self.custom_data_perc:
            load:str = self.Download(f'data.quantpedia.com/backtesting_data/economic/{variable}.csv')
            df:dataframe = pd.read_csv(StringIO(load), delimiter=';')
            df['TIME'] = pd.to_datetime(df['TIME']).dt.date
            df.set_index('TIME', inplace=True)
            self.custom_data_df[variable] = df
        
        # calculate yield sread and trade balance of countries
        self.custom_data_df['yield_spread'] = self.custom_data_df['LTIR_countries'] - self.custom_data_df['STIR_countries']
        self.custom_data_df['trade_balance'] = (self.custom_data_df['goods_export_countries'] - self.custom_data_df['goods_import_countries']) \
             / (self.custom_data_df['goods_export_countries'] + self.custom_data_df['goods_import_countries'])
        self.custom_data_df = {k:v for k, v in self.custom_data_df.items() if k not in ['goods_export_countries', 'goods_import_countries']}
        self.symbols:Dict[str, str] = {"AUDUSD": 'AUS', "GBPUSD": 'GBR', "CADUSD": 'CAN', "EURUSD": 'EU', "JPYUSD": 'JPN', "NOKUSD": 'NOR', "SEKUSD": 'SWE', "NZDUSD": 'NZL', "CHFUSD": 'CHE'}
        
        # data subscription
        for symbol in self.symbols:
            data:Security = self.AddForex(symbol, Resolution.Daily, Market.Oanda)
            data.SetFeeModel(data_tools.CustomFeeModel())
            data.SetLeverage(self.leverage)
            self.data[data.Symbol] = data_tools.SymbolData(self.rolling_period)
        self.selection_flag:bool = False            
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.recent_month:int = -1
    def OnData(self, data: Slice) -> None:
        # monthly rebalance
        if self.recent_month == self.Time.month:
            return
        self.recent_month = self.Time.month
        
        # store monthly prices
        for symbol, symbol_data in self.data.items():
            if symbol in data and data[symbol]:
                symbol_data.update_price(data[symbol].Price)
        look_back:DateTime.date = self.Time.date() - relativedelta(months=self.period)
        current_date:DateTime.date = self.Time.date()
        currency_returns:Dict[Symbol, float] = {self.symbols[sym.Value] : data.get_return() for sym, data in self.data.items() if data.is_ready()}
        if len(currency_returns) == 0:
            return
        returns_df:dataframe = pd.dataframe([currency_returns])
        for i in range(self.min_lookback_period, self.period):
            measure_df:dataframe = pd.dataframe()
            weights_df:dataframe = pd.dataframe()
            for data_title, df in self.custom_data_df.items():
                total_performance:float = 0
                df:dataframe = df.loc[df.index < current_date]
                # check if current data are in dataframe
                if (current_date - df.iloc[-1].name).days >= self.max_missing_days:
                    if data_title in self.measures:
                        self.measures.pop(data_title)
                    continue
                
                # shifted values for measures
                measure:float = np.log(df.iloc[-2, df.columns.isin(list(returns_df.columns))]) - np.log(df.iloc[-i - 1, df.columns.isin(list(returns_df.columns))]) \
                                if data_title in self.custom_data_index else df.iloc[-2, df.columns.isin(list(returns_df.columns))] - df.iloc[-i - 1, df.columns.isin(list(returns_df.columns))]
                
                # rank based on measures
                df_ranks:dataframe = pd.dataframe([measure.rank()]).dropna(axis=1)
                measure_df = returns_df
                for country in df_ranks:
                    weight:float = (df_ranks[country] - df_ranks.mean(axis=1)) / ((df_ranks.max(axis=1) - df_ranks.min(axis=1)) / 2)
                    if country in returns_df:
                        country_performance = returns_df[country] * weight
                        total_performance += country_performance
                    weights_df[country] = weight
                if data_title not in self.measures:
                    self.measures[data_title] = {}
                if i not in self.measures[data_title]:
                    self.measures[data_title][i] = deque(maxlen=self.volatility_period)
                self.measures[data_title][i].append((total_performance, weights_df))
        if self.Time.month % 3 != 0: 
            return
        traded_portfolio_portion:Dict[str, float] = {}
        if any([any([len(data) == data.maxlen for lb, data in value.items()]) for key, value in self.measures.items()]):
            aggregated_inverse_volatility:float = sum([sum([1 / np.std(list(map(lambda x:x[0][0], v))) for i,v in y.items()]) for x, y in self.measures.items()])
            for variable, lookback in self.measures.items():
                for lb, perf_weights in lookback.items():
                    weight:float = (1 / np.std(list(map(lambda x:x[0][0], list(perf_weights))))) / aggregated_inverse_volatility
                    for country in perf_weights[-1][1]:
                        portion:float = (self.Portfolio.TotalPortfolioValue / len(self.measures) / len(lookback) / len(perf_weights[-1][1])) * weight 
                        if np.isnan(portion):
                            continue
                        if country not in traded_portfolio_portion:
                            traded_portfolio_portion[country] = 0
                        traded_portfolio_portion[country] += portion
        # trade execution
        invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
        for symbol in invested:
            if self.symbols[symbol.Value] not in traded_portfolio_portion:
                self.Liquidate(symbol)
        for symbol in list(self.data.keys()):
            if symbol in data and data[symbol]:
                if self.symbols[symbol.Value] in traded_portfolio_portion:
                    quantity:float = (traded_portfolio_portion[self.symbols[symbol.Value]] // data[symbol].Price) - self.Portfolio[symbol].Quantity
                    self.MarketOrder(symbol, quantity)