Market Timing with Merton Rule for Earnings Yield

1from AlgorithmImports import *
2from typing import List, Dict
3import data_tools
4# endregion
5
6class MarketTimingWithMertonRuleForEarningsYield(QCAlgorithm):
7
8def Initialize(self) -> None:
9self.SetStartDate(2000, 1, 1) 
10self.SetCash(100000)
11
12self.risk_aversion_coefficient: int = 2
13
14self.long_period: int = 10 * 12 + 1      # 10 years of daily data
15self.short_period: int = 2 * 12 + 1      # 2 years of daily data
16self.SetWarmup(self.long_period, Resolution.Daily)
17
18self.long_term_weight: float = 0.75
19self.short_term_weight: float = 0.25
20
21self.min_market_allocation: int = 0
22self.max_market_allocation: int = 1
23
24self.leverage: int = 5
25
26self.market_prices: RollingWindow = RollingWindow[float](self.long_period)
27
28self.sp_earnings_yield_value: Union[None, float] = None
29self.real_yield_value: Union[None, float] = None
30
31# traded symbols
32security: Securities = self.AddEquity('SPY', Resolution.Daily)
33security.SetLeverage(self.leverage)
34self.market:Symbol = security.Symbol
35
36security: Securities = self.AddEquity('TIP', Resolution.Daily)
37security.SetLeverage(self.leverage)
38self.bonds:Symbol = security.Symbol
39
40self.sp_earnings_yield: Symbol = self.AddData(data_tools.MonthlyCustomData, 'SP500_EARNINGS_YIELD_MONTH', Resolution.Daily).Symbol
41
42us_treasury_long_term: bool = True
43if us_treasury_long_term:
44    self.real_yield: Symbol = self.AddData(data_tools.DailyCustomData, 'DLTIIT', Resolution.Daily).Symbol
45    self.real_yield_property = 'LT Real Average (>10Yrs)'
46else:
47    self.real_yield: Symbol = self.AddData(data_tools.DailyCustomData, 'DFII10', Resolution.Daily).Symbol
48    self.real_yield_property = '10 YR'
49
50self.custom_data: List[Symbol] = [self.sp_earnings_yield, self.real_yield]
51
52self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
53self.recent_month: int = -1
54
55def OnData(self, data: Slice) -> None:
56custom_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.LastDateHandler.get_last_update_date()
57
58# rebalance monthly
59if self.recent_month != self.Time.month:
60    self.recent_month = self.Time.month
61
62    if all(self.Securities[symbol].GetLastData() for symbol in self.custom_data) and any(self.Time.date() > custom_data_last_update_date[symbol] for symbol in self.custom_data):
63        self.Liquidate()
64        return
65
66    # update market prices
67    if self.market in data and data[self.market]:
68        price:float = data[self.market].Value
69        self.market_prices.Add(price)
70
71    if self.IsWarmingUp:
72        return
73
74    if self.sp_earnings_yield_value != None and self.real_yield_value != None and self.market_prices.IsReady \
75        and self.market in data and self.bonds in data and data[self.market] and data[self.bonds]:
76
77        excess_yield: float = self.sp_earnings_yield_value - self.real_yield_value
78        vollatility_forecast: float = self.VolatilityForecast()
79        
80        # Merton Rule calculation
81        market_allocation: float = excess_yield / (self.risk_aversion_coefficient * vollatility_forecast)
82        
83        # cap market allocation
84        market_allocation: float = min(max(market_allocation, self.min_market_allocation), self.max_market_allocation)
85
86        self.SetHoldings(self.market, market_allocation)
87        self.SetHoldings(self.bonds, 1 - market_allocation)
88    else:
89        self.Liquidate()
90
91    # reset data collected from previous month
92    self.sp_earnings_yield_value = None
93    self.real_yield_value = None
94
95if self.sp_earnings_yield in data and data[self.sp_earnings_yield] and data[self.sp_earnings_yield].Value != 0:
96    self.sp_earnings_yield_value = data[self.sp_earnings_yield].Value / 100
97
98if self.real_yield in data and data[self.real_yield] and data[self.real_yield].Price != 0:
99    self.real_yield_value = data[self.real_yield].Value / 100
100
101def VolatilityForecast(self) -> float:
102# volatility measures, both expressed as variances, and then take the square root of that weighted average to arrive at the spot estimate of equity volatility
103prices: List[float] = list(self.market_prices)
104
105long_term_prices: np.ndarray = np.array(prices[:self.long_period])
106short_term_prices: np.ndarray = np.array(prices[:self.short_period])
107
108long_term_variance: float = self.Variance(long_term_prices)
109short_term_variance: float = self.Variance(short_term_prices)
110
111weighted_average: float = (self.long_term_weight * long_term_variance + self.short_term_weight * short_term_variance) / (self.long_term_weight + self.short_term_weight)
112
113# take the square root of the weighted average
114return np.sqrt(weighted_average)
115
116def Variance(self, prices: np.ndarray) -> float:
117returns: np.ndarray = (prices[:-1] - prices[1:]) / prices[1:]
118ann_volatility: float = np.std(returns) * np.sqrt(12)
119variance: float = ann_volatility ** 2
120return variance