股市中的异常换手效应

1from AlgorithmImports import *
2import statsmodels.api as sm
3from typing import List, Dict
4from numpy import isnan
5class AbnormalTurnoverEffectInTheStockMarket(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9
10self.leverage:int = 5
11self.quantile:int = 5
12self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']	
13self.symbol:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
14
15self.data:Dict[Symbol, SymbolData] = {}
16self.long:List[Symbol] = []
17self.short:List[Symbol] = []
18
19self.period:int = 21 # need n of daily volumes
20self.regression_period:int = 12 # need n of last turnovers and n * (self.turnover_period - 1) for regression
21
22self.fundamental_sorting_key = lambda x: x.DollarVolume
23self.fundamental_count:int = 500 # selecting n stocks by dollar volume from fundamentalSelectionFunction
24self.selection_flag:bool = False
25self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
26self.UniverseSettings.Resolution = Resolution.Daily
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
29self.settings.daily_precise_end_time = False
30
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34    security.SetLeverage(self.leverage)
35    
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
37# update stocks volumes every day
38for stock in fundamental:
39    symbol = stock.Symbol
40    
41    # update stock's volume
42    if symbol in self.data:
43        self.data[symbol].update(stock.Volume)
44
45# rebalance monthly
46if not self.selection_flag:
47    return Universe.Unchanged
48
49# select stocks, which had spin off
50selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and not isnan(x.EarningReports.BasicAverageShares.ThreeMonths > 0) and x.EarningReports.BasicAverageShares.ThreeMonths > 0 \
51                    and x.SecurityReference.ExchangeId in self.exchange_codes]
52if len(selected) > self.fundamental_count:
53    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
54u_turn:Dict[Symbol, float] = {} # storing U-TURN of filtered stocks
55
56# warm up selected symbols
57for stock in selected:
58    symbol:Symbol = stock.Symbol
59    if symbol not in self.data:
60        self.data[symbol] = SymbolData(self.period, self.regression_period)
61        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
62        if history.empty:
63            self.Log(f"Not enough data for {symbol} yet")
64            continue
65        volumes:Closes = history.loc[symbol].volume
66        for _, volume in volumes.items():
67            self.data[symbol].update(volume)
68    
69    if not self.data[symbol].is_ready():
70        continue
71    
72    # check if there is enough data for regression
73    if self.data[symbol].turnovers_ready():
74        # get x and y for regression
75        x, y = self.data[symbol].get_regression_data(self.regression_period)
76        
77        # calculate regression
78        regression_model = self.MultipleLinearRegression(x, y)
79        
80        # get last residual
81        last_resid:float = regression_model.resid[-1]
82        # calculate std of all regression residuals
83        resid_std:float = np.std(regression_model.resid)
84        
85        # calculate and store stock's U-TURN
86        u_turn[symbol] = last_resid / resid_std
87    
88    # get stock's volumes for last month
89    monthly_volume:float = self.data[symbol].monthly_volume()
90    # get stock's shares oustanding
91    shares_outstanding:float = stock.EarningReports.BasicAverageShares.ThreeMonths
92    
93    # calculate and update turnovers for current stock
94    self.data[symbol].update_turnovers(monthly_volume / shares_outstanding)
95    
96# check if there are enough stocks for quintile selection
97if len(u_turn) < self.quantile:
98    return Universe.Unchanged
99
100# sort stocks by U-TURN
101quintile:int = int(len(u_turn) / self.quantile)
102sorted_by_u_turn:List[Symbol] = [x[0] for x in sorted(u_turn.items(), key=lambda item: item[1])]
103
104# select long stocks
105self.long = sorted_by_u_turn[:quintile]
106# select short stocks
107self.short = sorted_by_u_turn[-quintile:]
108
109return [x for x in self.long + self.short]
110def OnData(self, data: Slice) -> None:
111if not self.selection_flag:
112    return
113self.selection_flag = False
114
115# trade execution
116targets:List[PortfolioTarget] = []
117for i, portfolio in enumerate([self.long, self.short]):
118    for symbol in portfolio:
119        if symbol in data and data[symbol]:
120            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
121
122self.SetHoldings(targets, True)
123    
124self.long.clear()
125self.short.clear()
126
127def MultipleLinearRegression(self, x, y):
128# x = np.array(x).T
129# x = sm.add_constant(x)
130result = sm.OLS(endog=y, exog=x).fit()
131return result
132
133def Selection(self) -> None:
134self.selection_flag = True
135
136class SymbolData():
137def __init__(self, period:int, regression_period:int):
138self.volumes:RollingWindow = RollingWindow[float](period)
139# storing turnovers for regression
140self.turnovers:RollingWindow = RollingWindow[float](regression_period * 2)
141    
142def update(self, volume:float):
143self.volumes.Add(volume)
144
145def update_turnovers(self, turnover:float):
146self.turnovers.Add(turnover)
147    
148def is_ready(self) -> bool:
149return self.volumes.IsReady
150
151def turnovers_ready(self) -> bool:
152return self.turnovers.IsReady
153
154def monthly_volume(self) -> float:
155volumes = [x for x in self.volumes]
156return sum(volumes)
157    
158def get_regression_data(self, regression_period:int):
159# get turnovers
160turnovers:List[float] = [x for x in self.turnovers]
161# reverse list for easier implementation of storing regression data
162turnovers.reverse()
163
164x:List[float] = [] # storing one data point of regression_x in loop
165regression_y:List[float] = []
166regression_x:List[float] = []
167
168for turnover in turnovers:
169    if len(x) == (regression_period - 1):
170        # add intercept to current x data point
171        x = [1] + x
172        # add last turnover for current data point in regression to regression_y
173        regression_y.append(turnover)
174        # add one data point of x to regression_x
175        regression_x.append(x)
176        # remove intercept and firstly added turnover
177        x = x[2:]
178    
179    # keep adding turnovers to x
180    x.append(turnover)
181    
182return regression_x, regression_y
183
184# custom fee model
185class CustomFeeModel(FeeModel):
186def GetOrderFee(self, parameters):
187fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
188return OrderFee(CashAmount(fee, "USD"))