预期投资增长与股票回报的横截面关系

“该策略投资于十分位数EIG投资组合，根据使用动量、现金流和市场价值的预测回归对股票进行排序。投资组合做多高EIG十分位数，做空低EIG十分位数。”

资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 增长

I. 策略概要

投资范围包括公司子样本的十分位数EIG投资组合，不包括投资组合形成时小于纽约证券交易所规模截止值20%的股票。每个月，投资者通过预测回归公式计算EIG因子，使用动量、现金流和市场价值（q）作为自变量。然后，根据股票的EIG值将股票分为十分位数。投资者做多十分位数10（最高EIG），做空十分位数1（最低EIG）。投资组合每月重新平衡，并为头寸分配相等的权重。

II. 策略合理性

对于高EIG溢价，有两种解释：基于风险的和基于行为的。基于风险的解释是顺周期的，低EIG股票具有负消费贝塔，高EIG股票具有正消费贝塔，这使得低EIG股票成为对冲商业周期波动的工具。相反，高EIG股票具有更高的风险溢价。基于行为的解释认为，低EIG股票类似于彩票类资产，由于投资者对此类资产的偏好，可能被高估，从而导致未来回报较低。高信息不确定性通过彩票偏好等偏差影响投资决策，从而加剧了这种情况。

III. 来源论文

Expected Investment Growth and the Cross Section of Stock Returns [点击查看论文]

• 李俊 和 王慧君，德克萨斯大学达拉斯分校，奥本大学

<摘要>

我们提出了一个衡量公司投资计划的指标，即预期投资增长（EIG）。我们记录了一个稳健的发现，即高EIG的公司比低EIG的公司具有更大的未来投资增长，并获得显著更高的回报，这无法被领先的因子模型完全解释。进一步的分析表明，EIG与困境风险密切相关，尤其是在一年内的短期范围内。与传统困境风险指标的详细比较突出了在调和文献中记录的困境溢价相反符号时，短期和长期范围之间的区别。

IV. 回测表现

年化回报	14.52%
波动率	18.62%
β值	-0.045
夏普比率	0.78
索提诺比率	-0.337
最大回撤	N/A
胜率	49%

V. 完整的 Python 代码

