Trading based on Higher Moments in Currency Markets

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)