Growth Gap Factor in Fixed Income

1import numpy as np
2from AlgorithmImports import *
3import data_tools
4from typing import Dict, List
5class GrowthGapFactorinFixedIncome(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(1990, 1, 1)
8self.SetCash(100000)
9
10# Bond symbol and GDP symbol. (GDP at current prices)
11self.symbols = {
12    "ASX_XT1" : "AUS_GDP",        # 10 Year Commonwealth Treasury Bond Futures, Continuous Contract #1 (Australia)
13    "MX_CGB1" : "CAN_GDP",        # Ten-Year Government of Canada Bond Futures, Continuous Contract #1 (Canada)
14    "EUREX_FGBL1" : "DEU_GDP",    # Euro-Bund (10Y) Futures, Continuous Contract #1 (Germany)
15    "LIFFE_R1" : "GBR_GDP",       # Long Gilt Futures, Continuous Contract #1 (U.K.)
16    "CME_TY1" : "USA_GDP"         # 10 Yr Note Futures, Continuous Contract #1 (USA)
17}
18
19# Yearly GDP data used for SMA.
20self.data:Dict[str, float] = {}
21self.sma_period:int = 5
22self.leverage:int = 3
23
24for bond_future, gdp_symbol in self.symbols.items():
25    # Futures data.
26    data = self.AddData(data_tools.QuantpediaFutures, bond_future, Resolution.Daily)
27    data.SetFeeModel(data_tools.CustomFeeModel())
28    data.SetLeverage(self.leverage)
29    self.data[gdp_symbol] = RollingWindow[float](self.sma_period)
30    
31    # Bond yield data.
32    self.AddData(data_tools.GDPData, gdp_symbol, Resolution.Daily)
33self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
34
35def OnData(self, data: Slice) -> None:
36trade_flag:bool = False
37gdp_last_update_date:Dict[str, datetime.date] = data_tools.GDPData.get_last_update_date()
38future_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
39symbols_to_delete:List[str] = []
40# store yearly gdp data
41for bond_future, gdp_symbol in self.symbols.items():
42    # data is still coming
43    if self.Securities[bond_future].GetLastData() and self.Time.date() > future_last_update_date[bond_future] \
44        or self.Securities[gdp_symbol].GetLastData() and self.Time.date() > gdp_last_update_date[gdp_symbol]:
45        symbols_to_delete.append(bond_future)
46        continue
47    if gdp_symbol in data and data[gdp_symbol]:
48        gdp:float = data[gdp_symbol].Value
49        self.data[gdp_symbol].Add(gdp)
50        trade_flag = True
51    
52if len(symbols_to_delete) != 0:
53    for symbol in symbols_to_delete:
54        self.symbols.pop(symbol)
55# rebalance once the new data arrived
56if not trade_flag: 
57    return
58
59# SMA gap
60sma_gap:Dict[str, float] = { x[0] : ((self.Securities[x[1]].Price - np.average([gdp for gdp in self.data[x[1]]])) / np.average([gdp for gdp in self.data[x[1]]])) for x in self.symbols.items()
61                                    if self.Securities.ContainsKey(x[1]) and x[1] in self.data and self.data[x[1]].IsReady and self.Securities[x[0]].GetLastData() and (self.Time.date() - self.Securities[x[0]].GetLastData().Time.date()).days < 5}
62weight:Dict[str, float] = {}
63if len(sma_gap) != 0:
64    avg_gap:float = np.average([x[1] for x in sma_gap.items()])
65    
66    avg_gap_diff:Dict[str, float] = {x[0] : x[1] - avg_gap for x in sma_gap.items()}
67    total_avg_gap_diff:float = sum([abs(x[1]) for x in avg_gap_diff.items()])
68    weight = {x[0] : x[1] / total_avg_gap_diff for x in avg_gap_diff.items()}
69
70# trade execution
71invested:List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
72for symbol in invested:
73    if symbol not in weight:
74        self.Liquidate(symbol)
75for symbol, w in weight.items():
76    if symbol in data and data[symbol]:
77        self.SetHoldings(symbol, w)