from collections import deque
from AlgorithmImports import *
import numpy as np
import statsmodels.api as sm
from numpy import isnan
from typing import Dict, List, Deque, Tuple
class ExpectedInvestmentGrowthwithintheCrosssectionofStocksReturns(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2005, 1, 1)  
        self.SetCash(100000)
        market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        # Monthly close data.
        self.data:Dict[Symbol, RollingWindow[float]] = {}
        self.period:int = 13
        self.leverage:int = 5
        self.rebalance_month:int = 12
        self.long:List[Symbol] = []
        self.short:List[Symbol] = []
        
        # Regression data.
        self.regression_flag:bool = False
        self.regression_data:Dict[Symbol, Deque[Tuple[float, float, float, float]]] = {}
        self.regression_coefficients:Dict[Symbol, float] = {}
        self.regression_min_period:int = 5  # years
        
        # Last year's capital stock and CAPX data.
        self.last_year_data:Dict[Symbol, Tuple[float, float]] = {}
        
        self.fundamental_count:int = 1000
        self.selection_flag:bool = False
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
        self.settings.daily_precise_end_time = False
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
            security.SetLeverage(self.leverage)
            
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        if not self.selection_flag:
            return Universe.Unchanged
        # Update the rolling window every month.
        for stock in fundamental:
            symbol:Symbol = stock.Symbol
            # Store monthly price.
            if symbol in self.data:
                self.data[symbol].Add(stock.AdjustedPrice)
    
        selected:Dict[Symbol, Fundamental] = {x.Symbol: x
            for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
                and (not isnan(x.FinancialStatements.CashFlowStatement.CapitalExpenditure.ThreeMonths) and x.FinancialStatements.CashFlowStatement.CapitalExpenditure.ThreeMonths != 0 and
                    not isnan(x.FinancialStatements.BalanceSheet.TotalCapitalization.ThreeMonths) and x.FinancialStatements.BalanceSheet.TotalCapitalization.ThreeMonths != 0 and
                    not isnan(x.FinancialStatements.BalanceSheet.Inventory.ThreeMonths) and x.FinancialStatements.BalanceSheet.Inventory.ThreeMonths != 0 and
                    not isnan(x.FinancialStatements.BalanceSheet.CurrentDeferredTaxesLiabilities.ThreeMonths) and x.FinancialStatements.BalanceSheet.CurrentDeferredTaxesLiabilities.ThreeMonths != 0 and
                    not isnan(x.FinancialStatements.BalanceSheet.CapitalStock.ThreeMonths) and x.FinancialStatements.BalanceSheet.CapitalStock.ThreeMonths != 0 and
                    not isnan(x.FinancialStatements.IncomeStatement.PretaxIncome.ThreeMonths) and x.FinancialStatements.IncomeStatement.PretaxIncome.ThreeMonths != 0 and
                    not isnan(x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths) and x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths != 0 and 
                    not isnan(x.FinancialStatements.BalanceSheet.PreferredStock.ThreeMonths) and x.FinancialStatements.BalanceSheet.PreferredStock.ThreeMonths != 0
                    )],
                key = lambda x: x.DollarVolume, reverse = True)[:self.fundamental_count]}
        
        predicted_eig:Dict[Symbol, float] = {}
        # Warmup price rolling windows.
        for stock in list(selected.values()):
            symbol:Symbol = stock.Symbol
            if symbol not in self.data:
                if symbol not in self.regression_data:
                    self.regression_data[symbol] = deque()
                
                self.data[symbol] = RollingWindow[float](self.period)
                history:dataframe = self.History(symbol, self.period * 30, Resolution.Daily)
                if history.empty:
                    self.Log(f"Not enough data for {symbol} yet.")
                    continue
                closes:Series = history.loc[symbol].close
                
                closes_len:int = len(closes.keys())
                # Find monthly closes.
                for index, time_close in enumerate(closes.items()):
                    # index out of bounds check.
                    if index + 1 < closes_len:
                        date_month:int = time_close[0].date().month
                        next_date_month:int = closes.keys()[index + 1].month
                    
                        # Found last day of month.
                        if date_month != next_date_month:
                            self.data[symbol].Add(time_close[1]) 
            
            capital_stock_t:float = stock.FinancialStatements.BalanceSheet.CapitalStock.ThreeMonths
            capx_t:float = stock.FinancialStatements.CashFlowStatement.CapitalExpenditure.ThreeMonths
            
            if symbol not in self.data or not self.data[symbol].IsReady: 
                if self.regression_flag:
                    self.last_year_data[symbol] = (capital_stock_t, capx_t)
                continue
            # Momentum calc.
            prices:List[float] = [x for x in self.data[symbol]][1:]
            momentum:float = prices[0] / prices[-1] - 1
            
            # Q calc
            # NOTE: Preffered stock field is not filled in many cases. If it is not filled, ignore it in calculation.
            pref_stock:float = stock.FinancialStatements.BalanceSheet.PreferredStock.ThreeMonths
            if pref_stock == 0:  
                q:float = (stock.FinancialStatements.BalanceSheet.TotalCapitalization.ThreeMonths - stock.FinancialStatements.BalanceSheet.Inventory.ThreeMonths - stock.FinancialStatements.BalanceSheet.CurrentDeferredTaxesLiabilities.ThreeMonths) / capital_stock_t
            else:
                q:float = (stock.FinancialStatements.BalanceSheet.TotalCapitalization.ThreeMonths + stock.FinancialStatements.BalanceSheet.PreferredStock.ThreeMonths - stock.FinancialStatements.BalanceSheet.Inventory.ThreeMonths - stock.FinancialStatements.BalanceSheet.CurrentDeferredTaxesLiabilities.ThreeMonths) / capital_stock_t
            
            # CF calc.
            cf:float = stock.FinancialStatements.IncomeStatement.PretaxIncome.ThreeMonths + stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths / capital_stock_t
                        
            if self.regression_flag:
                if symbol in self.last_year_data:
                    # EIG calc.
                    capital_stock_t_1:float = self.last_year_data[symbol][0]
                    capx_t_1:float = self.last_year_data[symbol][1]
                    eig:float = np.log(capx_t / capx_t_1)
                    reg_data:Tuple[float, float, float, float] = (eig, momentum, q, cf)
                    if symbol not in self.regression_data:
                        self.regression_data[symbol] = deque()
                        
                    self.regression_data[symbol].append(reg_data)
                    
                    if len(self.regression_data[symbol]) >= self.regression_min_period:
                        # Regression coefficients calc.
                        eigs:List[float] = [float(x[0]) for x in self.regression_data[symbol]]
                        momentums:List[float] = [float(x[1]) for x in self.regression_data[symbol]]
                        qs:List[float] = [float(x[2]) for x in self.regression_data[symbol]]
                        cfs:List[float] = [float(x[3]) for x in self.regression_data[symbol]]
        
                        x:List[float] = [momentums[:-1], qs[:-1], cfs[:-1]]
                        regression_model = self.MultipleLinearRegression(x, eigs[1:])
                        self.regression_coefficients[symbol] = regression_model.params
                if symbol not in self.last_year_data:
                    self.last_year_data[symbol] = None
                self.last_year_data[symbol] = (capital_stock_t, capx_t)
            
            if symbol in self.regression_coefficients:
                alpha:float = self.regression_coefficients[symbol][0]
                betas:np.ndarray = np.array(self.regression_coefficients[symbol][1:])
                prediction_x:List[float] = [momentum, q, cf]
                if len(prediction_x) == len(betas):
                    predicted_eig[symbol] = alpha + sum(np.multiply(betas, prediction_x))
        
        if self.regression_flag:
            self.regression_flag = False
        
        if len(predicted_eig) != 0:
            eig_values:List[float] = [x[1] for x in predicted_eig.items()]
            top_decile:float = np.percentile(eig_values, 90)
            bottom_decile:float = np.percentile(eig_values, 10)
            self.long:List[Symbol] = [x[0] for x in predicted_eig.items() if x[1] > top_decile]
            self.short:List[Symbol] = [x[0] for x in predicted_eig.items() if x[1] < bottom_decile]
        # Remove not updated symbols.
        symbols_to_remove:List[Symbol] = []
        for symbol in self.last_year_data:
            if symbol not in selected:
                symbols_to_remove.append(symbol)
        for symbol in symbols_to_remove:
            if symbol in self.last_year_data:
                del self.last_year_data[symbol]
            if symbol in self.regression_data:
                del self.regression_data[symbol]
        
        return self.long + self.short
    def OnData(self, data: Slice) -> None:
        if not self.selection_flag:
            return
        self.selection_flag = False
        
        # Trade execution.
        targets:List[PortfolioTarget] = []
        for i, portfolio in enumerate([self.long, self.short]):
            for symbol in portfolio:
                if symbol in data and data[symbol]:
                    targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
        
        self.SetHoldings(targets, True)
        self.long.clear()
        self.short.clear()
    
    def Selection(self) -> None:
        if self.Time.month == self.rebalance_month:
            self.regression_flag = True    
        self.selection_flag = True
    
    def MultipleLinearRegression(self, x:List[float], y:List[float]):
        x:np.ndarray = np.array(x).T
        x = sm.add_constant(x)
        result = sm.OLS(endog=y, exog=x).fit()
        return result
    
# Custom fee model
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))