Return Range Predicts Stock Returns

1from AlgorithmImports import *
2from functools import reduce
3#endregion
4
5class ReturnRangePredictsStockReturns(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2010, 1, 1)
9self.SetCash(100000)
10
11self.period:int = 22        # need n daily prices
12self.cap_quantile:int = 5
13self.range_quantile:int = 5
14self.leverage:int = 5
15self.min_share_price:float = 5.
16
17self.prices:Dict[Symbol, RollingWindow] = {}
18self.weight:Dict[Symbol, float] = {}
19
20self.countries_ISO:List[str] = [
21    'AUS', 'AUT', 'BEL', 'CAN', 'DNK', 'FIN', 'FRA', 'DEU', 'GRC', 'HKG', 'IRL', 'ITA',
22    'JPN', 'NLD', 'NZL', 'NOR', 'PRT', 'SGP', 'ESP', 'SWE', 'CHE', 'GBR', 'USA', 'ARG',
23    'BRA', 'CHL', 'CHN', 'IND', 'KOR', 'MYS', 'MEX', 'PHL', 'POL', 'ZAF', 'TWN', 'THA', 'TUR'
24]
25
26self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
27
28self.fundamental_count:int = 3000
29self.fundamental_sorting_key = lambda x: x.MarketCap
30
31self.selection_flag:bool = False
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Settings.MinimumOrderMarginPortfolioPercentage = 0
35self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), self.Selection)
36
37def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
38for security in changes.AddedSecurities:
39    security.SetFeeModel(CustomFeeModel())
40    security.SetLeverage(self.leverage)
41    
42def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> None:
43# daily update stock prices
44for stock in fundamental:
45    symbol:Symbol = stock.Symbol
46
47    if symbol in self.prices:
48        self.prices[symbol].update(stock.AdjustedPrice)
49
50if not self.selection_flag:
51    return Universe.Unchanged
52
53selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and \
54    x.AdjustedPrice >= self.min_share_price and x.CompanyReference.BusinessCountryID in self.countries_ISO]
55if len(selected) > self.fundamental_count:
56    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
57    
58stocks_bucket:Dict[str, List[Fundamental]] = {}
59
60for stock in selected:
61    country_ISO:str = stock.CompanyReference.BusinessCountryID 
62    sector:str = str(stock.AssetClassification.MorningstarSectorCode)
63    bucket_identificator:str = country_ISO + '+' + sector
64
65    if bucket_identificator not in stocks_bucket:
66        stocks_bucket[bucket_identificator] = []
67
68    stocks_bucket[bucket_identificator].append(stock)
69
70# make sure there are enough buckets
71if len(stocks_bucket) < self.cap_quantile: return Universe.Unchanged
72
73quantile:int = int(len(stocks_bucket) / self.cap_quantile)
74stocks_bucket:List[List[Symbol]] = list(stocks_bucket.values())
75
76sorted_by_bucket_cap:List[List[Symbol]] = sorted(stocks_bucket,
77    key=lambda stocks: np.average([stock.MarketCap for stock in stocks]))
78
79small_cap:List[Symbol] = reduce(lambda x,y: x + y, sorted_by_bucket_cap[:quantile])
80large_cap:List[Symbol] = reduce(lambda x,y: x + y, sorted_by_bucket_cap[-quantile:])
81
82# SelectLowAndHighRangeStocks returns two empty list, if there aren't enough stocks for range selection
83small_cap_low_range, small_cap_high_range = self.SelectLowAndHighRangeStocks(small_cap)
84large_cap_low_range, large_cap_high_range = self.SelectLowAndHighRangeStocks(large_cap)
85
86# in the large-cap portfolio do the converse – long low ‘range’ stocks, short high ‘range’ stocks
87# in the small-cap portfolio, long the stocks with high ‘range’ value, short the stocks with low ‘range’ value
88long_part:List[Symbol] = small_cap_high_range + large_cap_low_range
89short_part:List[Symbol] = small_cap_low_range + large_cap_high_range
90
91# calc weights for each portfolio part
92for i, portfolio in enumerate([long_part, short_part]):
93    for symbol in portfolio:
94        self.weight[symbol] = ((-1) ** i) / len(portfolio)
95
96return list(self.weight.keys())
97
98def OnData(self, data: Slice) -> None:
99# rebalance monthly
100if not self.selection_flag:
101    return
102self.selection_flag = False
103
104# trade execution
105portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
106self.SetHoldings(portfolio, True)
107
108self.weight.clear()
109
110def SelectLowAndHighRangeStocks(self, stocks:List) -> (List, List):
111stocks_ranges:Dict[Symbol, float] = {}
112
113for stock in stocks:
114    symbol:Symbol = stock.Symbol
115
116    if symbol not in self.prices:
117        history = self.History(symbol, self.period, Resolution.Daily)
118        if history.empty or len(history) < self.period:
119            continue
120        
121        # init stock's RollingWindow
122        self.prices[symbol] = SymbolData(self.period)
123
124        closes:List[float] = list(history.loc[symbol, 'close'])
125
126        for close in closes:
127            self.prices[symbol].update(close)
128
129    range_value:float = self.prices[symbol].get_range()
130
131    if range_value != 0:
132        stocks_ranges[symbol] = range_value
133
134# return empty lists in case of not enough stocks with range value
135if len(stocks_ranges) < self.range_quantile:
136    return [], []
137
138quantile:int = int(len(stocks_ranges) / self.range_quantile)
139sorted_by_range:List[Symbol] = [x[0] for x in sorted(stocks_ranges.items(), key=lambda item: item[1])]
140
141low_range:List[Symbol] = sorted_by_range[:quantile]
142high_range:List[Symbol] = sorted_by_range[-quantile:]
143
144return low_range, high_range
145
146def Selection(self) -> None:
147self.selection_flag = True
148
149class SymbolData:
150def __init__(self, period: int):
151self.prices:RollingWindow = RollingWindow[float](period)
152
153def update(self, price: float) -> None:
154self.prices.Add(price)
155
156def get_range(self) -> float:
157daily_prices:np.array = np.array([x for x in self.prices])
158daily_returns:List[float] = list((daily_prices[:-1] - daily_prices[1:]) / daily_prices[1:])
159
160daily_max_return:float = max(daily_returns)
161daily_min_return:float = min(daily_returns)
162
163return daily_max_return - daily_min_return
164
165def is_ready(self) -> bool:
166return self.prices.IsReady
167
168# Custom fee model
169class CustomFeeModel(FeeModel):
170def GetOrderFee(self, parameters):
171fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
172return OrderFee(CashAmount(fee, "USD"))