Term Spread and Term Premium Predict US Government Bonds Returns

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3# endregion
4
5class TermSpreadandTermPremiumPredictUSGovernmentBondsReturns(QCAlgorithm):
6
7def Initialize(self):
8 self.SetStartDate(2007, 6, 1) # BIL inception
9 self.SetCash(100000)
10
11 self.long_duration_bond:Symbol = self.AddEquity('IEF', Resolution.Daily).Symbol
12 self.short_duration_bond:Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
13 self.term_spread:Symbol = self.AddData(TermSpread, 'T10Y3M', Resolution.Daily).Symbol
14
15 self.period:int = 40 * 21
16 self.rebalance_flag:bool = False
17
18 self.term_spread_sma = self.SMA(self.term_spread, self.period, Resolution.Daily)
19 self.SetWarmup(self.period, Resolution.Daily)
20
21 self.recent_month:int = -1
22
23def OnData(self, data: Slice) -> None:
24 if self.IsWarmingUp:
25     return
26
27 t10y3m_last_update_date:datetime.date = TermSpread._last_update_date
28
29 # check if custom data is still arriving
30 if self.Securities[self.term_spread].GetLastData() and self.Time.date() >= t10y3m_last_update_date:
31     self.Liquidate()
32     return
33
34 # rebalance monthly
35 if self.Time.month == self.recent_month:
36     return
37 self.recent_month = self.Time.month
38
39 # compare latest value with 40-month moving average
40 traded_asset:Symbol|None = None
41 risk_flag:bool = False
42
43 if self.term_spread in data and data[self.term_spread]:
44     if data[self.term_spread].Price < self.term_spread_sma.Current.Value:
45         traded_asset = self.long_duration_bond
46     else:
47         traded_asset = self.short_duration_bond
48
49 # trade execution
50 if all(x in data and data[x] for x in [self.long_duration_bond, self.short_duration_bond]):
51     if not self.Portfolio[traded_asset].Invested:
52         self.Liquidate()
53         self.SetHoldings(traded_asset, 1)
54
55# Source: https://fred.stlouisfed.org/series/T10Y3M
56class TermSpread(PythonData):
57def GetSource(self, config, date, isLiveMode):
58 return SubscriptionDataSource('data.quantpedia.com/backtesting_data/economic/T10Y3M.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
59
60_last_update_date:datetime.date = datetime(1,1,1).date()
61
62@staticmethod
63def get_last_update_date() -> Dict[Symbol, datetime.date]:
64return TermSpread._last_update_date
65
66def Reader(self, config, line, date, isLiveMode):
67 data = TermSpread()
68 data.Symbol = config.Symbol
69
70 if not line[0].isdigit(): return None
71 split = line.split(';')
72 
73 # Parse the CSV file's columns into the custom data class
74 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
75 if split[1] != '.':
76     data.Value = float(split[1])
77
78 if data.Time.date() > TermSpread._last_update_date:
79     TermSpread._last_update_date = data.Time.date()
80 
81 return data