Combined Momentum and Nearness to 52-week High

1from AlgorithmImports import *
2import data_tools
3# endregion
4
5class CombinedMomentumAndNearnessTo52WeekHigh(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10
11self.leverage:int = 5
12self.quantile:int = 5
13self.total_portfolios:int = 2
14self.portfolio_percentage:float = 1
15
16self.min_price_period:int = 15
17self.min_volume_period:int = 15
18self.min_share_price:float = 1.
19self.high_period:int = 52 * 5
20
21self.exchanges:List[str] = ['NYS', 'NAS', 'ASE']
22
23self.data:Dict[Symbol, data_tools.SymbolData] = {}
24self.weight:Dict[Symbol, float] = {}
25self.selected_symbols:List[Symbol] = []
26
27self.fundamental_count:int = 500
28self.fundamental_sorting_key = lambda x: x.DollarVolume
29
30self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
31
32self.selection_flag:bool = False
33self.UniverseSettings.Resolution = Resolution.Daily
34self.AddUniverse(self.FundamentalSelectionFunction)
35self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
36self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), self.Selection)
37
38def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
39for security in changes.AddedSecurities:
40    security.SetFeeModel(data_tools.CustomFeeModel())
41    security.SetLeverage(self.leverage)
42
43def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
44if not self.selection_flag:
45    return Universe.Unchanged
46
47selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice >= self.min_share_price and x.MarketCap != 0 and \
48    not np.isnan(x.EarningReports.BasicAverageShares.ThreeMonths) and x.EarningReports.BasicAverageShares.ThreeMonths != 0 and x.SecurityReference.ExchangeId in self.exchanges]
49
50if len(selected) > self.fundamental_count:
51    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
52
53self.selected_symbols.clear()
54
55for stock in selected:
56    symbol:Symbol = stock.Symbol
57    self.selected_symbols.append(symbol)
58
59    if symbol not in self.data:
60        self.data[symbol] = data_tools.SymbolData(self.high_period)
61
62    self.data[symbol].update_shares_outstanding(stock.EarningReports.BasicAverageShares.ThreeMonths)
63    self.data[symbol].update_market_cap(stock.MarketCap)
64
65symbols_to_delete:List[Symbol] = list(filter(lambda symbol: symbol not in self.selected_symbols, self.data))
66for symbol in symbols_to_delete:
67    del self.data[symbol]
68
69return self.selected_symbols
70
71def OnData(self, data: Slice) -> None:
72for symbol in self.selected_symbols:
73    if symbol in data and data[symbol] and data[symbol].High != 0 and \
74     data[symbol].Price != 0 and data[symbol].Volume != 0:
75        high:float = data[symbol].High
76        volume:float = data[symbol].Volume
77        price:float = data[symbol].Price
78
79        self.data[symbol].update(price, volume, high)
80
81if not self.selection_flag:
82    return
83self.selection_flag = False
84
85performance:Dict[Symbol, float] = {}
86turnover:Dict[Symbol, float] = {}
87PTH:Dict[Symbol, float] = {}
88
89for symbol in self.selected_symbols:
90    symbol_obj:data_tools.SymbolData = self.data[symbol]
91
92    if symbol_obj.PTH_data_ready():
93        symbol_obj.update_PTH_values()
94    else:
95        symbol_obj.reset_PTH_values()
96
97    if symbol_obj.is_ready(self.min_price_period, self.min_volume_period):
98        performance[symbol] = symbol_obj.get_performance()
99        turnover[symbol] = symbol_obj.get_turnover()
100        PTH[symbol] = symbol_obj.get_prev_PTH_value()
101
102    symbol_obj.reset_prices()
103    symbol_obj.reset_volumes()
104
105if len(performance) < self.quantile:
106    self.Liquidate()
107    return
108
109quantile:int = int(len(performance) / self.quantile)
110sorted_by_perf:List[Symbol] = [x[0] for x in sorted(performance.items(), key=lambda item: item[1])]
111sorted_by_turnover:List[Symbol] = [x[0] for x in sorted(turnover.items(), key=lambda item: item[1])]
112sorted_by_PTH:List[Symbol] = [x[0] for x in sorted(PTH.items(), key=lambda item: item[1])]
113
114perf_winners:List[Symbol] = sorted_by_perf[-quantile:]
115perf_losers:List[Symbol] = sorted_by_perf[:quantile]
116
117lowest_turnover:List[Symbol] = sorted_by_turnover[:quantile]
118highest_turnover:List[Symbol] = sorted_by_turnover[-quantile:]
119
120lowest_PTH:List[Symbol] = sorted_by_PTH[:quantile]
121highest_PTH:List[Symbol] = sorted_by_PTH[-quantile:]
122
123# For the first portfolio, go long Hight-PTH Winners and go short Low-PTH Winners in the lowest turnover quintile.
124long_leg = [x for x in lowest_turnover if x in highest_PTH and x in perf_winners and x in data and data[x]]
125short_leg = [x for x in lowest_turnover if x in lowest_PTH and x in perf_winners and x in data and data[x]]
126if len(long_leg) != 0 and len(short_leg) != 0:
127    self.CalculateWeights(long_leg, short_leg)
128
129# For the second portfolio, go long High PTH-Winners and go short Low-PTH Winners in the highest turnover quintile.
130long_leg = [x for x in highest_turnover if x in highest_PTH and x in perf_winners and x in data and data[x]]
131short_leg = [x for x in highest_turnover if x in lowest_PTH and x in perf_winners and x in data and data[x]]
132
133if len(long_leg) != 0 and len(short_leg) != 0:
134    self.CalculateWeights(long_leg, short_leg)
135
136# trade execution
137portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
138self.SetHoldings(portfolio, True)
139
140self.weight.clear()
141
142def CalculateWeights(self, long_leg:List, short_leg:List) -> None:
143for i, portfolio in enumerate([long_leg, short_leg]):
144    mc_sum:float = sum(list(map(lambda symbol: self.data[symbol].get_market_cap(), portfolio)))
145    for symbol in portfolio:
146        self.weight[symbol] = ((self.data[symbol].get_market_cap() / mc_sum) / self.total_portfolios) * self.portfolio_percentage
147
148def Selection(self) -> None:
149self.selection_flag = True