“该策略基于员工情绪指数(StESSt^{ES}),通过Glassdoor的调查数据计算每月的正面与负面评分差异,并得出公司的情绪平均值。投资者使用回归方程预测未来收益(Rt+1=α+β⋅StES),并在市场投资组合与无风险债券之间进行资产分配。投资组合每月重新调整。”

策略概述

每月的员工情绪指数 StESSt^{ES}StES 是根据Glassdoor的调查数据为公司t定义的,具体计算方式如公式(1)所示:某个月公司i的正面总体评分(四星和五星)的评价数量减去负面评分(一星和二星)的评价数量,再除以公司i该月的总评价数量。员工情绪指数则为各公司的等权重情绪平均值。

根据样本外预测(使用回归方程(7)的信息:Rt+1=α+β⋅StESR_{t+1} = \alpha + \beta \cdot St^{ES}Rt+1​=α+β⋅StES),投资者将优化其效用,并在市场投资组合和无风险债券之间分配资产。投资组合每月重新调整。

策略合理性

陈健、唐国豪、姚佳权和周国富(2022年10月)的论文提供了强有力的统计证据,表明员工情绪可以负面预测后续的股票市场回报。他们提出了一些经济解释,包括延展性偏差(员工倾向于将近期公司表现推测至未来,期望公司继续表现良好)、劳动力市场的不流动性增加了公司的雇佣成本(如工资)、员工情绪(对难以估值的公司影响更强)以及特征排序投资组合。

论文来源

Employee Sentiment and Stock Returns [点击浏览原文]

<摘要>

我们提出了员工情绪指数,补充了投资者情绪和管理者情绪指数的研究,发现高员工情绪显著预测了未来较低的月度(或每周)市场回报,且在样本内外都具有统计显著性。这种可预测性能够为均值-方差投资者在资产配置中带来显著的经济收益。研究还发现,员工情绪的影响在总部所在地工作的员工以及经验较少的员工中更为明显。推动这一预测性的经济动力是独特的:高员工情绪导致了由于劳动力市场流动性不足带来的当期工资增长,从而导致公司现金流下降,进而影响股票回报。

回测表现

年化收益率7.5%
波动率10%
Beta0.471
夏普比率0.75
索提诺比率N/A
最大回撤N/A
胜率43%

完整python代码

from AlgorithmImports import *
from typing import List
import statsmodels.api as sm
import numpy as np
# endregion

