Lottery Stocks and the 52-Week High

1import numpy as np
2from AlgorithmImports import *
3from typing import List, Dict
4class LotteryStocks52WeekHigh(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2010, 1, 1)
7self.SetCash(100000)
8self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']    
9self.period: int = 52 * 5
10self.month_period: int = 21
11self.leverage: int = 10
12self.min_share_price: int = 5
13self.quantile: int = 10
14
15self.data: Dict[Symbol, RollingWindow] = {}
16self.long: List[Symbol] = []
17self.short: List[Symbol] = []
18
19market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
20self.fundamental_count: int = 1_000
21self.fundamental_sorting_key = lambda x: x.DollarVolume
22self.selection_flag: bool = True
23self.settings.daily_precise_end_time = False
24self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
25self.UniverseSettings.Resolution = Resolution.Daily
26self.AddUniverse(self.FundamentalSelectionFunction)
27self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
28
29def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
30for security in changes.AddedSecurities:
31    security.SetFeeModel(CustomFeeModel())
32    security.SetLeverage(self.leverage)
33
34def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
35# update the rolling window every day
36for stock in fundamental:
37    symbol: Symbol = stock.Symbol
38    if symbol in self.data:
39        self.data[symbol].Add(stock.AdjustedPrice)
40if not self.selection_flag:
41    return Universe.Unchanged
42selected: List[Fundamental] = [
43    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' \
44    and x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchange_codes
45]
46
47if len(selected) > self.fundamental_count:
48    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
49MAX: Dict[Symbol, float] = {}
50NH: Dict[Symbol, float] = {}
51for stock in selected:
52    symbol: Symbol = stock.Symbol
53    # warmup price rolling windows
54    if symbol not in self.data:
55        self.data[symbol] = RollingWindow[float](self.period)
56        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
57        if history.empty:
58            self.Log(f"Not enough data for {symbol} yet")
59            continue
60        closes:Series = history.loc[symbol].close
61        for time, close in closes.items():
62            self.data[symbol].Add(close)
63    if not self.data[symbol].IsReady:
64        continue
65    
66    closes: List[float] = list(self.data[symbol])
67    last_month_closes: np.ndarray = np.array(closes[:self.month_period])
68    last_close: float = closes[0]
69    
70    daily_returns: np.ndarray = (last_month_closes[:-1] - last_month_closes[1:]) / last_month_closes[1:]
71    MAX[symbol] = max(daily_returns)
72    
73    # NH calc    
74    if last_close != 0:
75        local_highest_close: float = max(closes)
76        NH[symbol] = last_close / local_highest_close
77if len(MAX) < self.quantile or len(NH) < self.quantile:
78    return Universe.Unchanged
79# NH sorting
80sorted_by_NH: List[Tuple[Symbol, float]] = sorted(NH.items(), key = lambda x: x[1], reverse = True)
81quintile:int = int(len(sorted_by_NH) / self.quantile)
82low_NH: List[Symbol] = [x[0] for x in sorted_by_NH[-quintile:]]
83
84# MAX sorting
85sorted_by_MAX: List[Tuple[Symbol, float]] = sorted(low_NH, key = lambda x: MAX[x], reverse = True)
86quintile: int = int(len(sorted_by_MAX) / self.quantile)
87
88self.long = [x for x in sorted_by_MAX[:quintile]]
89self.short = [x for x in sorted_by_MAX[-quintile:]]
90return self.long + self.short
91
92def OnData(self, slice: Slice) -> None:
93if not self.selection_flag:
94    return
95self.selection_flag = False
96
97# Trade execution
98targets:List[PortfolioTarget] = []
99for i, portfolio in enumerate([self.long, self.short]):
100    for symbol in portfolio:
101        if symbol in slice and slice[symbol]:
102            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
103
104self.SetHoldings(targets, True)
105self.long.clear()
106self.short.clear()
107
108def Selection(self) -> None:
109self.selection_flag = True
110
111class CustomFeeModel(FeeModel):
112def GetOrderFee(self, parameters):
113fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
114return OrderFee(CashAmount(fee, "USD"))