A Multi Strategy Approach to Trading Foreign Exchange Futures

1from AlgorithmImports import *
2import numpy as np
3from typing import List, Dict, Tuple
4from scipy import stats
5class AMultiStrategyApproachtoTradingForeignExchangeFutures(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9
10# Symbols - Currency future, equity future, 10Y bond yield, cash rate data.
11# Cash rate source: https://www.quandl.com/data/OECD-Organisation-for-Economic-Co-operation-and-Development
12# 10Y bond yield source: www.investing.com
13self.symbols:List[Tuple[str, str, str, str]] = [
14    ('CME_AD1', 'ASX_YAP1', 'AU10YT', 'IR3TIB01AUM156N'),    # Australian Dollar Futures, Continuous Contract #1
15    ('CME_CD1', 'LIFFE_FCE1', 'CA10YT', 'IR3TIB01CAM156N'),  # Canadian Dollar Futures, Continuous Contract #1
16    ('CME_SF1', 'EUREX_FSMI1', 'CH10YT', 'IR3TIB01CHM156N'), # Swiss Franc Futures, Continuous Contract #1
17    ('CME_EC1', 'EUREX_FSTX1', 'DE10YT', 'IR3TIB01EZM156N'), # Euro FX Futures, Continuous Contract #1
18    ('CME_BP1', 'LIFFE_Z1', 'GB10YT', 'LIOR3MUKM'),          # British Pound Futures, Continuous Contract #1
19    ('CME_JY1', 'SGX_NK1', 'JP10YT', 'IR3TIB01JPM156N'),     # Japanese Yen Futures, Continuous Contract #1
20]
21            
22# Symbol data.
23self.data:Dict[Symbol, SymbolData] = {}
24self.short_period:int = 3
25self.long_period:int = 12
26self.leverage:int = 20
27
28# Target risk of the allocation.
29self.target_risk:float = 0.1
30
31for currency_future, equity_future, bond_yield_symbol, cash_rate_symbol in self.symbols:
32    # Currency future data.
33    data = self.AddData(data_tools.QuantpediaFutures, currency_future, Resolution.Daily)
34    data.SetFeeModel(data_tools.CustomFeeModel())
35    data.SetLeverage(self.leverage)
36    self.data[currency_future] = data_tools.SymbolData(currency_future, self, self.short_period*21, self.long_period*21, True)
37    
38    # Equity future data.
39    self.AddData(data_tools.QuantpediaFutures, equity_future, Resolution.Daily)
40    self.data[equity_future] = data_tools.SymbolData(equity_future, self, self.short_period*21, self.long_period*21, False)
41    
42    # Bond yield data.
43    self.AddData(data_tools.QuantpediaBondYield, bond_yield_symbol, Resolution.Daily)
44    # Interbank rate data.
45    self.AddData(data_tools.InterestRate3M, cash_rate_symbol, Resolution.Daily)
46
47self.usa_10Y_yield:str = 'US10Y'
48self.usa_cash_rate:str = 'IR3TIB01USM156N'
49self.AddData(data_tools.QuantpediaBondYield, self.usa_10Y_yield, Resolution.Daily)
50self.AddData(data_tools.InterestRate3M, self.usa_cash_rate, Resolution.Daily)
51self.commodity_index:Symbol = self.AddEquity('DBC', Resolution.Daily).Symbol
52self.data[self.commodity_index] = data_tools.SymbolData(self.commodity_index, self, self.short_period*21, self.long_period*21, False)
53
54self.last_month:int = -1
55
56def OnData(self, data:Slice) -> None:
57# Rebalance once a month.
58if self.last_month != self.Time.month:
59    self.last_month = self.Time.month
60    ir_last_update_date:Dict[str, datetime.date] = data_tools.InterestRate3M.get_last_update_date()
61    qp_futures_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
62    
63    # Create indicators.
64    weight:Dict[Symbol, float] = {}
65    for currency_future, equity_future, bond_yield_symbol, cash_rate_symbol in self.symbols:
66        # data is still coming
67        if self.Securities[currency_future].GetLastData() and qp_futures_last_update_date[currency_future] <= self.Time.date() \
68            or self.Securities[cash_rate_symbol].GetLastData() and ir_last_update_date[cash_rate_symbol] <= self.Time.date():
69            continue
70        # If data needed is ready (Momentum and SMA's).
71        if self.data[currency_future].is_ready() and self.data[equity_future].is_ready() and self.data[self.commodity_index].is_ready():
72            
73            long_interest_rate_carry:Union[float,None] = self.Securities[bond_yield_symbol].Price - self.Securities[self.usa_10Y_yield].Price if self.Securities.ContainsKey(bond_yield_symbol) and self.Securities.ContainsKey(self.usa_10Y_yield) else None
74            short_interest_rate_carry:Union[float,None] = self.Securities[cash_rate_symbol].Price - self.Securities[self.usa_cash_rate].Price if self.Securities.ContainsKey(cash_rate_symbol) and self.Securities.ContainsKey(self.usa_cash_rate) else None
75            
76            curr_short_term_momentum:float = self.data[currency_future].short_term_momentum()
77            curr_long_term_momentum:float = self.data[currency_future].long_term_momentum()
78            
79            curr_long_term_reversal:Union[float,None] =  self.Securities[currency_future].Price / self.data[currency_future].long_term_sma() if self.Securities.ContainsKey(currency_future) else None
80            curr_short_term_reversal:Union[float,None] =  self.Securities[currency_future].Price / self.data[currency_future].short_term_sma() if self.Securities.ContainsKey(currency_future) else None
81            equity_short_term_momentum:float = self.data[equity_future].short_term_momentum()
82            equity_long_term_momentum:float = self.data[equity_future].long_term_momentum()
83            
84            commodity_short_term_momentum:float = self.data[self.commodity_index].short_term_momentum()
85            commodity_long_term_momentum:float = self.data[self.commodity_index].long_term_momentum()
86            
87            if long_interest_rate_carry and short_interest_rate_carry and curr_long_term_reversal and curr_short_term_reversal:
88                indicator_value:Tuple[float] = (long_interest_rate_carry, short_interest_rate_carry, curr_long_term_momentum, curr_short_term_momentum,    \
89                                                curr_long_term_reversal, curr_short_term_reversal, equity_long_term_momentum, equity_short_term_momentum,  \
90                                                commodity_long_term_momentum, commodity_short_term_momentum)
91                
92                self.data[currency_future].add_indicator_value(indicator_value)
93                
94                if self.data[currency_future].indicator_history_is_ready():
95                    num_of_properties:int = len(indicator_value)
96                    scores:List[float] = []
97                    for propert_index in range(num_of_properties):
98                        scores.append(abs((stats.percentileofscore([x[propert_index] for x in self.data[currency_future].IndicatorsHistory], indicator_value[propert_index]) / 100 ) - 0.5))
99                    
100                    score_sum:float = sum(scores)
101                    scores:np.ndarray = np.array(scores)
102                    scores = (scores / score_sum) * self.target_risk
103                    risk_budget:float = sum(scores)
104                    weight[currency_future] = risk_budget
105    
106    # Trade execution.
107    invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
108    for symbol in invested:
109        if symbol not in weight:
110            self.Liquidate(symbol)
111            
112    for symbol, w in weight.items():
113        self.SetHoldings(symbol, w)