class EmployeeSentimentandStockReturns(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2007, 5, 1) # BIL inception date
        self.SetCash(100000)

        self.leverage:int = 5

        self.risk_aversion:float = 3.
        self.allocation_limits:List[float] = [-0.5, 1.5]

        self.tickers:List[str] = [
            'A', 'AA', 'AAPL', 'AXP', 'BA', 'BAC', 'CAT', 'CSCO', 'CVX', 'DD', 
            'DIS', 'GE', 'HD', 'HPQ', 'IBM', 'INTC', 'JNJ', 'KFT', 'KO', 'MRK',
            'MSFT', 'PFE', 'PG', 'TRV', 'UTX', 'VZ', 'XOM'] # MCD, WMT

        self.negative_scores:List[float] = [1., 2.]
        self.positive_scores:List[float] = [4., 5.]

        self.reviews_by_ticker:Dict[str, List[Tuple[datetime.date, float]]] = {}

        self.last_review_date:datetime.date = datetime(1,1,1).date()

        # import employee review data
        for ticker in self.tickers:
            csv_string_file:str = self.Download(f"data.quantpedia.com/backtesting_data/economic/indeed_review/{ticker}.csv")
            lines:List[str] = csv_string_file.split('\r\n')
            for line in lines[1:]: # skip header
                if line == '': continue

                line_split:List[str] = line.split(';')
                date:datetime.date = datetime.strptime(line_split[0], "%d.%m.%Y").date()
                score:float = float(line_split[1])

                # store date of the review with associated score
                if ticker not in self.reviews_by_ticker:
                    self.reviews_by_ticker[ticker] = []
                
                self.reviews_by_ticker[ticker].append((date, score))
                
                # mark up last available review date
                if date > self.last_review_date:
                    self.last_review_date = date

        # subscribe price data
        data:Equity = self.AddEquity("SPY", Resolution.Daily)
        data.SetLeverage(self.leverage)
        self.market:Symbol = data.Symbol

        data:Equity = self.AddEquity("BIL", Resolution.Daily)
        data.SetLeverage(self.leverage)
        self.t_bills:Symbol = data.Symbol

        # sentiment data
        self.m_period:int = 12
        self.sentiment_data:SentimentData = SentimentData(self.m_period)

        self.SetWarmUp(self.m_period * 30, Resolution.Daily)
                                                            
        self.recent_month:int = -1

    def OnData(self, data: Slice) -> None:
        # market data are present in the algorithm
        if self.market not in data or not data[self.market]:
            return

        # monthly rebalance
        if self.Time.month == self.recent_month:
            return
        self.recent_month = self.Time.month

        last_month:datetime.datetime = self.Time.replace(day=1) - timedelta(days=1)
        
        employee_sentiment:float = 0.
        for ticker, reviews_list in self.reviews_by_ticker.items():
            n_positive_reviews:float = len([x for x in reviews_list if x[1] in self.positive_scores and x[0].month == last_month.month and x[0].year == last_month.year])
            n_negative_reviews:float = len([x for x in reviews_list if x[1] in self.negative_scores and x[0].month == last_month.month and x[0].year == last_month.year])
            firm_employee_sentiment:float = (n_positive_reviews - n_negative_reviews) / (n_positive_reviews + n_negative_reviews) if (n_positive_reviews + n_negative_reviews) != 0. else 0
            employee_sentiment += firm_employee_sentiment
        
        # update sentiment and market data
        employee_sentiment /= len(self.reviews_by_ticker)
        self.sentiment_data.update_data(data[self.market].Value, employee_sentiment)
        
        if self.IsWarmingUp: return

        # no more review data available
        if self.Time.date() > self.last_review_date:
            self.Liquidate()
            return

        if self.sentiment_data.is_ready():
            x:Tuple[np.ndarray, np.ndarray] = self.sentiment_data.get_regression_data()
            model = self.multiple_linear_regression(x[1][:-1], x[0][1:])
            forecast_return:float = model.predict([1, x[1][-1]])[0]
            forecast_variance:float = np.std(x[0]) ** 2 # * np.sqrt(12)
            w_t:float = (1. / self.risk_aversion) * (forecast_return / forecast_variance)
            w_t = min(max(w_t, self.allocation_limits[0]), self.allocation_limits[1])
            t_bill_w:float = 1. - w_t

            self.SetHoldings(self.market, w_t)
            self.SetHoldings(self.t_bills, t_bill_w)

    def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
        x:np.ndarray = np.array(x).T
        x = sm.add_constant(x)
        result = sm.OLS(endog=y, exog=x).fit()
        return result

class SentimentData():
    def __init__(self, period:int) -> None:
        self._period:int = period
        self._market_prices:RollingWindow = RollingWindow[float](period + 1)
        self._sentiment_index:RollingWindow = RollingWindow[float](period)

    def is_ready(self) -> bool:
        return self._market_prices.IsReady and self._sentiment_index.IsReady
    
    def update_data(self, price:float, sentiment_index_value:float) -> None:
        self._market_prices.Add(price)
        self._sentiment_index.Add(sentiment_index_value)
    
    def get_regression_data(self) -> Tuple[np.ndarray, np.ndarray]:
        prices:np.ndarray = np.array(list(self._market_prices))
        returns:np.ndarray = prices[:-1] / prices[1:] - 1

        sentiment_index:np.ndarray = np.array(list(self._sentiment_index))
        return returns[::-1], sentiment_index[::-1]

Leave a Reply

Discover more from Quant Buffet

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

Continue reading