ESG Factor Long-Short North America Strategy

1from AlgorithmImports import *
2from numpy import floor
3from typing import List, Dict
4from dataclasses import dataclass
5from decimal import *
6#endregion
7class ESGFactorInvestingStrategy(QCAlgorithm):
8def Initialize(self) -> None:
9 self.SetStartDate(2009, 6, 1)
10 self.SetCash(100_000)
11 # Decile weighting.
12 # True - Value weighted
13 # False - Equally weighted
14 self.value_weighting: bool = True
15 
16 # self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
17 self.esg_data: Data = self.AddData(ESGData, 'ESG', Resolution.Daily)
18 
19 # All tickers from ESG database.
20 self.tickers: List[str] = []
21 
22 self.ticker_deciles: Dict[str, float] = {}
23 
24 self.holding_period: float = 12
25 self.leverage: int = 10
26 self.threshold: List[int] = [0.2, 0.8]
27 self.managed_queue: List[RebalanceQueueItem] = []
28 
29 self.latest_price: Dict[Symbol, float] = {}
30 
31 self.selection_flag: bool = False
32 self.UniverseSettings.Leverage = self.leverage
33 self.UniverseSettings.Resolution = Resolution.Daily
34 self.AddUniverse(self.FundamentalSelectionFunction)
35 self.settings.daily_precise_end_time = False
36 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
37
38def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
39 for security in changes.AddedSecurities:
40     security.SetFeeModel(CustomFeeModel())
41
42def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
43 if not self.selection_flag:
44     return Universe.Unchanged
45 
46 self.latest_price.clear()
47 
48 selected: List[Fundamental] = [
49     x for x in fundamental 
50     if x.MarketCap != 0
51     and (x.Symbol.Value).lower() in self.tickers
52 ]
53 
54 for stock in selected:
55     symbol: Symbol = stock.Symbol
56     self.latest_price[symbol] = stock.AdjustedPrice
57 # Store symbol/market cap pair.
58 long: List[Fundamental] = [
59     x for x in selected if (x.Symbol.Value in self.ticker_deciles) and \
60     (self.ticker_deciles[x.Symbol.Value] is not None) and \
61     (self.ticker_deciles[x.Symbol.Value] >= self.threshold[1])
62 ]
63 
64 short: List[Fundamental] = [
65     x for x in selected if (x.Symbol.Value in self.ticker_deciles) and \
66     (self.ticker_deciles[x.Symbol.Value] is not None) and \
67     (self.ticker_deciles[x.Symbol.Value] <= self.threshold[0])
68 ]
69 
70 weights: List[Tuple[Symbol, float]] = []
71 # ew
72 if not self.value_weighting:
73     for i, portfolio in enumerate([long, short]):
74         for stock in portfolio:
75             w: float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(portfolio)
76             weights.append((stock.Symbol, ((-1) ** i) * floor(w / self.latest_price[stock.Symbol])))
77 # vw
78 else:
79     for i, portfolio in enumerate([long, short]):
80         mc_sum: float = sum(list(map(lambda x: x.MarketCap, portfolio)))
81         for stock in portfolio:
82             w: float = self.Portfolio.TotalPortfolioValue / self.holding_period
83             weights.append((stock.Symbol, ((-1) ** i) * floor((w * (stock.MarketCap / mc_sum))) / self.latest_price[stock.Symbol]))
84 self.managed_queue.append(RebalanceQueueItem(weights))
85 
86 self.ticker_deciles.clear()
87 
88 return [x.Symbol for x in long + short]
89def OnData(self, slice: Slice) -> None:
90 new_data_arrived: bool = False
91 custom_data_last_update_date: datetime.date = ESGData.get_last_update_date()
92 if self.esg_data.get_last_data() and self.time.date() > custom_data_last_update_date:
93     self.liquidate()
94     return
95 
96 if slice.contains_key('ESG') and slice['ESG']:
97     # Store universe tickers.
98     if len(self.tickers) == 0:
99         # TODO '_typename' in storage dictionary?
100         self.tickers = [x.Key for x in self.esg_data.GetLastData().GetStorageDictionary()][1:-1]
101 
102     # Store history for every ticker.
103     for ticker in self.tickers:
104         ticker_u: str = ticker.upper()
105         if ticker_u not in self.ticker_deciles:
106             self.ticker_deciles[ticker_u] = None
107         
108         decile: float = self.esg_data.GetLastData()[ticker]
109         self.ticker_deciles[ticker_u] = decile
110         
111         # trigger selection after new esg data arrived.
112         if not self.selection_flag:
113             new_data_arrived = True
114 
115 if new_data_arrived:
116     self.selection_flag = True
117     return
118 
119 if not self.selection_flag:
120     return
121 self.selection_flag = False
122 # Trade execution
123 remove_item: Union[None, RebalanceQueueItem] = None
124 
125 # Rebalance portfolio
126 for item in self.managed_queue:
127     if item.holding_period == self.holding_period:
128         for symbol, quantity in item.symbol_q:
129             self.MarketOrder(symbol, -quantity)
130                     
131         remove_item = item
132         
133     elif item.holding_period == 0:
134         open_symbol_q: List[RebalanceQueueItem] = []
135         
136         for symbol, quantity in item.symbol_q:
137             if abs(quantity) >= 1:
138                 if slice.contains_key(symbol) and slice[symbol]:
139                     self.MarketOrder(symbol, quantity)
140                     open_symbol_q.append((symbol, quantity))
141                     
142         # Only opened orders will be closed        
143         item.symbol_q = open_symbol_q
144         
145     item.holding_period += 1
146     
147 if remove_item:
148     self.managed_queue.remove(remove_item)
149@dataclass
150class RebalanceQueueItem():
151# symbol/quantity collections
152symbol_q: List[Tuple[Symbol, float]] 
153holding_period: int = 0
154 
155# ESG data.
156class ESGData(PythonData):
157_last_update_date:datetime.date = datetime(1,1,1).date()
158@staticmethod
159def get_last_update_date() -> datetime.date:
160return ESGData._last_update_date
161def __init__(self):
162 self.tickers = []
163
164def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
165 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/economic/esg_deciles_data.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
166
167def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
168 data = ESGData()
169 data.Symbol = config.Symbol
170 
171 if not line[0].isdigit():
172     self.tickers = [x for x in line.split(';')][1:]
173     return None
174     
175 split = line.split(';')
176 
177 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
178 index = 1
179 for ticker in self.tickers:
180     data[ticker] = float(split[index])
181     index += 1
182     
183 data.Value = float(split[1])
184 if data.Time.date() > ESGData._last_update_date:
185     ESGData._last_update_date = data.Time.date()
186 return data
187 
188# Custom fee model.
189class CustomFeeModel(FeeModel):
190def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
191 fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
192 return OrderFee(CashAmount(fee, "USD"))