股票中的特质因子动量效应

1from scipy import stats
2from AlgorithmImports import *
3from typing import List, Deque, Tuple
4from collections import deque
5class IdiosyncraticMomentumStocks(QCAlgorithm):
6def Initialize(self) -> None:
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000) 
9# Daily price data.
10self.data:Dict[Symbol, RollingWindow] = {}
11self.period:int = 21
12self.quantile:int = 5
13self.leverage:int = 5
14self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
15self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
16self.data[self.symbol] = RollingWindow[float](self.period)
17self.regression_period:int = 36
18self.regression_data:Dict[Symbol, Tuple] = {}
19
20# Monthly residuals for stocks.
21self.residuals_period:int = 12
22self.residual:Dict[Symbol, RollingWindow] = {}
23
24self.long:List[Symbol] = []
25self.short:List[Symbol] = []
26
27self.fundamental_count:int = 1000
28self.fundamental_sorting_key = lambda x: x.DollarVolume
29self.selection_flag:bool = False
30self.UniverseSettings.Resolution = Resolution.Daily
31self.AddUniverse(self.FundamentalSelectionFunction)
32self.settings.daily_precise_end_time = False
33self.settings.minimum_order_margin_portfolio_percentage = 0.
34self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)            
35def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
36for security in changes.RemovedSecurities:
37    symbol:Symbol = security.Symbol
38    
39    if symbol in self.regression_data:
40        del self.regression_data[symbol]
41        
42    if symbol in self.residual:
43        del self.residual[symbol]
44        
45for security in changes.AddedSecurities:
46    symbol:Symbol = security.Symbol
47    security.SetFeeModel(CustomFeeModel())
48    security.SetLeverage(self.leverage)
49    
50def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
51# Update the rolling window every day.
52for stock in fundamental:
53    symbol:Symbol = stock.Symbol
54    # Store daily price.
55    if symbol in self.data:
56        self.data[symbol].Add(stock.AdjustedPrice)
57if not self.selection_flag:
58    return Universe.Unchanged
59# selected = [x.Symbol for x in fundamental if x.HasFundamentalData and x.Market == 'usa']
60selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchange_codes]
61if len(selected) > self.fundamental_count:
62    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
63idiosyncratic_momentum:Dict[Symbol, float] = {}
64
65# Warmup price rolling windows.
66for stock in selected:
67    symbol:Symbol = stock.Symbol
68    if symbol not in self.data:
69        self.data[symbol] = RollingWindow[float](self.period)
70        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
71        if history.empty:
72            self.Log(f"Not enough data for {symbol} yet.")
73            continue
74        closes:Series = history.loc[symbol].close
75        for time, close in closes.items():
76            self.data[symbol].Add(close)
77
78    # Market data is not ready.
79    if not self.data[self.symbol].IsReady: 
80        continue
81    market_excess_return:float = self.data[self.symbol][0] / self.data[self.symbol][self.period-1] - 1
82
83    if not self.data[symbol].IsReady:
84        continue
85    stock_excess_return:float = self.data[symbol][0] / self.data[symbol][self.period-1] - 1
86    
87    # store regression data
88    if symbol not in self.regression_data:
89        self.regression_data[symbol] = deque(maxlen = self.regression_period)
90    self.regression_data[symbol].append((market_excess_return, stock_excess_return))
91    # Regression.
92    if len(self.regression_data[symbol]) == self.regression_data[symbol].maxlen:
93        # Y = α + (β ∗ X)
94        # intercept = alpha
95        # slope = beta
96        market_excess_returns:List[float] = [x[0] for x in self.regression_data[symbol]]
97        stock_excess_returns:List[float] = [x[1] for x in self.regression_data[symbol]]
98        
99        slope, intercept, r_value, p_value, std_err = stats.linregress(market_excess_returns, stock_excess_returns)
100        
101        # Calculate every residual for recent months.
102        # residuals = []
103        # for idx, x in enumerate(market_excess_returns):
104        #     yfit = intercept + (slope * x)
105        #     residuals.append(yfit - stock_excess_returns[idx])
106        # idiosyncratic_momentum[symbol] = sum(residuals[:-2])
107        
108        # Calculate only latest residual.
109        actual_value:float = stock_excess_returns[-1]
110        estimate_value:float = intercept + (slope * market_excess_returns[-1])
111        residual:float = actual_value - estimate_value
112        
113        # store residual data
114        if symbol not in self.residual:
115            self.residual[symbol] = RollingWindow[float](self.residuals_period)
116        
117        if self.residual[symbol].IsReady:
118            # idiosyncratic_momentum[symbol] = self.residual[symbol][1] / self.residual[symbol][self.residuals_period-1] - 1
119            idiosyncratic_momentum[symbol] = sum([x for x in self.residual[symbol]][1:])
120        
121        self.residual[symbol].Add(residual)
122
123if len(idiosyncratic_momentum) >= self.quantile:
124    sorted_by_idiosyncratic_momentum:List[Tuple[Symbol, float]] = sorted(idiosyncratic_momentum.items(), key = lambda x: x[1], reverse = True)
125    quintile:int = int(len(sorted_by_idiosyncratic_momentum) / 5)
126    self.long:List[Symbol] = [x[0] for x in sorted_by_idiosyncratic_momentum[:quintile]]
127    self.short:List[Symbol] = [x[0] for x in sorted_by_idiosyncratic_momentum[-quintile:]]
128
129return self.long + self.short
130
131def OnData(self, data: Slice) -> None:
132if not self.selection_flag:
133    return
134self.selection_flag = False        
135# Trade execution.
136targets:List[PortfolioTarget] = []
137for i, portfolio in enumerate([self.long, self.short]):
138    for symbol in portfolio:
139        if symbol in data and data[symbol]:
140            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
141
142self.SetHoldings(targets, True)
143    
144self.long.clear()
145self.short.clear()
146def Selection(self) -> None:
147self.selection_flag = True
148# Custom fee model
149class CustomFeeModel(FeeModel):
150def GetOrderFee(self, parameters):
151fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
152return OrderFee(CashAmount(fee, "USD"))