“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.”

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]

<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 Return3.25%
VolatilityN/A
Beta-0.051
Sharpe RatioN/A
Sortino Ratio-0.467
Maximum DrawdownN/A
Win Rate44%

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"))

Leave a Reply

Discover more from Quant Buffet

Subscribe now to keep reading and get access to the full archive.

Continue reading