气候贝塔与共同基金

“通过气候贝塔交易共同基金，做多最低贝塔五分位，做空最高贝塔五分位，使用等权重投资组合，每月重新平衡，排除资产低于1500万美元的基金。”

资产类别: 基金 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 气候、贝塔、共同基金

I. 策略概要

投资范围包括CRSP无幸存者偏差共同基金数据库中的共同基金，不包括总净资产低于1500万美元的基金。关键变量气候贝塔衡量基金与气候变化新闻指数创新之间的协方差。气候贝塔通过共同基金超额回报与Crimson Hexagon（CH）气候变化新闻指数变化的24个月回归进行估计。基金根据气候贝塔分为五分位。该策略做多最低五分位（最低气候贝塔），做空最高五分位（最高气候贝塔），采用等权重投资组合，每月重新平衡。

II. 策略合理性

该策略的功能围绕气候贝塔展开，其中气候贝塔最高的基金表现优于气候贝塔最低的基金。高气候贝塔基金依赖于具有高气候贝塔的股票，这些股票在气候意识提高期间经历价值增长。在股票层面，那些具有更高气候贝塔的股票由于更好的业务基本面和投资者对气候风险对冲股票的需求增加而产生更高的月度回报。Fama-MacBeth回归证实了基金层面气候贝塔与回报之间的正相关关系，即使在考虑了可观察的基金特征和风格之后，这表明气候贝塔有效地捕捉了由气候风险意识驱动的回报。

III. 来源论文

Climate sensitivity and mutual fund performance [点击查看论文]

• Thang Ho。布拉德福德大学管理学院

<摘要>

在对气候变化的担忧日益加剧，可能影响投资者投资组合公司风险和回报的情况下，积极型投资者可能会分散气候风险敞口。我们计算了共同基金与市场范围气候变化新闻指数的协方差，发现高（正）气候贝塔基金每月在风险调整后的基础上比低（负）气候贝塔基金高出0.24%。高气候贝塔基金将其持仓倾向于具有高潜力对冲气候变化的股票。在横截面上，此类股票产生更高的超额回报，这在我们的样本期内是由更大的定价压力和卓越的财务表现驱动的。

IV. 回测表现

年化回报	2.55%
波动率	2.27%
β值	-0.028
夏普比率	1.12
索提诺比率	N/A
最大回撤	N/A
胜率	49%

V. 完整的 Python 代码

import statsmodels.api as sm
from AlgorithmImports import *
from data_tools import CustomFeeModel, MutualFund, SymbolData, ClimateChangeData, ClimateChange
from typing import Dict, List
# endregion
class ClimateBetaAndMutualFunds(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2004, 1, 1)
        self.SetCash(100000)
        
        self.leverage:int = 5
        self.period:int = 24
        self.quantile:int = 5
        self.max_missing_days:int = 5
        self.max_missing_days_climate_change:int = 40
        self.min_prices:int = 15
        self.long_only_flag:bool = False
        self.recent_month:int = -1
        self.data:Dict[Symbol, SymbolData] = {}
        self.climate_change:Symbol = self.AddData(ClimateChange, 'CLIMATE_CHANGE', Resolution.Daily).Symbol
        self.climate_change_data:ClimateChangeData = ClimateChangeData(self.period)
        self.symbol_count:int = 500
        ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/mutual_funds/500_mutual_funds_tickers.csv')
        ticker_file_str = ticker_file_str.replace('\r', '')
        self.tickers:List[str] = ticker_file_str.split('\n')[:self.symbol_count]
        for t in self.tickers:
            data = self.AddData(MutualFund, t, Resolution.Daily)
            data.SetFeeModel(CustomFeeModel())
            data.SetLeverage(self.leverage)
            self.data[data.Symbol] = SymbolData(self.period)
    def OnData(self, data: Slice) -> None:
        curr_date:datetime.date = self.Time.date()
        
        # store data
        for symbol, symbol_data in self.data.items():
            if symbol in data and data[symbol] and data[symbol].Value != 0:
                symbol_data.update_daily_prices(curr_date, data[symbol].Value)
        if self.climate_change in data and data[self.climate_change]:
            search_value:float = data[self.climate_change].Value
            self.climate_change_data.update(curr_date, search_value)
        # rebalance
        if self.recent_month == self.Time.month:
            return
        self.recent_month = self.Time.month
        # data is still comming in
        if not self.climate_change_data.data_still_coming(curr_date, self.max_missing_days_climate_change):
            self.climate_change_data.reset()
        x:List[float]|None = self.climate_change_data.get_monthly_changes() \
            if self.climate_change_data.is_ready() else None
        beta_values:Dict[Symbol, float] = {}
        # run regression
        for symbol, symbol_data in self.data.items():
            if not symbol_data.prices_still_coming(curr_date, self.max_missing_days):
                symbol_data.reset()
                continue
            if symbol_data.daily_prices_ready(self.min_prices):
                symbol_data.update_monthly_returns()
            if x != None and symbol_data.monthly_returns_ready() and symbol in data \
                and data[symbol] and data[symbol].Value != 0:
                monthly_returns:List[float] = symbol_data.get_monthly_returns()
                regression_model = self.MultipleLinearRegression(x, monthly_returns)
                beta:float = regression_model.params[1]
                beta_values[symbol] = beta
            symbol_data.reset_daily_prices()
        if len(beta_values) < self.quantile:
            self.Liquidate()
            return
        # sort by beta
        quantile:int = int(len(beta_values) / self.quantile)
        sorted_by_beta:List[Symbol] = [x[0] for x in sorted(beta_values.items(), key=lambda item: item[1])]
        short_leg:List[Symbol] = [] if self.long_only_flag else sorted_by_beta[-quantile:]
        long_leg:List[Symbol] = sorted_by_beta[:quantile]
        # trade execution
        invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
        for symbol in invested:
            if symbol not in long_leg+ short_leg:
                self.Liquidate(symbol)
        for symbol in long_leg:
            self.SetHoldings(symbol, 1 / quantile)
        for symbol in short_leg:
            self.SetHoldings(symbol, -1 / quantile)
    def MultipleLinearRegression(self, x:list, y:list):
        x:np.array = np.array(x).T
        x = sm.add_constant(x)
        result = sm.OLS(endog=y, exog=x).fit()
        return result