外汇 · 全球 · 差价合约、期货、互换 · 每月 · 全球 Free

基于高阶矩在外汇市场中的交易

登录后收藏

学术论文

A Low-Risk Strategy Based on Higher Moments in Currency Markets

作者货币市场中基于高阶矩的低风险策略 [点击查看论文]

机构

DEGoethe University Frankfurt
?Goethe University Frankfurt - Department of Finance
?Quoniam Asset Management GmbH

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

策略概要

该策略涉及20种兑美元的货币对，投资者计算每种货币过去一个月每日数据的多个高阶矩（从第4阶到第100阶）。然后将这些矩与其在不同时间范围（12、24、36、48、60个月和整个样本期）内的平均值进行比较。这产生了78个独立的排序（13个矩乘以6个回顾期），形成了78个不同的策略。每个策略将货币分为五分位数，做多相对于历史水平高阶回报矩较低的货币，做空高阶矩较高的货币。投资者采用自适应方法，选择过去3个月平均回报最高的策略。选定的策略用于形成下一个月的等权重多空货币投资组合。该过程每月重复进行，以捕捉最佳的货币市场低效率。

II. 策略合理性

学术

回测表现

波动率8%

夏普比率1.38

索提诺比率-0.37

胜率47%

完整 Python 代码

1from AlgorithmImports import *
2from typing import List, Dict
3from dateutil.relativedelta import relativedelta
4import data_tools
5from enum import Enum
6# endregion
7class UniverseType(Enum):
8QP_FUTURES = 1
9FOREX = 2
10class TradingbasedonHigherMomentsinCurrencyMarkets(QCAlgorithm):
11def Initialize(self) -> None:
12self.SetStartDate(2000, 1, 1)
13self.SetCash(100000)
14self.universe_type: UniverseType = UniverseType.QP_FUTURES
15self.leverage: int = 5
16self.quantile: int = 3
17self.min_period: int = 3
18self.periods: List[int] = [12, 24, 36, 48, 60, 'all']
19self.moments: List[int] = [4, 6, 8, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
20self.tickers: List[str] = [
21    'CME_AD1', 'CME_BP1', 'CME_NE1',
22    'CME_CD1', 'CME_SF1', 'CME_JY1'
23]
24if self.universe_type == UniverseType.FOREX:
25    self.tickers = [
26        'AUDUSD', 'CHFUSD', 'EURUSD', 'GBPUSD', 'NZDUSD', 'USDCAD', 
27        'USDCZK', 'USDDKK', 'USDHUF', 'USDJPY', 'USDMXN', 'USDNOK', 
28        'USDPLN', 'USDSGD', 'USDZAR', 'USDSEK', 'USDTWD', 'USDTRY'
29    ]
30self.data: Dict[Symbol, data_tools.SymbolData] = {}
31self.strategy_manager: data_tools.StrategyManager = data_tools.StrategyManager(self.periods, self.moments, self.min_period)
32for ticker in self.tickers:
33    security = self.AddData(data_tools.QuantpediaFutures, ticker, Resolution.Daily) if self.universe_type == UniverseType.QP_FUTURES \
34            else self.AddForex(ticker, Resolution.Daily, Market.Oanda)
35    
36    if self.universe_type == UniverseType.QP_FUTURES:
37        security.SetFeeModel(data_tools.CustomFeeModel())
38    security.SetLeverage(self.leverage)
39    
40    symbol: Symbol = security.Symbol
41    self.data[symbol] = data_tools.SymbolData(symbol, self.periods, self.moments)
42self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
43self.recent_month: int = -1
44def OnData(self, data: Slice) -> None:
45if self.universe_type == UniverseType.QP_FUTURES:
46    qp_futures_last_update_date: Dict[Symbol, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
47rebalance_flag: bool = False
48# store daily prices
49for symbol, symbol_data in self.data.items():
50    if symbol in data and data[symbol]:
51        symbol_data.update_price(data[symbol].Price)
52# monthly rebalance
53if self.recent_month != self.Time.month:
54    self.recent_month = self.Time.month
55    rebalance_flag = True
56if not rebalance_flag:
57    return
58for symbol, symbol_data in self.data.items():
59    if self.universe_type == UniverseType.QP_FUTURES:
60        if self.Securities[symbol].GetLastData() and self.Time.date() > qp_futures_last_update_date[symbol]:
61            self.Liquidate()
62            return
63    if symbol_data.is_ready():
64        symbol_data.calculate_moments()
65        symbol_data._daily_prices.clear()
66    
67    if symbol_data.moment_is_ready():
68        symbol_data.compare_moments()
69avg_returns: List[Tuple[int, int, float]] = []
70if not all(symbol_data.strategy_is_ready() for _, symbol_data in self.data.items()):
71    return
72for period in self.periods:
73    for i, moment in enumerate(self.moments):
74        # compare last higher moments with avereage of period
75        if self.strategy_manager.is_ready():
76            avg_returns.append((period, moment, self.strategy_manager.get_average_return(period, moment)))
77        # sort strategies and divide into quantiles
78        sorted_strategy: List[Tuple[List, List, float]] = sorted({symbol: symbol_data.get_moment(period)[i] for symbol, symbol_data in self.data.items()}.items(), key=lambda x: x[1])
79        quantile: int = int(len(sorted_strategy) / self.quantile)
80        long: List[Symbol] = [x[0] for x in sorted_strategy][:quantile]
81        short: List[Symbol] = [x[0] for x in sorted_strategy][-quantile:]
82        self.strategy_manager.update_data(period, moment, (long, short, sum([self.data[symbol].get_last_month_return() for symbol in long]) - sum([self.data[symbol].get_last_month_return() for symbol in short])))
83
84if len(avg_returns) >= len(self.periods) * len(self.moments):
85    sorted_avg: List[Tuple[int, int, float]] = sorted(avg_returns, key=lambda x: x[2], reverse=True)
86    period, moment, _ = sorted_avg[0]
87    long: List[Symbol] = self.strategy_manager.get_long_symbols(period, moment)
88    short: List[Symbol] = self.strategy_manager.get_short_symbols(period, moment)
89    # order execution
90    targets:List[PortfolioTarget] = []
91    for i, portfolio in enumerate([long, short]):
92        for symbol in portfolio:
93            if symbol in data and data[symbol]:
94                targets.append(PortfolioTarget(symbol, (((-1) ** i) / len(portfolio)) * self.data[symbol].trade_direction))
95    
96    self.SetHoldings(targets, True)