Employee Sentiment and Stock Returns

1from AlgorithmImports import *
2from typing import List
3import statsmodels.api as sm
4import numpy as np
5# endregion
6
7class EmployeeSentimentandStockReturns(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2007, 5, 1) # BIL inception date
11self.SetCash(100000)
12
13self.leverage:int = 5
14
15self.risk_aversion:float = 3.
16self.allocation_limits:List[float] = [-0.5, 1.5]
17
18self.tickers:List[str] = [
19    'A', 'AA', 'AAPL', 'AXP', 'BA', 'BAC', 'CAT', 'CSCO', 'CVX', 'DD', 
20    'DIS', 'GE', 'HD', 'HPQ', 'IBM', 'INTC', 'JNJ', 'KFT', 'KO', 'MRK',
21    'MSFT', 'PFE', 'PG', 'TRV', 'UTX', 'VZ', 'XOM'] # MCD, WMT
22
23self.negative_scores:List[float] = [1., 2.]
24self.positive_scores:List[float] = [4., 5.]
25
26self.reviews_by_ticker:Dict[str, List[Tuple[datetime.date, float]]] = {}
27
28self.last_review_date:datetime.date = datetime(1,1,1).date()
29
30# import employee review data
31for ticker in self.tickers:
32    csv_string_file:str = self.Download(f"data.quantpedia.com/backtesting_data/economic/indeed_review/{ticker}.csv")
33    lines:List[str] = csv_string_file.split('\r\n')
34    for line in lines[1:]: # skip header
35        if line == '': continue
36
37        line_split:List[str] = line.split(';')
38        date:datetime.date = datetime.strptime(line_split[0], "%d.%m.%Y").date()
39        score:float = float(line_split[1])
40
41        # store date of the review with associated score
42        if ticker not in self.reviews_by_ticker:
43            self.reviews_by_ticker[ticker] = []
44        
45        self.reviews_by_ticker[ticker].append((date, score))
46        
47        # mark up last available review date
48        if date > self.last_review_date:
49            self.last_review_date = date
50
51# subscribe price data
52data:Equity = self.AddEquity("SPY", Resolution.Daily)
53data.SetLeverage(self.leverage)
54self.market:Symbol = data.Symbol
55
56data:Equity = self.AddEquity("BIL", Resolution.Daily)
57data.SetLeverage(self.leverage)
58self.t_bills:Symbol = data.Symbol
59
60# sentiment data
61self.m_period:int = 12
62self.sentiment_data:SentimentData = SentimentData(self.m_period)
63
64self.SetWarmUp(self.m_period * 30, Resolution.Daily)
65                                                    
66self.recent_month:int = -1
67
68def OnData(self, data: Slice) -> None:
69# market data are present in the algorithm
70if self.market not in data or not data[self.market]:
71    return
72
73# monthly rebalance
74if self.Time.month == self.recent_month:
75    return
76self.recent_month = self.Time.month
77
78last_month:datetime.datetime = self.Time.replace(day=1) - timedelta(days=1)
79
80employee_sentiment:float = 0.
81for ticker, reviews_list in self.reviews_by_ticker.items():
82    n_positive_reviews:float = len([x for x in reviews_list if x[1] in self.positive_scores and x[0].month == last_month.month and x[0].year == last_month.year])
83    n_negative_reviews:float = len([x for x in reviews_list if x[1] in self.negative_scores and x[0].month == last_month.month and x[0].year == last_month.year])
84    firm_employee_sentiment:float = (n_positive_reviews - n_negative_reviews) / (n_positive_reviews + n_negative_reviews) if (n_positive_reviews + n_negative_reviews) != 0. else 0
85    employee_sentiment += firm_employee_sentiment
86
87# update sentiment and market data
88employee_sentiment /= len(self.reviews_by_ticker)
89self.sentiment_data.update_data(data[self.market].Value, employee_sentiment)
90
91if self.IsWarmingUp: return
92
93# no more review data available
94if self.Time.date() > self.last_review_date:
95    self.Liquidate()
96    return
97
98if self.sentiment_data.is_ready():
99    x:Tuple[np.ndarray, np.ndarray] = self.sentiment_data.get_regression_data()
100    model = self.multiple_linear_regression(x[1][:-1], x[0][1:])
101    forecast_return:float = model.predict([1, x[1][-1]])[0]
102    forecast_variance:float = np.std(x[0]) ** 2 # * np.sqrt(12)
103    w_t:float = (1. / self.risk_aversion) * (forecast_return / forecast_variance)
104    w_t = min(max(w_t, self.allocation_limits[0]), self.allocation_limits[1])
105    t_bill_w:float = 1. - w_t
106
107    self.SetHoldings(self.market, w_t)
108    self.SetHoldings(self.t_bills, t_bill_w)
109
110def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
111x:np.ndarray = np.array(x).T
112x = sm.add_constant(x)
113result = sm.OLS(endog=y, exog=x).fit()
114return result
115
116class SentimentData():
117def __init__(self, period:int) -> None:
118self._period:int = period
119self._market_prices:RollingWindow = RollingWindow[float](period + 1)
120self._sentiment_index:RollingWindow = RollingWindow[float](period)
121
122def is_ready(self) -> bool:
123return self._market_prices.IsReady and self._sentiment_index.IsReady
124
125def update_data(self, price:float, sentiment_index_value:float) -> None:
126self._market_prices.Add(price)
127self._sentiment_index.Add(sentiment_index_value)
128
129def get_regression_data(self) -> Tuple[np.ndarray, np.ndarray]:
130prices:np.ndarray = np.array(list(self._market_prices))
131returns:np.ndarray = prices[:-1] / prices[1:] - 1
132
133sentiment_index:np.ndarray = np.array(list(self._sentiment_index))
134return returns[::-1], sentiment_index[::-1]