Avoid Equity Bear Markets with a Market Timing Strategy

1from AlgorithmImports import *
2from typing import List
3from pandas.core.frame import DataFrame
4import data_tools
5# endregion
6
7class AvoidEquityBearMarketswithaMarketTimingStrategy(QCAlgorithm):
8
9def Initialize(self) -> None:
10self.SetStartDate(2001, 1, 1)
11self.SetCash(100000)
12
13self.period:int = 365
14self.moving_average_period:int = 200
15self.leverage:int = 5
16
17# custom data subscription
18self.market_sec:Security = self.AddData(data_tools.MarketEQ, 'MKT', Resolution.Daily)
19self.t10y3m:Symbol = self.AddData(data_tools.TreasureBill, 'T10Y3M', Resolution.Daily).Symbol
20self.rrsfs:Symbol = self.AddData(data_tools.RRSFS, 'RRSFS_YOY', Resolution.Daily).Symbol
21self.indpro:Symbol = self.AddData(data_tools.IndustrialProduction, 'INDPRO_YOY', Resolution.Daily).Symbol
22self.house_started:Symbol = self.AddData(data_tools.HouseStarted, 'HOUST_YOY', Resolution.Daily).Symbol
23
24self.signal_market:Symbol = self.market_sec.Symbol
25self.traded_market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
26self.bill:Symbol = self.AddEquity("SHY", Resolution.Daily).Symbol
27
28for symbol in [self.traded_market, self.bill]:
29    self.Securities[symbol].SetLeverage(self.leverage)
30
31self.rebalance:bool = False
32
33self.Schedule.On(self.DateRules.MonthEnd(self.bill), self.TimeRules.BeforeMarketClose(self.bill), self.Selection)
34
35def OnData(self, data: Slice) -> None:
36if not self.rebalance:
37    return
38self.rebalance = False
39
40market_last_update_date:datetime.date = data_tools.MarketEQ._last_update_date
41t10y3m_last_update_date:datetime.date = data_tools.TreasureBill._last_update_date
42rrsfs_last_update_date:datetime.date = data_tools.RRSFS._last_update_date
43indpro_last_update_date:datetime.date = data_tools.IndustrialProduction._last_update_date
44house_started_last_update_date:datetime.date = data_tools.HouseStarted._last_update_date
45
46# call history on assets
47symbols:List[Symbol] = [self.t10y3m, self.rrsfs, self.indpro, self.house_started]
48history:DataFrame = self.History(symbols, self.period, Resolution.Daily)['value'].unstack(level=0)
49
50history_market:DataFrame = self.History([self.signal_market], self.period, Resolution.Daily)['value']
51
52if len(history.dropna().iloc[-3:]) != 3 or history.dropna().iloc[-3:].isnull().values.any() or not all(x in list(history.columns) for x in symbols): 
53    return
54
55if len(history) >= self.moving_average_period:
56    history.reset_index(inplace=True)
57    history.set_index('time', inplace=True)
58    market_returns:DataFrame = history_market.iloc[-self.moving_average_period:].pct_change()
59    
60    rr:float = self.rachev_ratio(market_returns.iloc[1:], 0.5)
61
62    trend_signal:bool = False
63    macro_signal:bool = False
64
65    # trend signal evaluation
66    if history_market.iloc[-1] > history_market.iloc[-self.moving_average_period:].mean() and rr >= 1. \
67        and history[self.t10y3m].iloc[-1] > 0:
68        trend_signal = True
69
70    if (self.Securities[self.signal_market].GetLastData() and self.Time.date() >= market_last_update_date) or \
71        (self.Securities[self.t10y3m].GetLastData() and self.Time.date() >= t10y3m_last_update_date) or \
72        (self.Securities[self.rrsfs].GetLastData() and self.Time.date() >= rrsfs_last_update_date) or \
73        (self.Securities[self.indpro].GetLastData() and self.Time.date() >= indpro_last_update_date) or \
74        (self.Securities[self.house_started].GetLastData() and self.Time.date() >= house_started_last_update_date):
75        
76        self.Liquidate()
77        return
78
79    # macro signal evaluation
80    if history[self.rrsfs].dropna().iloc[-2] > 0 and history[self.indpro].dropna().iloc[-2] > 0 and history[self.house_started].dropna().iloc[-2] > history[self.house_started].dropna().iloc[-3]:
81        macro_signal = True
82
83    if self.traded_market in data and data[self.traded_market] and self.bill in data and data[self.bill]:
84        traded_asset:Symbol = self.traded_market
85
86        if not self.Portfolio.Invested:
87            if trend_signal:
88                traded_asset:Symbol = self.traded_market
89            else:
90                traded_asset = self.bill
91
92        elif self.Portfolio[self.bill].Invested and trend_signal:
93            traded_asset = self.traded_market
94        elif self.Portfolio[self.traded_market].Invested and not trend_signal and not macro_signal:
95            traded_asset = self.bill
96        
97        if not self.Portfolio[traded_asset].Invested:
98            self.Liquidate()
99            self.SetHoldings(traded_asset, 1)
100
101def Selection(self) -> None:
102self.rebalance = True
103
104def rachev_ratio(self, df:DataFrame, alpha=0.5) -> float:
105# calculate VaR for left and right tails
106left_var:float = df.quantile(alpha)
107right_var:float = df.quantile(1 - alpha)
108
109# calculate Expected Shortfall for left and right tails
110left_es:float = df[df <= left_var].mean()
111right_es:float = df[df > right_var].mean()
112
113# calculate the Rachev Ratio
114rr:float = right_es / -left_es
115
116return rr