Climate Policy Uncertainty and the Cross-Section of Stock Returns

1from AlgorithmImports import *
2import statsmodels.api as sm
3import data_tools
4# endregion
5
6class ClimatePolicyUncertaintyAndTheCrossSectionOfStockReturns(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11
12self.leverage:int = 5
13
14self.period:int = 21
15self.regression_period:int = 60
16self.max_missing_months:int = 2
17self.quantile:int = 10  
18
19self.weights:dict[Symbol, float] = {}    
20self.prices:dict[Symbol, RollingWindow] = {}
21
22self.exchanges:list[str] = ['NYS', 'NAS', 'ASE']
23
24self.cpu_index:Symbol = self.AddData(data_tools.CPUIndex, 'CPU_INDEX', Resolution.Daily).Symbol        
25self.fama_french:Symbol = self.AddData(data_tools.QuantpediaFamaFrench, 'fama_french_3_factor_monthly', Resolution.Daily).Symbol
26
27self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
28self.prev_market_price:float = None
29
30self.regression_data:RegressionData = data_tools.RegressionData(self.regression_period)        
31
32self.coarse_count:int = 1000
33self.recent_month:int = -1
34self.selection_flag:bool = False
35self.UniverseSettings.Resolution = Resolution.Daily
36self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
37
38def OnSecuritiesChanged(self, changes):
39for security in changes.AddedSecurities:
40    security.SetFeeModel(data_tools.CustomFeeModel())
41    security.SetLeverage(self.leverage)
42
43def CoarseSelectionFunction(self, coarse):
44if not self.selection_flag:
45    return Universe.Unchanged
46
47if self.coarse_count < 3000:
48    selected:list = sorted([x for x in coarse if x.HasFundamentalData and x.Price >= 2 and x.Price <= 1000],
49            key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
50else:
51    selected:list = [x for x in coarse if x.HasFundamentalData and x.Price >= 2 and x.Price <= 1000]
52
53selected_symbols:list[Symbol] = []
54
55for stock in selected:
56    symbol:Symbol = stock.Symbol
57
58    if symbol not in self.prices:
59        # init rolling window for newly selected stock
60        self.prices[symbol] = RollingWindow[float](self.regression_period + 1)
61
62    selected_symbols.append(symbol)
63
64for equity in coarse:
65    symbol:Symbol = equity.Symbol
66
67    if symbol in self.prices:
68        # update monthly prices
69        self.prices[symbol].Add(equity.AdjustedPrice)
70
71return [x for x in selected_symbols if self.prices[x].IsReady]
72
73def FineSelectionFunction(self, fine):
74# make sure regression data are ready
75if not self.regression_data.is_ready():
76    return Universe.Unchanged
77
78# make sure regression data are up to date, otherwise reset them
79elif not self.regression_data.data_still_coming(self.max_missing_months, self.Time.date()):
80    self.regression_data.reset()
81    return Universe.Unchanged
82
83# filter fine
84fine:list = [x for x in fine if x.MarketCap != 0 and x.MarketCap >= 5000000 and x.SecurityReference.ExchangeId in self.exchanges]
85
86if len(fine) > self.coarse_count:
87    sorted_by_market_cap:list = sorted(fine, key = lambda x: x.MarketCap, reverse=True)
88    fine:list = sorted_by_market_cap[:self.coarse_count]
89
90cpu_beta:dict = {}
91regression_x:list = self.regression_data.regression_x()
92
93for stock in fine:
94    symbol:Symbol = stock.Symbol
95
96    prices:np.array = np.array([x for x in self.prices[symbol]])
97    regression_y:np.array = (prices[:-1] / prices[1:]) - 1
98
99    regression_model = self.MultipleLinearRegression(regression_x, regression_y)
100    cpu_beta_value:float = regression_model.params[1]
101    cpu_beta[stock] = cpu_beta_value
102
103# make sure there are enough stocks for selection
104if len(cpu_beta) < self.quantile:
105    return Universe.Unchanged
106
107quantile:int = int(len(cpu_beta) / self.quantile)
108# low CPU beta stocks are in the top decile whereas high CPU beta stocks are in the bottom decile
109sorted_by_beta:list = [x[0] for x in sorted(cpu_beta.items(), key=lambda item: item[1], reverse=True)]
110
111# the zero-investment portfolio is constructed by long the top decile and short the bottom decile
112long_part:list = sorted_by_beta[-quantile:]
113short_part:list = sorted_by_beta[:quantile]
114
115total_long_cap:float = sum([stock.MarketCap for stock in long_part])
116for stock in long_part:
117    self.weights[stock.Symbol] = stock.MarketCap / total_long_cap
118
119total_short_cap:float = sum([stock.MarketCap for stock in short_part])
120for stock in short_part:
121    self.weights[stock.Symbol] = -stock.MarketCap / total_short_cap
122
123return list(self.weights.keys())
124
125def OnData(self, data):
126rebalance_flag = False
127
128# update regression data when fama french and cpu data come and market prices are ready
129if self.fama_french in data and data[self.fama_french] and self.cpu_index in data and data[self.cpu_index]:
130    if self.recent_month != self.Time.month:
131        rebalance_flag = True
132        self.recent_month = self.Time.month
133
134    if self.Securities.ContainsKey(self.market_symbol) and self.Securities[self.market_symbol].Price != 0:
135        # get latest market price - market may not be opened at the CPU and FF arrival date
136        # market_price:float = data[self.market_symbol].Value
137        market_price:float = self.Securities[self.market_symbol].Price
138
139        if self.prev_market_price != None:
140            market_return:float = (market_price - self.prev_market_price) / self.prev_market_price
141
142            self.regression_data.update(
143                data[self.cpu_index].Value,
144                market_return,
145                data[self.fama_french].Market,
146                data[self.fama_french].Size,
147                data[self.fama_french].Value,
148                self.Time.date()
149            )
150        
151        self.prev_market_price = market_price            
152
153# rebalance monthly
154if rebalance_flag:
155    self.selection_flag = True
156    return
157
158if not self.selection_flag:
159    return
160self.selection_flag = False
161
162# trade execution
163invested = [x.Key for x in self.Portfolio if x.Value.Invested]
164for symbol in invested:
165    if symbol not in self.weights:
166        self.Liquidate(symbol)
167
168for symbol, w in self.weights.items():
169    if symbol in data and data[symbol]:
170        self.SetHoldings(symbol, w)
171        
172self.weights.clear()
173
174def MultipleLinearRegression(self, x:list, y:list):
175x:np.array = np.array(x).T
176x = sm.add_constant(x)
177result = sm.OLS(endog=y, exog=x).fit()
178return result