
“The strategy ranks North American stocks by ESG scores, going long top 20% and short bottom 20%, combining equally-weighted E, S, and G strategies, rebalanced annually.”
ASSET CLASS: stocks | REGION: United States | FREQUENCY:
Yearly | MARKET: equities | KEYWORD: ESG, Level, Factor, Investing, Strategy
I. STRATEGY IN A NUTSHELL
The study uses Asset4 ESG scores, updated annually, to assess environmental, social, and governance performance of North American stocks (Canada and the US). Stocks priced below $1 are excluded. ESG scores are held constant until the next assessment. Returns are evaluated as abnormal returns using the Daniel et al. (1997) methodology, which accounts for risk factors like size, book-to-market ratio, and momentum by matching each stock to a 4×4 benchmark portfolio with similar characteristics.
Stocks are ranked monthly by their E, S, and G scores. The strategy involves going long on the top 20% and short on the bottom 20% of each score, creating three individual strategies. These are combined into a single, equally-weighted strategy, rebalanced annually. This approach evaluates the impact of ESG factors on returns while controlling for key risk characteristics.
II. ECONOMIC RATIONALE
Socially responsible investing (SRI) is gaining popularity, with increasing global investments driven by profit and non-profit motives. High ESG scores, reflecting sustainability and long-term viability, are linked to positive or zero abnormal returns in the short term for Europe and North America, and significant abnormal returns in the long run across all ESG categories—Environment, Social, and Governance. Firms with high ESG scores benefit from reduced regulatory fines, lower risk exposure, better management, and enhanced brand reputation. Additionally, customers may pay a premium for products from environmentally responsible firms. In the long term, strong corporate social performance translates into cost savings and unexpected high cash flows, making ESG-driven strategies financially advantageous.
III. SOURCE PAPER
Where and When Does It Pay to Be Good? A Global Long-Term Analysis of ESG Investing [Click to Open PDF]
- Dorfleitner, Gregor and Utz, Sebastian and Wimmer, Maximilian, Department of Finance, University of Regensburg, 93040 Regensburg, Germany, Area Finance, University of Mannheim, 68131 Mannheim, Germany
<Abstract>
This paper explores the long-term performance of stocks with high corporate social performance (CSP), measured by so-called ESG scores depicting the environmental (E), social (S), and governance (G) dimension. We investigate the buy-and-hold abnormal returns of a long/short investment strategy including the top and low 20% stocks with respect to each of the ESG dimensions. The results of the bootstrap tests in a world-wide perspective indicate that financial markets are not capable to price different levels of CSP in the short run and in particular in the long run properly. The zero investment strategy produces significantly positive abnormal returns up to 20% in North America and Europe in a five year period. We also identify regional differences, for instance, a high social score does not pay in Japan and strong corporate governance yields significantly negative abnormal returns in Asia Pacific.

