Return Cross-Predictability in Firms with Similar Employee Satisfaction

1from AlgorithmImports import *
2import numpy as np
3from typing import List, Dict
4from pandas.core.frame import DataFrame
5from pandas.core.series import Series
6from dataclasses import dataclass, field
7#endregion
8class ReturnCrossPredictabilityInFirmsWithSimilarEmployeeSatisfaction(QCAlgorithm):
9def Initialize(self) -> None:
10self.SetStartDate(2010, 1, 1) # First ES data are for 2009
11self.SetCash(100000)
12
13symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
14
15self.holding_period: int = 12 # Holding portfolio for 12 months 
16self.invest_count: int = 5 # Long n stocks and short n stocks
17self.period: int = 21 # Storing n daily prices
18self.leverage: int = 5
19self.min_share_price: int = 5
20
21self.selected_stock_count: int = 2 # Calculate the value-weighted average of n stock over and under current stock
22
23self.select_current_stock: bool = True # If this flag is True, calculation of value-weighted average will include current stock too
24
25self.data: Dict[Symbol, SymbolData] = {} # Storing daily prices of stocks
26self.employee_satisfaction: Dict[str, Dict[int, float]] = {} # Storing stock's ratings for specific years
27
28self.managed_queue: List[RebalanceQueueItem] = []
29
30# Download companies ratings for each year
31csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/index/EMPLOYEE_SATISFACTION.csv')
32lines: List[str] = csv_string_file.split('\r\n')
33for line in lines[1:]: # Skip header
34    line_split: List[str] = line.split(';')
35    date: datetime.date = datetime.strptime(line_split[0], "%d.%m.%Y").date()
36    
37    company_ticker: str = line_split[1]
38    rating: str = line_split[2]
39    
40    # Create dictionary for each company ticker
41    if company_ticker not in self.employee_satisfaction:
42        self.employee_satisfaction[company_ticker] = {}
43    
44    # Under company ticker and year store rating
45    self.employee_satisfaction[company_ticker][date.year] = rating
46    
47self.selection_flag: bool = False
48self.settings.daily_precise_end_time = False
49self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
50self.UniverseSettings.Resolution = Resolution.Daily
51self.AddUniverse(self.FundamentalSelectionFunction)
52self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
53def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
54for security in changes.AddedSecurities:
55    security.SetFeeModel(CustomFeeModel())
56    security.SetLeverage(self.leverage)
57
58def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
59for stock in fundamental:
60    symbol: Symbol = stock.Symbol
61    
62    # Store stock price if it is in universe
63    if symbol.Value in self.employee_satisfaction and symbol in self.data:
64        self.data[symbol].update(stock.AdjustedPrice)
65
66# Rebalance monthly
67if not self.selection_flag:
68    return Universe.Unchanged
69
70# Universe will be created based on stock tickers in self.employee_satisfaction
71selected: List[Fundamental] = [
72    x for x in fundamental 
73    if x.HasFundamentalData 
74    and x.MarketCap != 0 
75    and x.Market == 'usa' 
76    and x.Symbol.Value in self.employee_satisfaction 
77    and x.Price > self.min_share_price
78    ]
79stock_ratings = {} # Storing objects of StockRatings
80market_cap: Dict[Symbol, float] = {} # Storing stocks market capitalization
81
82prev_year = self.Time.year - 1 # Getting stocks ratings for previous year
83
84for stock in selected:
85    symbol: Symbol = stock.Symbol
86    if symbol not in self.data:
87        # Warmup RollingWindow with History for current stock symbol
88        self.data[symbol] = SymbolData(self.period)
89        history: DataFrame = self.History(symbol, self.period, Resolution.Daily)
90        # Continue only if history isn't empty
91        if history.empty:
92            self.Log(f"Not enough data for {symbol} yet.")
93            continue
94        
95        # Get and store stock's closes
96        closes: Series = history.loc[symbol].close
97        for _, close in closes.items():
98            self.data[symbol].update(close)
99    if not self.data[symbol].is_ready():
100        continue
101    # Check if stock has rating for previous year
102    if prev_year in self.employee_satisfaction[symbol.Value]:
103        # Get stock's rating for last year
104        rating: str = self.employee_satisfaction[symbol.Value][prev_year]
105        
106        if rating not in stock_ratings:
107            stock_ratings[rating] = StockRatings()
108            
109        # Add new stock's symbol to stocks dictionary
110        stock_ratings[rating].stocks[symbol] = stock.MarketCap
111        
112        # Store stock's market capitalization
113        market_cap[symbol] = stock.MarketCap
114
115# Sort stocks by thier ES ratings and stocks with same rating sort by market capitalization
116sorted_by_rating: List[Symbol] = self.SortStocks(stock_ratings)
117
118# Storing stocks value weight average
119value_weight_avg: Dict[Symbol, float] = {}
120
121# Exclude first and last n stocks
122for i in range(self.selected_stock_count, len(sorted_by_rating) - self.selected_stock_count):
123    current_stock_symbol: Symbol = sorted_by_rating[i]
124    
125    # Select current stock, n stocks with smaller rating and n stocks with larger rating
126    selected_stocks: List[Symbol] = sorted_by_rating[i-self.selected_stock_count:i+self.selected_stock_count + 1]
127    # Calculate and store current stock's value weight average
128    value_weight_avg[current_stock_symbol] = self.ValueWeightAverage(current_stock_symbol, selected_stocks, market_cap)
129
130# Sort stocks by value weight average
131sorted_by_value_weight_avg: List[Symbol] = [x[0] for x in sorted(value_weight_avg.items(), key=lambda item: item[1])]
132
133# Buy stocks of n firms, which have the best performing peers over the last period
134long: List[Symbol] = sorted_by_value_weight_avg[-self.invest_count:]
135# Short stocks of n firms, which have the worst-performing peers
136short: List[Symbol] = sorted_by_value_weight_avg[:self.invest_count]
137
138if len(long) != 0 and len(short) != 0:
139    # Calculate portfolio weight for long and short part
140    long_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
141    short_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
142    
143    long_symbol_q = [(x, np.floor(long_w / self.data[x].last_price)) for x in long]
144    short_symbol_q = [(x, -np.floor(short_w / self.data[x].last_price)) for x in short]
145    
146    self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
147
148return long + short
149def OnData(self, data: Slice) -> None:
150if not self.selection_flag:
151    return
152self.selection_flag = False
153
154remove_item: Union[None, RebalanceQueueItem] = None
155
156# Rebalance portfolio
157for item in self.managed_queue:
158    if item.holding_period == self.holding_period: # All portfolio parts are held for n months
159        for symbol, quantity in item.opened_symbol_q:
160            self.MarketOrder(symbol, -quantity)
161                    
162        remove_item = item
163    
164    # Trade execution    
165    if item.holding_period == 0: # All portfolio parts are held for n months
166        opened_symbol_q: List[Tuple[Symbol, int]] = []
167        
168        for symbol, quantity in item.opened_symbol_q:
169            if symbol in data and data[symbol]:
170                self.MarketOrder(symbol, quantity)
171                opened_symbol_q.append((symbol, quantity))
172                    
173        # Only opened orders will be closed        
174        item.opened_symbol_q = opened_symbol_q
175        
176    item.holding_period += 1
177    
178# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
179if remove_item:
180    self.managed_queue.remove(remove_item)
181
182def Selection(self) -> None:
183self.selection_flag = True
184
185def SortStocks(self, stocks_ratings) -> List[Symbol]:
186sorting_result: List[Symbol] = []
187
188# Sort dictionary by key values
189sorted_by_rating: List[StockRatings] = [x[1] for x in sorted(stocks_ratings.items(), key=lambda item: item[0], reverse=True)]
190
191# Sort stocks with same rating by market capitalizations
192for value in sorted_by_rating:
193    # Sort dictionary of stocks with same ratings by their market capitalization
194    sorted_by_cap = [x[0] for x in sorted(value.stocks.items(), key=lambda item: item[1], reverse=True)]
195    
196    sorting_result += sorted_by_cap
197    
198return sorting_result
199
200def ValueWeightAverage(self, 
201                current_stock_symbol: Symbol, 
202                stocks: List[Symbol], 
203                market_cap: Dict[Symbol, float]) -> float:
204value_weights: List[float] = []
205# Sum total market capitalization of needed stocks
206total_cap: float = sum([market_cap[x] for x in stocks if x != current_stock_symbol or self.select_current_stock])
207
208for symbol in stocks:
209    if symbol != current_stock_symbol or self.select_current_stock:
210        # Calculate stock performance on whole period
211        performance: float = self.data[symbol].performance()
212        # Calculate current stock weight
213        weight: float = market_cap[symbol] / total_cap
214        # Calculate value weight return
215        value_weights.append(weight * performance)
216
217# Return value-weighted average    
218return np.mean(value_weights)
219
220@dataclass
221class RebalanceQueueItem():
222# symbol/quantity collections
223opened_symbol_q: Symbol 
224holding_period: float = 0
225class StockRatings():
226def __init__(self) -> None:
227self.stocks: Dict[str, int] = {}
228
229class SymbolData():
230def __init__(self, period: int) -> None:
231self.closes: RollingWindow = RollingWindow[float](period)
232self.last_price: Union[None, float] = None
233
234def update(self, close: float) -> None:
235self.closes.Add(close)
236self.last_price = close
237
238def is_ready(self) -> None:
239return self.closes.IsReady
240
241def performance(self) -> float:
242closes: List[float] = [x for x in self.closes]
243return (closes[0] - closes[-1]) / closes[-1]
244
245# Custom fee model.
246class CustomFeeModel(FeeModel):
247def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
248fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
249return OrderFee(CashAmount(fee, "USD"))