Sovereign CDS Predicts FX Market Return

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