IV. BACKTEST PERFORMANCE
| Annualised Return | 3.25% |
| Volatility | N/A |
| Beta | -0.051 |
| Sharpe Ratio | N/A |
| Sortino Ratio | -0.467 |
| Maximum Drawdown | N/A |
| Win Rate | 44% |
V. FULL PYTHON CODE
from AlgorithmImports import *
from numpy import floor
from typing import List, Dict
from dataclasses import dataclass
from decimal import *
#endregion
class ESGFactorInvestingStrategy(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2009, 6, 1)
self.SetCash(100_000)
# Decile weighting.
# True - Value weighted
# False - Equally weighted
self.value_weighting: bool = True
# self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.esg_data: Data = self.AddData(ESGData, 'ESG', Resolution.Daily)
# All tickers from ESG database.
self.tickers: List[str] = []
self.ticker_deciles: Dict[str, float] = {}
self.holding_period: float = 12
self.leverage: int = 10
self.threshold: List[int] = [0.2, 0.8]
self.managed_queue: List[RebalanceQueueItem] = []
self.latest_price: Dict[Symbol, float] = {}
self.selection_flag: bool = False
self.UniverseSettings.Leverage = self.leverage
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
if not self.selection_flag:
return Universe.Unchanged
self.latest_price.clear()
selected: List[Fundamental] = [
x for x in fundamental
if x.MarketCap != 0
and (x.Symbol.Value).lower() in self.tickers
]
for stock in selected:
symbol: Symbol = stock.Symbol
self.latest_price[symbol] = stock.AdjustedPrice
# Store symbol/market cap pair.
long: List[Fundamental] = [
x for x in selected if (x.Symbol.Value in self.ticker_deciles) and \
(self.ticker_deciles[x.Symbol.Value] is not None) and \
(self.ticker_deciles[x.Symbol.Value] >= self.threshold[1])
]
short: List[Fundamental] = [
x for x in selected if (x.Symbol.Value in self.ticker_deciles) and \
(self.ticker_deciles[x.Symbol.Value] is not None) and \
(self.ticker_deciles[x.Symbol.Value] <= self.threshold[0])
]
weights: List[Tuple[Symbol, float]] = []
# ew
if not self.value_weighting:
for i, portfolio in enumerate([long, short]):
for stock in portfolio:
w: float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(portfolio)
weights.append((stock.Symbol, ((-1) ** i) * floor(w / self.latest_price[stock.Symbol])))
# vw
else:
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda x: x.MarketCap, portfolio)))
for stock in portfolio:
w: float = self.Portfolio.TotalPortfolioValue / self.holding_period
weights.append((stock.Symbol, ((-1) ** i) * floor((w * (stock.MarketCap / mc_sum))) / self.latest_price[stock.Symbol]))
self.managed_queue.append(RebalanceQueueItem(weights))
self.ticker_deciles.clear()
return [x.Symbol for x in long + short]
def OnData(self, slice: Slice) -> None:
new_data_arrived: bool = False
custom_data_last_update_date: datetime.date = ESGData.get_last_update_date()
if self.esg_data.get_last_data() and self.time.date() > custom_data_last_update_date:
self.liquidate()
return
if slice.contains_key('ESG') and slice['ESG']:
# Store universe tickers.
if len(self.tickers) == 0:
# TODO '_typename' in storage dictionary?
self.tickers = [x.Key for x in self.esg_data.GetLastData().GetStorageDictionary()][1:-1]
# Store history for every ticker.
for ticker in self.tickers:
ticker_u: str = ticker.upper()
if ticker_u not in self.ticker_deciles:
self.ticker_deciles[ticker_u] = None
decile: float = self.esg_data.GetLastData()[ticker]
self.ticker_deciles[ticker_u] = decile
# trigger selection after new esg data arrived.
if not self.selection_flag:
new_data_arrived = True
if new_data_arrived:
self.selection_flag = True
return
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
remove_item: Union[None, RebalanceQueueItem] = None
# Rebalance portfolio
for item in self.managed_queue:
if item.holding_period == self.holding_period:
for symbol, quantity in item.symbol_q:
self.MarketOrder(symbol, -quantity)
remove_item = item
elif item.holding_period == 0:
open_symbol_q: List[RebalanceQueueItem] = []
for symbol, quantity in item.symbol_q:
if abs(quantity) >= 1:
if slice.contains_key(symbol) and slice[symbol]:
self.MarketOrder(symbol, quantity)
open_symbol_q.append((symbol, quantity))
# Only opened orders will be closed
item.symbol_q = open_symbol_q
item.holding_period += 1
if remove_item:
self.managed_queue.remove(remove_item)
@dataclass
class RebalanceQueueItem():
# symbol/quantity collections
symbol_q: List[Tuple[Symbol, float]]
holding_period: int = 0
# ESG data.
class ESGData(PythonData):
_last_update_date:datetime.date = datetime(1,1,1).date()
@staticmethod
def get_last_update_date() -> datetime.date:
return ESGData._last_update_date
def __init__(self):
self.tickers = []
def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/economic/esg_deciles_data.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
data = ESGData()
data.Symbol = config.Symbol
if not line[0].isdigit():
self.tickers = [x for x in line.split(';')][1:]
return None
split = line.split(';')
data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
index = 1
for ticker in self.tickers:
data[ticker] = float(split[index])
index += 1
data.Value = float(split[1])
if data.Time.date() > ESGData._last_update_date:
ESGData._last_update_date = data.Time.date()
return data
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))