“该策略交易20种货币对，使用更高的回报矩和历史平均值创建78种策略。选择过去3个月中表现最佳的策略，并每月进行交易。”
资产类别: 差价合约、期货、互换 | 地区: 全球 | 周期: 每月 | 市场: 外汇 | 关键词: 高阶矩
I. 策略概要
该策略涉及20种兑美元的货币对，投资者计算每种货币过去一个月每日数据的多个高阶矩（从第4阶到第100阶）。然后将这些矩与其在不同时间范围（12、24、36、48、60个月和整个样本期）内的平均值进行比较。这产生了78个独立的排序（13个矩乘以6个回顾期），形成了78个不同的策略。每个策略将货币分为五分位数，做多相对于历史水平高阶回报矩较低的货币，做空高阶矩较高的货币。投资者采用自适应方法，选择过去3个月平均回报最高的策略。选定的策略用于形成下一个月的等权重多空货币投资组合。该过程每月重复进行，以捕捉最佳的货币市场低效率。
II. 策略合理性
学术来源论文与研究低风险异常的实证文献相关，其中风险指标基于高阶矩。投资者愿意为具有类似彩票回报的资产支付更高的价格并获得更低的预期回报。
III. 来源论文
货币市场中基于高阶矩的低风险策略 [点击查看论文]
- 克劳迪娅·尊夫特，法兰克福歌德大学 – 金融系；Quoniam资产管理公司。
<摘要>
本文首次为货币市场中低风险异常现象的存在提供了深刻的证据。具体而言，我发现了一种货币远期市场的新型策略，该策略做多相对于过去水平高阶回报矩较低的货币，做空相对于过去水平高阶回报矩较高的货币。与传统货币策略相比，基于高阶矩的低风险策略并未被它们涵盖。在约25年的样本中，它提供了最高的平均超额收益和夏普比率，以及最小的跌幅。该策略的盈利能力无法用标准风险因素和套利限制来解释。
IV. 回测表现
| 年化回报 | 11.07% |
| 波动率 | 8% |
| β值 | -0.016 |
| 夏普比率 | 1.38 |
| 索提诺比率 | -0.37 |
| 最大回撤 | N/A |
| 胜率 | 47% |
V. 完整的 Python 代码
from AlgorithmImports import *
from typing import List, Dict
from dateutil.relativedelta import relativedelta
import data_tools
from enum import Enum
# endregion
class UniverseType(Enum):
QP_FUTURES = 1
FOREX = 2
class TradingbasedonHigherMomentsinCurrencyMarkets(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.universe_type: UniverseType = UniverseType.QP_FUTURES
self.leverage: int = 5
self.quantile: int = 3
self.min_period: int = 3
self.periods: List[int] = [12, 24, 36, 48, 60, 'all']
self.moments: List[int] = [4, 6, 8, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
self.tickers: List[str] = [
'CME_AD1', 'CME_BP1', 'CME_NE1',
'CME_CD1', 'CME_SF1', 'CME_JY1'
]
if self.universe_type == UniverseType.FOREX:
self.tickers = [
'AUDUSD', 'CHFUSD', 'EURUSD', 'GBPUSD', 'NZDUSD', 'USDCAD',
'USDCZK', 'USDDKK', 'USDHUF', 'USDJPY', 'USDMXN', 'USDNOK',
'USDPLN', 'USDSGD', 'USDZAR', 'USDSEK', 'USDTWD', 'USDTRY'
]
self.data: Dict[Symbol, data_tools.SymbolData] = {}
self.strategy_manager: data_tools.StrategyManager = data_tools.StrategyManager(self.periods, self.moments, self.min_period)
for ticker in self.tickers:
security = self.AddData(data_tools.QuantpediaFutures, ticker, Resolution.Daily) if self.universe_type == UniverseType.QP_FUTURES \
else self.AddForex(ticker, Resolution.Daily, Market.Oanda)
if self.universe_type == UniverseType.QP_FUTURES:
security.SetFeeModel(data_tools.CustomFeeModel())
security.SetLeverage(self.leverage)
symbol: Symbol = security.Symbol
self.data[symbol] = data_tools.SymbolData(symbol, self.periods, self.moments)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.recent_month: int = -1
def OnData(self, data: Slice) -> None:
if self.universe_type == UniverseType.QP_FUTURES:
qp_futures_last_update_date: Dict[Symbol, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
rebalance_flag: bool = False
# store daily prices
for symbol, symbol_data in self.data.items():
if symbol in data and data[symbol]:
symbol_data.update_price(data[symbol].Price)
# monthly rebalance
if self.recent_month != self.Time.month:
self.recent_month = self.Time.month
rebalance_flag = True
if not rebalance_flag:
return
for symbol, symbol_data in self.data.items():
if self.universe_type == UniverseType.QP_FUTURES:
if self.Securities[symbol].GetLastData() and self.Time.date() > qp_futures_last_update_date[symbol]:
self.Liquidate()
return
if symbol_data.is_ready():
symbol_data.calculate_moments()
symbol_data._daily_prices.clear()
if symbol_data.moment_is_ready():
symbol_data.compare_moments()
avg_returns: List[Tuple[int, int, float]] = []
if not all(symbol_data.strategy_is_ready() for _, symbol_data in self.data.items()):
return
for period in self.periods:
for i, moment in enumerate(self.moments):
# compare last higher moments with avereage of period
if self.strategy_manager.is_ready():
avg_returns.append((period, moment, self.strategy_manager.get_average_return(period, moment)))
# sort strategies and divide into quantiles
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])
quantile: int = int(len(sorted_strategy) / self.quantile)
long: List[Symbol] = [x[0] for x in sorted_strategy][:quantile]
short: List[Symbol] = [x[0] for x in sorted_strategy][-quantile:]
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])))
if len(avg_returns) >= len(self.periods) * len(self.moments):
sorted_avg: List[Tuple[int, int, float]] = sorted(avg_returns, key=lambda x: x[2], reverse=True)
period, moment, _ = sorted_avg[0]
long: List[Symbol] = self.strategy_manager.get_long_symbols(period, moment)
short: List[Symbol] = self.strategy_manager.get_short_symbols(period, moment)
# order execution
targets:List[PortfolioTarget] = []
for i, portfolio in enumerate([long, short]):
for symbol in portfolio:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, (((-1) ** i) / len(portfolio)) * self.data[symbol].trade_direction))
self.SetHoldings(targets, True)