Sovereign CDS Currency Factor

1from AlgorithmImports import *
2import numpy as np
3#endregion
4
5class SovereignCDSCurrencyFactor(QCAlgorithm):
6
7def Initialize(self):
8 self.SetStartDate(2008, 1, 1)
9 self.SetCash(100000)
10 
11 # forex pair symbol : (CDS country symbol, GDP symbol, long-short switch position flag)
12 self.symbols:Dict[str, Tuple[List[str], bool]] = {
13     'AUDUSD' : (['AU'], False),
14     'USDCAD' : (['CA'], True),
15     'EURUSD' : (['ES', 'FR', 'IT', 'GR'], False),
16     'GBPUSD' : (['GB'], False),
17     'USDMXN' : (['MX'], True),
18     'USDTRY' : (['TR'], True),
19     'RUBUSD' : (['RU'], False),
20     'BRLUSD' : (['BR'], False),
21 }
22
23 self.cds_symbols:Dict[str, tuple] = {}
24
25 for fx_symbol, (country_codes, _) in self.symbols.items():
26     # subscribe forex symbol
27     data = self.AddForex(fx_symbol, Resolution.Minute, Market.Oanda)
28     data.SetLeverage(5)
29     
30     # subscribe CDS symbols
31     for country_code in country_codes:
32         cds_symbol:Symbol = self.AddData(CDSData5Y, country_code, Resolution.Daily).Symbol
33         self.cds_symbols[country_code] = cds_symbol
34 
35 self.quantile:int = 3
36 self.recent_month:int = -1
37 
38def OnData(self, data):
39 if self.Time.month == self.recent_month:
40     return
41 self.recent_month = self.Time.month
42
43 # store actual CDS
44 actual_cds:Dict[str, float] = {}
45
46 # end of custom data
47 last_update_date_5Y:Dict[str, datetime.date] = CDSData5Y.get_last_update_date()
48
49 for fx_symbol, (country_codes, _) in self.symbols.items():
50     # price data are available
51     if fx_symbol in data and data[fx_symbol]:
52         cds_values:List[float] = []
53         for country_code in country_codes:
54             # CDS data are available
55             if self.Securities[self.cds_symbols[country_code]].GetLastData():
56                 if self.Time.date() <= last_update_date_5Y[self.cds_symbols[country_code]]:
57                     # get most recent CDS value
58                     cds:float = self.Securities[self.cds_symbols[country_code]].Price
59                     cds_values.append(cds)
60         
61         if len(cds_values) != 0:
62             actual_cds[fx_symbol] = np.mean(cds_values)
63 
64 if len(actual_cds) < self.quantile:
65     self.Liquidate()
66     return
67
68 # sort by CDS
69 sorted_by_cds:List = sorted(actual_cds.items(), key = lambda x: x[1], reverse=True)
70 quantile:int = int(len(sorted_by_cds) / self.quantile)
71 
72 # going long quintile Lowest SR and short quintile Highest SR
73 long:List[str] = [x[0] for x in sorted_by_cds[-quantile:]]
74 short:List[str] = [x[0] for x in sorted_by_cds[:quantile]]
75 
76 long_c:int = len(long)
77 short_c:int = len(short)
78 
79 # liquidate
80 invested:List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
81 for symbol in invested:
82     if symbol not in long + short:
83         self.Liquidate(symbol)
84 
85 # EW portfolio
86 for symbol in long:
87     # long-short swap position flag
88     ls_switch:bool = self.symbols[symbol][1]
89     if not ls_switch:
90         self.SetHoldings(symbol, 1 / long_c)
91     else:
92         self.SetHoldings(symbol, -1 / long_c)
93         
94 for symbol in short:
95     # long-short swap position flag
96     ls_switch:bool = self.symbols[symbol][1]
97     if not ls_switch:
98         self.SetHoldings(symbol, -1 / short_c)
99     else:
100         self.SetHoldings(symbol, 1 / short_c)
101
102# 5Y Credit Default Swap data.
103# Source: https://www.investing.com/search/?q=CDS%205%20years&tab=quotes
104# NOTE: IMPORTANT: Data order must be ascending (datewise)
105class CDSData5Y(PythonData):
106def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
107 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/cds/{0}_CDS_5Y.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
108
109_last_update_date:Dict[str, datetime.date] = {}
110
111@staticmethod
112def get_last_update_date() -> Dict[str, datetime.date]:
113return CDSData5Y._last_update_date
114
115def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
116 data = CDSData5Y()
117 data.Symbol = config.Symbol
118 
119 if not line[0].isdigit(): return None
120 split = line.split(';')
121 
122 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
123
124 # store last date of the symbol
125 if data.Symbol not in CDSData5Y._last_update_date:
126     CDSData5Y._last_update_date[data.Symbol] = datetime(1,1,1).date()
127 if data.Time.date() > CDSData5Y._last_update_date[data.Symbol]:
128     CDSData5Y._last_update_date[data.Symbol] = data.Time.date()
129
130 data.Value = float(split[1])
131
132 return data