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

1from collections import deque
2from AlgorithmImports import *
3import numpy as np
4import statsmodels.api as sm
5from numpy import isnan
6from typing import Dict, List, Deque, Tuple
7class ExpectedInvestmentGrowthwithintheCrosssectionofStocksReturns(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2005, 1, 1)  
10self.SetCash(100000)
11market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
12# Monthly close data.
13self.data:Dict[Symbol, RollingWindow[float]] = {}
14self.period:int = 13
15self.leverage:int = 5
16self.rebalance_month:int = 12
17self.long:List[Symbol] = []
18self.short:List[Symbol] = []
19
20# Regression data.
21self.regression_flag:bool = False
22self.regression_data:Dict[Symbol, Deque[Tuple[float, float, float, float]]] = {}
23self.regression_coefficients:Dict[Symbol, float] = {}
24self.regression_min_period:int = 5  # years
25
26# Last year's capital stock and CAPX data.
27self.last_year_data:Dict[Symbol, Tuple[float, float]] = {}
28
29self.fundamental_count:int = 1000
30self.selection_flag:bool = False
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
35self.settings.daily_precise_end_time = False
36def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
37for security in changes.AddedSecurities:
38    security.SetFeeModel(CustomFeeModel())
39    security.SetLeverage(self.leverage)
40    
41def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
42if not self.selection_flag:
43    return Universe.Unchanged
44# Update the rolling window every month.
45for stock in fundamental:
46    symbol:Symbol = stock.Symbol
47    # Store monthly price.
48    if symbol in self.data:
49        self.data[symbol].Add(stock.AdjustedPrice)
50
51selected:Dict[Symbol, Fundamental] = {x.Symbol: x
52    for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
53        and (not isnan(x.FinancialStatements.CashFlowStatement.CapitalExpenditure.ThreeMonths) and x.FinancialStatements.CashFlowStatement.CapitalExpenditure.ThreeMonths != 0 and
54            not isnan(x.FinancialStatements.BalanceSheet.TotalCapitalization.ThreeMonths) and x.FinancialStatements.BalanceSheet.TotalCapitalization.ThreeMonths != 0 and
55            not isnan(x.FinancialStatements.BalanceSheet.Inventory.ThreeMonths) and x.FinancialStatements.BalanceSheet.Inventory.ThreeMonths != 0 and
56            not isnan(x.FinancialStatements.BalanceSheet.CurrentDeferredTaxesLiabilities.ThreeMonths) and x.FinancialStatements.BalanceSheet.CurrentDeferredTaxesLiabilities.ThreeMonths != 0 and
57            not isnan(x.FinancialStatements.BalanceSheet.CapitalStock.ThreeMonths) and x.FinancialStatements.BalanceSheet.CapitalStock.ThreeMonths != 0 and
58            not isnan(x.FinancialStatements.IncomeStatement.PretaxIncome.ThreeMonths) and x.FinancialStatements.IncomeStatement.PretaxIncome.ThreeMonths != 0 and
59            not isnan(x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths) and x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths != 0 and 
60            not isnan(x.FinancialStatements.BalanceSheet.PreferredStock.ThreeMonths) and x.FinancialStatements.BalanceSheet.PreferredStock.ThreeMonths != 0
61            )],
62        key = lambda x: x.DollarVolume, reverse = True)[:self.fundamental_count]}
63
64predicted_eig:Dict[Symbol, float] = {}
65# Warmup price rolling windows.
66for stock in list(selected.values()):
67    symbol:Symbol = stock.Symbol
68    if symbol not in self.data:
69        if symbol not in self.regression_data:
70            self.regression_data[symbol] = deque()
71        
72        self.data[symbol] = RollingWindow[float](self.period)
73        history:DataFrame = self.History(symbol, self.period * 30, Resolution.Daily)
74        if history.empty:
75            self.Log(f"Not enough data for {symbol} yet.")
76            continue
77        closes:Series = history.loc[symbol].close
78        
79        closes_len:int = len(closes.keys())
80        # Find monthly closes.
81        for index, time_close in enumerate(closes.items()):
82            # index out of bounds check.
83            if index + 1 < closes_len:
84                date_month:int = time_close[0].date().month
85                next_date_month:int = closes.keys()[index + 1].month
86            
87                # Found last day of month.
88                if date_month != next_date_month:
89                    self.data[symbol].Add(time_close[1]) 
90    
91    capital_stock_t:float = stock.FinancialStatements.BalanceSheet.CapitalStock.ThreeMonths
92    capx_t:float = stock.FinancialStatements.CashFlowStatement.CapitalExpenditure.ThreeMonths
93    
94    if symbol not in self.data or not self.data[symbol].IsReady: 
95        if self.regression_flag:
96            self.last_year_data[symbol] = (capital_stock_t, capx_t)
97        continue
98    # Momentum calc.
99    prices:List[float] = [x for x in self.data[symbol]][1:]
100    momentum:float = prices[0] / prices[-1] - 1
101    
102    # Q calc
103    # NOTE: Preffered stock field is not filled in many cases. If it is not filled, ignore it in calculation.
104    pref_stock:float = stock.FinancialStatements.BalanceSheet.PreferredStock.ThreeMonths
105    if pref_stock == 0:  
106        q:float = (stock.FinancialStatements.BalanceSheet.TotalCapitalization.ThreeMonths - stock.FinancialStatements.BalanceSheet.Inventory.ThreeMonths - stock.FinancialStatements.BalanceSheet.CurrentDeferredTaxesLiabilities.ThreeMonths) / capital_stock_t
107    else:
108        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
109    
110    # CF calc.
111    cf:float = stock.FinancialStatements.IncomeStatement.PretaxIncome.ThreeMonths + stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths / capital_stock_t
112                
113    if self.regression_flag:
114        if symbol in self.last_year_data:
115            # EIG calc.
116            capital_stock_t_1:float = self.last_year_data[symbol][0]
117            capx_t_1:float = self.last_year_data[symbol][1]
118            eig:float = np.log(capx_t / capx_t_1)
119            reg_data:Tuple[float, float, float, float] = (eig, momentum, q, cf)
120            if symbol not in self.regression_data:
121                self.regression_data[symbol] = deque()
122                
123            self.regression_data[symbol].append(reg_data)
124            
125            if len(self.regression_data[symbol]) >= self.regression_min_period:
126                # Regression coefficients calc.
127                eigs:List[float] = [float(x[0]) for x in self.regression_data[symbol]]
128                momentums:List[float] = [float(x[1]) for x in self.regression_data[symbol]]
129                qs:List[float] = [float(x[2]) for x in self.regression_data[symbol]]
130                cfs:List[float] = [float(x[3]) for x in self.regression_data[symbol]]
131
132                x:List[float] = [momentums[:-1], qs[:-1], cfs[:-1]]
133                regression_model = self.MultipleLinearRegression(x, eigs[1:])
134                self.regression_coefficients[symbol] = regression_model.params
135        if symbol not in self.last_year_data:
136            self.last_year_data[symbol] = None
137        self.last_year_data[symbol] = (capital_stock_t, capx_t)
138    
139    if symbol in self.regression_coefficients:
140        alpha:float = self.regression_coefficients[symbol][0]
141        betas:np.ndarray = np.array(self.regression_coefficients[symbol][1:])
142        prediction_x:List[float] = [momentum, q, cf]
143        if len(prediction_x) == len(betas):
144            predicted_eig[symbol] = alpha + sum(np.multiply(betas, prediction_x))
145
146if self.regression_flag:
147    self.regression_flag = False
148
149if len(predicted_eig) != 0:
150    eig_values:List[float] = [x[1] for x in predicted_eig.items()]
151    top_decile:float = np.percentile(eig_values, 90)
152    bottom_decile:float = np.percentile(eig_values, 10)
153    self.long:List[Symbol] = [x[0] for x in predicted_eig.items() if x[1] > top_decile]
154    self.short:List[Symbol] = [x[0] for x in predicted_eig.items() if x[1] < bottom_decile]
155# Remove not updated symbols.
156symbols_to_remove:List[Symbol] = []
157for symbol in self.last_year_data:
158    if symbol not in selected:
159        symbols_to_remove.append(symbol)
160for symbol in symbols_to_remove:
161    if symbol in self.last_year_data:
162        del self.last_year_data[symbol]
163    if symbol in self.regression_data:
164        del self.regression_data[symbol]
165
166return self.long + self.short
167def OnData(self, data: Slice) -> None:
168if not self.selection_flag:
169    return
170self.selection_flag = False
171
172# Trade execution.
173targets:List[PortfolioTarget] = []
174for i, portfolio in enumerate([self.long, self.short]):
175    for symbol in portfolio:
176        if symbol in data and data[symbol]:
177            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
178
179self.SetHoldings(targets, True)
180self.long.clear()
181self.short.clear()
182
183def Selection(self) -> None:
184if self.Time.month == self.rebalance_month:
185    self.regression_flag = True    
186self.selection_flag = True
187
188def MultipleLinearRegression(self, x:List[float], y:List[float]):
189x:np.ndarray = np.array(x).T
190x = sm.add_constant(x)
191result = sm.OLS(endog=y, exog=x).fit()
192return result
193
194# Custom fee model
195class CustomFeeModel(FeeModel):
196def GetOrderFee(self, parameters):
197fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
198return OrderFee(CashAmount(fee, "USD"))