ESG Exclusion Premium Factor

1from AlgorithmImports import *
2from datetime import datetime
3from data_tools import CustomFeeModel, QuantpediaESGExclusion, SymbolData
4from scipy import stats
5# endregion
6
7class ESGExclusionPremiumFactor(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2005, 1, 1) # ESG exclusion starts in 2005
11self.SetCash(100000)
12
13self.leverage:int = 5
14
15self.period:int = 12 * 21
16
17self.max_missing_days:int = 15
18
19self.data:dict[Symbol, SymbolData] = {}
20self.weights:dict[Symbol, float] = {}
21self.excluded_stocks:list[str] = []
22
23self.esg_exclusion_symbol:Symbol = self.AddData(QuantpediaESGExclusion, 'esg_exclusion', Resolution.Daily).Symbol
24
25security = self.AddEquity('ACWI', Resolution.Daily)
26security.SetFeeModel(CustomFeeModel())
27security.SetLeverage(self.leverage)
28self.market:Symbol = security.Symbol
29
30self.data[self.market] = SymbolData(self.period)
31history = self.History(self.market, self.period, Resolution.Daily)
32if not history.empty:
33    closes = history.loc[symbol].close
34
35    for time, close in closes.iteritems():
36        self.data[symbol].update(time.date(), close)
37
38self.rebalance_flag:bool = False
39self.selection_flag:bool = False
40self.UniverseSettings.Resolution = Resolution.Daily
41self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
42
43def OnSecuritiesChanged(self, changes):
44for security in changes.AddedSecurities:
45    security.SetFeeModel(CustomFeeModel())
46    security.SetLeverage(self.leverage)
47
48def CoarseSelectionFunction(self, coarse):
49curr_date:datetime.date = self.Time.date()
50
51for equity in coarse:
52    symbol:Symbol = equity.Symbol
53
54    if symbol in self.data:
55        self.data[symbol].update(curr_date, equity.AdjustedPrice)
56
57if not self.selection_flag or not self.data[self.market].is_ready():
58    return Universe.Unchanged
59
60if not self.data[self.market].data_still_coming(curr_date, self.max_missing_days):
61    self.data[self.market].reset_data()
62self.selection_flag = False
63
64warmed_up_symbols:list[Symbol] = []
65for equity in coarse:
66    if not equity.Symbol.Value in self.excluded_stocks:
67        continue
68
69    symbol:Symbol = equity.Symbol
70
71    if symbol not in self.data:                
72        self.data[symbol] = SymbolData(self.period)
73        history = self.History(symbol, self.period, Resolution.Daily)
74        if history.empty:
75            continue
76
77        closes = history.loc[symbol].close
78
79        for time, close in closes.iteritems():
80            self.data[symbol].update(time.date(), close)
81
82    if self.data[symbol].is_ready():
83        warmed_up_symbols.append(symbol)
84
85self.rebalance_flag = True
86
87return warmed_up_symbols
88
89def FineSelectionFunction(self, fine):
90fine:list[FineFundamental] = list(filter(lambda stock: stock.MarketCap != 0, fine))
91
92total_cap:float = sum(list(map(lambda stock: stock.MarketCap, fine)))
93
94market_daily_returns:list[float] = self.data[self.market].get_daily_returns()
95beta_values:list[float] = []
96
97for stock in fine:
98    symbol:Symbol = stock.Symbol
99
100    stock_daily_returns:list[float] = self.data[stock.Symbol].get_daily_returns()
101
102    slope, intercept, r_value, p_value, std_err = stats.linregress(market_daily_returns, stock_daily_returns)
103
104    beta_values.append(slope)
105
106    self.weights[symbol] = stock.MarketCap / total_cap
107
108self.weights[self.market] = -np.average(beta_values)
109
110return list(self.weights.keys())
111
112def OnData(self, data: Slice):
113if self.esg_exclusion_symbol in data and data[self.esg_exclusion_symbol]:
114    self.selection_flag = True
115    excluded_tickers:list[str] = [x for x in data[self.esg_exclusion_symbol].GetProperty('tickers')]
116    self.excluded_stocks += excluded_tickers
117
118if not self.rebalance_flag:
119    return
120self.rebalance_flag = False
121
122invested:list[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
123for symbol in invested:
124    if symbol not in self.weights:
125        self.Liquidate(symbol)
126        
127for symbol, w in self.weights.items():
128    if symbol in data and data[symbol]:
129        self.SetHoldings(symbol, w)
130
131self.weights.clear()