Share Issuance Effect

1from AlgorithmImports import *
2from math import isnan
3class ShareIssuanceEffect(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2000, 1, 1)
6self.SetCash(100000)
7
8self.long:List[Symbol] = []
9self.short:List[Symbol] = []
10self.shares_number:Dict[Symbol, RollingWindow] = {}
11self.leverage:int = 5
12self.min_share_price:float = 5.
13
14market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15
16self.fundamental_count:int = 500
17self.fundamental_sorting_key = lambda x: x.DollarVolume
18self.record_shares_flag = False
19self.record_shares_flag_month:int = 6
20self.selection_flag = False
21self.selection_flag_month:int = 11
22self.UniverseSettings.Resolution = Resolution.Daily
23self.AddUniverse(self.FundamentalSelectionFunction)
24self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
25self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
26self.settings.daily_precise_end_time = False
27def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
28for security in changes.AddedSecurities:
29    security.SetFeeModel(CustomFeeModel())
30    security.SetLeverage(self.leverage)
31
32def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
33if not self.selection_flag and not self.record_shares_flag:
34    return Universe.Unchanged
35selected:List[Fundamental] = [
36    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price >= self.min_share_price and \
37    not isnan(x.FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths) and x.FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths > 0
38]
39
40if len(selected) > self.fundamental_count:
41    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
42if self.record_shares_flag:
43    for stock in selected:
44        symbol:Symbol = stock.Symbol
45    
46        if symbol not in self.shares_number:
47            self.shares_number[symbol] = RollingWindow[float](2)
48        
49        shares_number:float = stock.FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths
50        self.shares_number[symbol].Add(shares_number)
51    
52    # NOTE: Get rid of old shares number records so we work with latest values.
53    del_symbols:List[Symbol] = []
54    for symbol in self.shares_number:
55        if symbol not in [x.Symbol for x in selected]:
56            del_symbols.append(symbol)
57    for symbol in del_symbols:
58        del self.shares_number[symbol]
59    
60    self.record_shares_flag = False
61    
62elif self.selection_flag:
63    net_issuance:Dict[Symbol, float] = {}
64    
65    for stock in selected:
66        symbol:Symbol = stock.Symbol
67        
68        if symbol in self.shares_number and self.shares_number[symbol].IsReady:
69            shares_values:List[float] = list(self.shares_number[symbol])
70            net_issuance[symbol] = shares_values[0] / shares_values[-1] - 1
71        
72    if len(net_issuance) != 0:
73        zero_net_issuance:List[float] = [x[0] for x in net_issuance.items() if x[1] == 0]
74        
75        pos_net_issuance:List = [x for x in net_issuance.items() if x[1] > 0]
76        sorted_pos_by_net_issuance:List = sorted(pos_net_issuance, key = lambda x: x[1], reverse = True)
77        quantile:int = int(len(sorted_pos_by_net_issuance)/5)
78        pos_high:List[Symbol] = [x[0] for x in sorted_pos_by_net_issuance[:quantile]]
79        
80        neg_net_issuance:List = [x for x in net_issuance.items() if x[1] < 0]
81        sorted_neg_by_net_issuance:List = sorted(neg_net_issuance, key = lambda x: x[1], reverse = False)
82        half:int = int(len(sorted_neg_by_net_issuance) / 2)
83        neg_high:List[Symbol] = [x[0] for x in sorted_neg_by_net_issuance[:half]]
84        
85        #self.long = zero_net_issuance
86        self.long = neg_high 
87        self.short = pos_high
88    
89return self.long + self.short
90
91def OnData(self, data: Slice) -> None:
92if not self.selection_flag:
93    return
94self.selection_flag = False
95# Trade execution and rebalance
96targets:List[PortfolioTarget] = []
97for i, portfolio in enumerate([self.long, self.short]):
98    for symbol in portfolio:
99        if symbol in data and data[symbol]:
100            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
101
102self.SetHoldings(targets, True)
103self.long.clear()
104self.short.clear()
105
106def Selection(self) -> None:
107if self.Time.month == self.record_shares_flag_month:
108    self.record_shares_flag = True
109elif self.Time.month == self.selection_flag_month:
110    self.selection_flag = True
111# Custom fee model.
112class CustomFeeModel(FeeModel):
113def GetOrderFee(self, parameters):
114fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
115return OrderFee(CashAmount(fee, "USD"))