Flight to Quality Factor in Fixed Income

1import numpy as np
2from AlgorithmImports import *
3class FlighttoQualityFactor(QCAlgorithm):
4def Initialize(self):
5 self.SetStartDate(2000, 1, 1)
6 self.SetCash(100000)
7 
8 self.symbols = {
9                 "EWA" : "ASX_XT1",        # 10 Year Commonwealth Treasury Bond Futures, Continuous Contract #1 (Australia)
10                 "EWC" : "MX_CGB1",        # Ten-Year Government of Canada Bond Futures, Continuous Contract #1 (Canada)
11                 "EWG" : "EUREX_FGBL1",    # Euro-Bund (10Y) Futures, Continuous Contract #1 (Germany)
12                 "EWJ" : "SGX_JB1",        # SGX 10-Year Mini Japanese Government Bond Futures, Continuous Contract #1 (Japan)
13                 "EWU" : "LIFFE_R1",       # Long Gilt Futures, Continuous Contract #1 (U.K.)
14                 "SPY" : "CME_TY1"         # 10 Yr Note Futures, Continuous Contract #1 (USA)
15                 }
16 self.data = {}
17 self.period = 12 * 21
18 leverage: int = 2
19 self.SetWarmUp(self.period)
20 for symbol in self.symbols:
21     bond = self.symbols[symbol]
22     
23     self.AddEquity(symbol, Resolution.Daily)
24     self.data[symbol] = RollingWindow[float](self.period)
25     data = self.AddData(QuantpediaFutures, bond, Resolution.Daily)
26     data.set_leverage(leverage)
27     data.SetFeeModel(CustomFeeModel())
28     
29def OnData(self, data):
30 for symbol, bond in self.symbols.items():
31     if self.securities[bond].get_last_data() and self.time.date() > QuantpediaFutures.get_last_update_date()[bond]:
32         self.liquidate(bond)
33         self.data[symbol].reset()
34         continue
35     if symbol in data and data[symbol]:
36         price = data[symbol].Value
37         self.data[symbol].Add(price)
38 if self.IsWarmingUp: return
39
40 volatility = {}
41 for symbol in self.symbols:
42     if self.data[symbol].IsReady:
43         prices = np.array([x for x in self.data[symbol]])
44         returns = prices[:-1] / prices[1:] - 1
45         volatility[symbol] = np.std(returns)
46 
47 if len(volatility) <= 1: return
48 
49 avg_volatility = np.mean([x[1] for x in volatility.items()])
50 
51 diff = {symbol : (volatility[symbol] - avg_volatility) for symbol in volatility}
52 total_diff = sum([abs(x[1]) for x in diff.items()])
53 
54 for symbol in diff:
55     bond = self.symbols[symbol]
56     if data.contains_key(bond) and data[bond]:
57         self.SetHoldings(bond, diff[symbol] / total_diff)
58# Custom fee model.
59class CustomFeeModel(FeeModel):
60def GetOrderFee(self, parameters):
61 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
62 return OrderFee(CashAmount(fee, "USD"))
63# Quantpedia data.
64# NOTE: IMPORTANT: Data order must be ascending (datewise)
65class QuantpediaFutures(PythonData):
66_last_update_date:Dict[Symbol, datetime.date] = {}
67@staticmethod
68def get_last_update_date() -> Dict[Symbol, datetime.date]:
69return QuantpediaFutures._last_update_date
70def GetSource(self, config, date, isLiveMode):
71 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
72def Reader(self, config, line, date, isLiveMode):
73 data = QuantpediaFutures()
74 data.Symbol = config.Symbol
75 
76 if not line[0].isdigit(): return None
77 split = line.split(';')
78 
79 data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
80 data['back_adjusted'] = float(split[1])
81 data['spliced'] = float(split[2])
82 data.Value = float(split[1])
83 if config.Symbol.Value not in QuantpediaFutures._last_update_date:
84     QuantpediaFutures._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
85 if data.Time.date() > QuantpediaFutures._last_update_date[config.Symbol.Value]:
86     QuantpediaFutures._last_update_date[config.Symbol.Value] = data.Time.date()
87 return data