Benchmarks Portfolios with Decreasing Carbon Footprints

1from AlgorithmImports import *
2from typing import List, Dict
3# endregion
4
5class BenchmarksPortfolioswithDecreasingCarbonFootprints(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2015, 1, 1)
9self.SetCash(100_000)
10
11self.leverage: int = 3
12self.quantile: int = 3
13self.portfolio_threshold: float = 0.75
14
15self.data: Dict[int, Dict[str, float]] = {}
16
17# DJI stocks
18self.tickers: List[str] = [
19    'AXP', 'AMGN', 'AAPL', 'BA', 'CAT', 'CSCO', 'CVX', 'GS', 'HD', 'HON', 
20    'IBM', 'INTC', 'JNJ', 'KO', 'JPM', 'MCD', 'MMM', 'MRK', 'MSFT', 'NKE', 
21    'PG', 'TRV', 'UNH', 'CRM', 'VZ', 'V', 'WBA', 'WMT', 'DIS', 'DOW'
22]
23
24for ticker in self.tickers:
25    data: Equity = self.AddEquity(ticker, Resolution.Daily)
26    data.SetLeverage(self.leverage)
27
28# Source: https://esg.exerica.com/Company?Name=Apple
29ghg_emissions: str = self.Download('data.quantpedia.com/backtesting_data/economic/GHG_emissions.csv')
30lines: List[str] = ghg_emissions.split('\r\n')
31
32for line in lines[1:]:  # skip first comment lines
33    if line == '':
34        continue
35    
36    line_split: List[str] = line.split(';')
37    year: str = line_split[0]
38    if year not in self.data:
39        self.data[year] = {}
40 
41    # N stocks -> n*4 properties
42    for i in range(1, len(line_split), 4):
43        # parse GHG emissions info
44        if line_split[i] not in self.data[year]:
45            self.data[year][line_split[i]] = sum( float(line_split[x]) for x in range(i+1, i+4) if line_split[x] != '' )
46
47self.rebalance_month: int = 7
48self.current_month: int = -1
49
50def OnData(self, slice: Slice) -> None:
51if self.Time.month == self.current_month:
52    return
53self.current_month = self.Time.month
54
55if self.Time.month != self.rebalance_month:
56    return
57
58if str(self.Time.year - 1) not in list(self.data.keys()):
59    return
60
61# calculate emission intensity
62emission_intensity: Dict[Symbol, float] = { 
63    self.Symbol(ticker) : self.data[str(self.Time.year - 1)][ticker] / self.Securities[ticker].Fundamentals.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths \
64    for ticker in self.tickers if self.data[str(self.Time.year - 1)][ticker] != 0 and self.Securities[ticker].Fundamentals.HasFundamentalData\
65    and not np.isnan(self.Securities[ticker].Fundamentals.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths) \
66    and self.Securities[ticker].Fundamentals.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths != 0 
67}
68
69weight: Dict[Symbol, float] = {}
70final_portfolio: List[Symbol] = []
71portfolio_percentage: float = 0.
72
73# sort and divide portfolio
74if len(emission_intensity) != 0:
75    sorted_emissions:List[Symbol] = sorted(emission_intensity, key=emission_intensity.get)
76
77    # calculate weights based on marketcap
78    mc_sum:float = sum(list(map(lambda symbol: self.Securities[symbol].Fundamentals.MarketCap, sorted_emissions)))
79    for symbol in sorted_emissions:
80        w: float = self.Securities[symbol].Fundamentals.MarketCap / mc_sum
81
82        portfolio_percentage += w
83        if portfolio_percentage <= self.portfolio_threshold:
84            final_portfolio.append(symbol)
85
86if len(final_portfolio) != 0:
87    mc_sum: float = sum(list(map(lambda symbol: self.Securities[symbol].Fundamentals.MarketCap, final_portfolio)))
88    for symbol in final_portfolio:
89        weight[symbol] = self.Securities[symbol].Fundamentals.MarketCap / mc_sum
90
91# trade execution
92portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in weight.items() if symbol in slice and slice[symbol]]
93self.SetHoldings(portfolio, True)