Curvature Factor in Currencies

1import data_tools
2from typing import List, Dict
3from AlgorithmImports import *
4class CurvatureFactorInCurrencies(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8self.data:Dict[Symbol, SymbolData] = {}
9self.period:int = 21
10self.number_of_currencies:int = 3
11self.leverage:int = 5
12# Symbols - Currency futures, 10Y bond yield, 5Y bond yield, cash rate data.
13# Cash rate source: https://www.quandl.com/data/OECD-Organisation-for-Economic-Co-operation-and-Development
14self.symbols:List[str] = [
15                ('CME_AD1', 'AU10YT', 'AU5YT', 'IR3TIB01AUM156N'), # Australian Dollar Futures, Continuous Contract #1
16                ('CME_CD1', 'CA10YT', 'CA5YT', 'IR3TIB01CAM156N'), # Canadian Dollar Futures, Continuous Contract #1
17                ('CME_SF1', 'CH10YT', 'CH5YT', 'IR3TIB01CHM156N'), # Swiss Franc Futures, Continuous Contract #1
18                ('CME_EC1', 'DE10YT', 'DE5YT', 'IR3TIB01EZM156N'), # Euro FX Futures, Continuous Contract #1
19                ('CME_BP1', 'GB10YT', 'GB5YT', 'LIOR3MUKM'),       # British Pound Futures, Continuous Contract #1
20                ('CME_JY1', 'JP10YT', 'JP5YT', 'IR3TIB01JPM156N'), # Japanese Yen Futures, Continuous Contract #1
21                ('CME_NE1', 'NZ10YT', 'NZ5YT', 'IR3TIB01NZM156N'), # New Zealand Dollar Futures, Continuous Contract #1
22                ('CME_MP1', 'MX10YT', 'MX5YT', 'IR3TIB01MXM156N')  # Mexican Peso Futures, Continuous Contract #1
23                ]
24                
25for currency_future, bond_yield_symbol_10, bond_yield_symbol_5, cash_rate_symbol in self.symbols:
26    # Currency futures data.
27    data = self.AddData(data_tools.QuantpediaFutures, currency_future, Resolution.Daily)
28    data.SetFeeModel(data_tools.CustomFeeModel())
29    data.SetLeverage(self.leverage)
30    
31    # Bond yield data.
32    self.AddData(data_tools.QuantpediaBondYield, bond_yield_symbol_10, Resolution.Daily)
33    self.AddData(data_tools.QuantpediaBondYield, bond_yield_symbol_5, Resolution.Daily)
34    
35    # Interbank rate data.
36    self.AddData(data_tools.InterestRate3M, cash_rate_symbol, Resolution.Daily)
37    
38for symbol_tuple in self.symbols:
39    self.data[symbol_tuple[0]] = data_tools.SymbolData(1) # Need to check data for trading  
40    for symbol in symbol_tuple[1:]:
41        self.data[symbol] = data_tools.SymbolData(self.period)
42
43self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
44self.current_month:int = -1
45def OnData(self, data:Slice) -> None:
46for symbol_tuple in self.symbols:
47    for symbol in symbol_tuple:
48        if symbol in data and data[symbol]:
49            price:float = data[symbol].Value
50            self.data[symbol].update(price)
51if self.Time.month == self.current_month:
52    return
53self.current_month = self.Time.month
54ir_last_update_date:Dict[str, datetime.date] = data_tools.InterestRate3M.get_last_update_date()
55qp_futures_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
56bond_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaBondYield.get_last_update_date()
57curvature = {}
58for currency_future, bond_yield_symbol_10, bond_yield_symbol_5, cash_rate_symbol in self.symbols:
59    # check if data is still coming
60    if (self.Securities[currency_future].GetLastData() and qp_futures_last_update_date[currency_future] > self.Time.date()) \
61        and (self.Securities[cash_rate_symbol].GetLastData() and ir_last_update_date[cash_rate_symbol] > self.Time.date()) \
62        and (self.Securities[bond_yield_symbol_10].GetLastData() and bond_last_update_date[bond_yield_symbol_10] > self.Time.date()) \
63        and (self.Securities[bond_yield_symbol_5].GetLastData() and bond_last_update_date[bond_yield_symbol_5] > self.Time.date()):
64        if self.data[currency_future].is_ready() and self.data[bond_yield_symbol_10].is_ready() \
65            and self.data[bond_yield_symbol_5].is_ready() and self.data[cash_rate_symbol].is_ready():
66            # curvature = (5y - 3m) - (10y - 5y)  
67            first_bracket:float = self.data[bond_yield_symbol_5].performance() - self.data[cash_rate_symbol].performance()
68            second_bracket:float = self.data[bond_yield_symbol_10].performance() - self.data[bond_yield_symbol_5].performance()
69            curvature[currency_future] = first_bracket - second_bracket
70
71long:List[Symbol] = []
72short:List[Symbol] = []
73if len(curvature) >= self.number_of_currencies * 2:
74    sorted_by_curvature:List[Symbol] = [x[0] for x in sorted(curvature.items(), key=lambda item: item[1], reverse=True)]
75    long = sorted_by_curvature[:self.number_of_currencies]
76    short = sorted_by_curvature[-self.number_of_currencies:]
77currencies_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
78for symbol in currencies_invested:
79    if symbol not in long + short: 
80        self.Liquidate(symbol)
81        
82for symbol in long:
83    if symbol in data and data[symbol]:
84        self.SetHoldings(symbol, 1 / len(long))
85    
86for symbol in short:
87    if symbol in data and data[symbol]:
88        self.SetHoldings(symbol, -1 / len(short))