气候情绪、碳价与“排放减去清洁”投资组合

“通过气候情绪和碳价格交易美国股票，每月根据情绪-价格动态调整价值加权的EMC投资组合，通过买入或卖空，或投资于无风险资产。”

资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 气候、碳价格

I. 策略概要

投资范围包括来自汤森路透Datastream的美国股票（纽约证券交易所和纳斯达克）。该策略结合了投资者气候情绪（使用StockTwits数据和sentimentr R包衡量）和碳价格（使用ICE-ECX碳排放配额（EUA）期货结算价格）。股票根据IPCC定义分为高排放或清洁股票。EMC投资组合做多高排放股票，做空清洁股票，两者均按价值加权。每月，如果气候情绪上升且碳价格下跌，则买入EMC投资组合；如果情绪下跌且碳价格上涨，则做空EMC投资组合。如果两者同向变动，则投资于无风险资产。投资组合按价值加权，每月重新平衡。

II. 策略合理性

先前的研究分析了与气候相关的Twitter帖子，但面临噪音和相关性问题。StockTwits，仅供投资者使用，提供了一个更清晰、更有针对性的数据集。该策略的成功取决于投资者情绪和时机。气候情绪受全球变暖和灾害等话题影响，进而影响股价：在高情绪时期，高排放股票被低估，清洁股票被高估，从而在EMC投资组合中产生错误定价。这种错误定价产生了正回报。股票被分为高排放和清洁两类，尽管“清洁”股票并非零排放，而是来自非碳密集型行业。该策略主要涉及根据气候情绪对高排放股票进行时机选择。

III. 来源论文

Investors’ Climate Sentiment and Financial Markets [点击查看论文]

• Santi, Caterina。列日大学 – HEC 列日管理学院

<摘要>

我们建议通过对StockTwits上关于气候变化和全球变暖的帖子进行情感分析来衡量投资者气候情感。在金融市场中，当投资者气候情感更积极时，排放（碳密集型）公司的股票表现不及清洁（低排放）股票。我们记录了投资者对气候变化风险的过度反应以及长期反转。显著但非信息性的气候变化事件，例如气候变化报告的发布和异常天气事件，促进了投资者学习过程和错误定价的纠正。

IV. 回测表现

年化回报	9.36%
波动率	12.71%
β值	0.056
夏普比率	0.74
索提诺比率	-0.139
最大回撤	N/A
胜率	43%

V. 完整的 Python 代码

from AlgorithmImports import *
from data_tools import QuantpediaClimateChange, QuantpediaFutures, ClimateChangeData, CarbonFutureData, CustomFeeModel, LastDateHandler
from typing import Dict, List
# endregion
class ClimateSentimentCarbonPricesAndEmissionMinusCleanPortfolio(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2007, 1, 1)   # climate change data starts in 2004 and carbon future data starts in 2007
        self.SetCash(100_000)
        
        self.leverage: int = 5
        self.carbon_max_missing_days:int = 5
        self.climate_max_missing_days:int = 40
        self.min_prices: int = 15
        
        self.weights: Dict[Symbol, float] = {}
        self.high_emission_industries: List[MorningstarIndustryGroupCode] = [
            MorningstarIndustryGroupCode.Agriculture,
            MorningstarIndustryGroupCode.BuildingMaterials,
            MorningstarIndustryGroupCode.Chemicals,
            MorningstarIndustryGroupCode.Construction,
            MorningstarIndustryGroupCode.FarmAndHeavyConstructionMachinery,
            MorningstarIndustryGroupCode.ForestProducts,
            MorningstarIndustryGroupCode.HomebuildingAndConstruction,
            MorningstarIndustryGroupCode.MetalsAndMining,
            MorningstarIndustryGroupCode.OilAndGas,
            MorningstarIndustryGroupCode.Steel,
            MorningstarIndustryGroupCode.Transportation,
        ]
        self.market_symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        self.climate_change: Symbol = self.AddData(QuantpediaClimateChange, 'CLIMATE_CHANGE', Resolution.Daily).Symbol
        self.climate_change_data: ClimateChangeData = ClimateChangeData()
        self.carbon_future: Symbol = self.AddData(QuantpediaFutures, 'ICE_EUA1', Resolution.Daily).Symbol
        self.carbon_future_data: CarbonFutureData = CarbonFutureData()
        self.fundamental_count: int = 500
        self.fundamental_sorting_key = lambda x: x.DollarVolume
        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.
        self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), self.Selection)
    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
        curr_date: datetime.date = self.Time.date()
        custom_data_last_update_date: Dict[Symbol, datetime.date] = LastDateHandler.get_last_update_date()
        if any((self.securities[symbol].get_last_data() and self.time.date() > custom_data_last_update_date[symbol]) for symbol in [self.climate_change, self.carbon_future]):
            self.Liquidate()
            return Universe.UNCHANGED
        # if not self.climate_change_data.data_still_coming(curr_date, self.climate_max_missing_days):
        #     self.climate_change_data.reset()
        # if not self.carbon_future_data.data_still_coming(curr_date, self.carbon_max_missing_days):
        #     self.carbon_future_data.reset()
        if not self.climate_change_data.is_ready() or not self.carbon_future_data.prices_ready(self.min_prices):
            self.carbon_future_data.reset()
            return Universe.Unchanged
        
        selected: List[Fundamental] = [
            x for x in fundamental 
            if x.HasFundamentalData 
            and x.MarketCap != 0
            and x.AssetClassification.MorningstarIndustryGroupCode 
        ]
        if len(selected) > self.fundamental_count:
            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
        high_emission_stocks: List[Fundamental] = []
        clean_stocks: List[Fundamental] = []
        for stock in selected:
            industry_group_code: int = stock.AssetClassification.MorningstarIndustryGroupCode
            if industry_group_code in self.high_emission_industries:
                high_emission_stocks.append(stock)
            else:
                clean_stocks.append(stock)
        
        long_leg: List[Fundamental] = []
        short_leg: List[Fundamental] = []
        climate_monthly_change: float = self.climate_change_data.get_monthly_change()
        carbon_monthly_change: float = self.carbon_future_data.get_monthly_change()
        if climate_monthly_change > 0 and carbon_monthly_change < 0:
            long_leg = high_emission_stocks
            short_leg = clean_stocks
        elif climate_monthly_change < 0 and carbon_monthly_change > 0:
            long_leg = clean_stocks
            short_leg = high_emission_stocks
        if len(long_leg) != 0 and len(short_leg) != 0:
            for i, portfolio in enumerate([long_leg, short_leg]):
                mc_sum: float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
                for stock in portfolio:
                    self.weights[stock.Symbol] = ((-1)**i) * stock.MarketCap / mc_sum
        return list(self.weights.keys())
        
    def OnData(self, slice: Slice) -> None:
        curr_date:datetime.date = self.Time.date()
        if slice.contains_key(self.climate_change) and slice[self.climate_change]:
            search_value: float = slice[self.climate_change].Value
            self.climate_change_data.update(curr_date, search_value)
        if slice.contains_key(self.carbon_future) and slice[self.carbon_future]:
            price: float = slice[self.carbon_future].Value
            self.carbon_future_data.update(curr_date, price)
        # rebalance monthly
        if not self.selection_flag:
            return
        self.selection_flag = False
        
        # trade execution
        portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weights.items() if slice.contains_key(symbol) and slice[symbol]]
        self.SetHoldings(portfolio, True)
        self.weights.clear()
        
    def Selection(self) -> None:
        self.selection_flag = True