When Short Sellers and Corporate Insiders Agree on Stock Pricing

1from AlgorithmImports import *
2from collections import deque
3import pandas as pd
4import numpy as np
5from io import StringIO
6#endregion
7class WhenShortSellersandCorporateInsidersAgreeonStockPricing(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2013, 1, 1)
10self.SetCash(100000)
11self.symbols = [
12                'AAPL','MSFT','AMZN','FB','BRKB','GOOGL','GOOG','JPM','JNJ','V',
13                'PG','XOM','UNH','BAC','MA','T','DIS','INTC','HD','VZ',
14                'MRK','PFE','CVX','KO','CMCSA','CSCO','PEP','WFC','C','BA',
15                'ADBE','WMT','CRM','MCD','MDT','BMY','ABT','NVDA','NFLX','AMGN',
16                'PM','PYPL','TMO','COST','ABBV','ACN','HON','NKE','UNP','UTX',
17                # 'NEE','IBM','TXN','AVGO','LLY','ORCL','LIN','SBUX','AMT','LMT',
18                # 'GE','MMM','DHR','QCOM','CVS','MO','LOW','FIS','AXP','BKNG',
19                # 'UPS','GILD','CHTR','CAT','MDLZ','GS','USB', 'CI','ANTM','BDX',
20                # 'TJX','ADP','TFC','CME','SPGI','COP','INTU','ISRG','CB','SO',
21                # 'D','FISV','PNC','DUK','SYK','ZTS','MS','RTN','AGN','BLK'
22                ]
23                
24self.period = 3 * 21
25self.quantile = 4
26self.max_SI_missing_days = 5
27
28# Create custom universe.
29self.UniverseSettings.Resolution = Resolution.Daily
30self.AddUniverseSelection(FineFundamentalUniverseSelectionModel(self.SelectCoarse, self.SelectFine))
31# Dataframe with insider trades for every stock.
32self.insiders_trading = {}
33
34# Daily short interest data.
35self.short_interest = {}
36
37# Short interest and investor demand quarterly pairs.
38self.data = {}
39
40self.long = []
41self.short = []
42
43self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
44
45for symbol in self.symbols:
46    # Import insiders trading data.
47    csv_string_file = self.Download(f'data.quantpedia.com/backtesting_data/economic/insiders_trading/{symbol}.csv')
48    if csv_string_file == "": continue
49    parser = lambda x: pd.datetime.strptime(x, "%Y-%m-%d")
50    self.insiders_trading[symbol] = pd.read_csv(StringIO(csv_string_file), sep=';', parse_dates=['Tran.Date'], date_parser=parser)
51    # Import short interest daily data.
52    self.AddData(NasdaqCustomColumns, 'FINRA/FNSQ_' + symbol, Resolution.Daily)
53    self.short_interest[symbol] = deque(maxlen = self.period)
54self.selection_flag = False
55self.rebalance_flag = False
56
57self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection)
58def OnSecuritiesChanged(self, changes):
59for security in changes.AddedSecurities:
60    security.SetFeeModel(CustomFeeModel())
61    security.SetLeverage(10)
62def SelectCoarse(self, coarse):
63if not self.selection_flag:
64    return Universe.Unchanged
65
66return [Symbol.Create(x, SecurityType.Equity, Market.USA) for x in self.symbols]
67def SelectFine(self, fine):
68fine = [x for x in fine if x.EarningReports.BasicAverageShares.ThreeMonths > 0 and x.Symbol.Value in self.insiders_trading]
69# Short interests and demand diffs.
70data_change = {}    
71for stock in fine:
72    symbol = stock.Symbol
73    ticker = symbol.Value
74    if self.Securities['FINRA/FNSQ_' + ticker].GetLastData() and (self.Time.date() - self.Securities['FINRA/FNSQ_' + ticker].GetLastData().Time.date()).days > self.max_SI_missing_days:
75        self.data[symbol].clear()
76        continue
77    # Last month's short_interest data is ready.
78    if len(self.short_interest[ticker]) == self.short_interest[ticker].maxlen:
79        # Calculate investor demand.
80        buys = [row['Shares'] for index, row in self.insiders_trading[ticker].iterrows() if row['Symbol'] == ticker and row['Tran.Date'] >= (self.Time - timedelta(days = self.period)) and row['Tran.Date'] <= self.Time and row['Action'] == 'B'] 
81        sells = [row['Shares'] for index, row in self.insiders_trading[ticker].iterrows() if row['Symbol'] == ticker and row['Tran.Date'] >= (self.Time - timedelta(days = self.period)) and row['Tran.Date'] <= self.Time and row['Action'] == 'S']
82                
83        total_buy_shares = sum(buys)
84        total_sell_shares = sum(sells)
85        
86        if len(buys) != 0 or len(sells) != 0:
87            demand = (total_buy_shares - total_sell_shares) / stock.EarningReports.BasicAverageShares.ThreeMonths
88            
89            # Calculate quarterly short interest.
90            short_interest = sum([x[0] for x in self.short_interest[ticker]]) / sum([x[1] for x in self.short_interest[ticker]])
91            
92            # Store quarterly data pairs.                    
93            if symbol not in self.data:
94                self.data[symbol] = deque()
95            self.data[symbol].append([short_interest, demand])
96        
97            # If there is at least of 4 quarters of data ready.
98            if len(self.data[symbol]) >= 4:
99                short_interest_diff = np.diff([x[0] for x in self.data[symbol]])
100                demand_diff = np.diff([x[1] for x in self.data[symbol]])
101                
102                data_change[symbol] = [np.median(short_interest_diff), np.median(demand_diff)]
103    
104if len(data_change) >= self.quantile:
105    # Sorting by short interest and demand diffs.
106    sorted_by_interest_change = sorted(data_change.items(), key = lambda x: x[1][0], reverse = True)
107    quantile = int(len(sorted_by_interest_change) / self.quantile)
108    high_by_interest_change = [x[0] for x in sorted_by_interest_change[:quantile]]
109    low_by_interest_change = [x[0] for x in sorted_by_interest_change[-quantile:]]
110    
111    sorted_by_demand_change = sorted(data_change.items(), key = lambda x: x[1][1], reverse = True)
112    quantile  = int(len(sorted_by_demand_change) / self.quantile)
113    high_by_demand_change = [x[0] for x in sorted_by_demand_change[:quantile]]
114    low_by_demand_change = [x[0] for x in sorted_by_demand_change[-quantile:]]
115    
116    self.long = [x for x in high_by_interest_change if x in low_by_demand_change]
117    self.short = [x for x in low_by_interest_change if x in high_by_demand_change]
118
119return self.long + self.short
120def OnData(self, data):
121# Store short interest data.
122for symbol in self.symbols:
123    look_up_symbol = 'FINRA/FNSQ_' + symbol
124    if look_up_symbol in data and data[look_up_symbol]:
125        short_vol = data[look_up_symbol].GetProperty("SHORTVOLUME")
126        total_vol = data[look_up_symbol].GetProperty("TOTALVOLUME")
127        
128        if symbol in self.short_interest:
129            self.short_interest[symbol].append((short_vol, total_vol))
130
131# rebalance once a month     
132if not self.rebalance_flag:
133    return
134self.selection_flag = False
135self.rebalance_flag = False
136
137# Trade execution
138stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
139for symbol in stocks_invested:
140    if symbol not in self.long + self.short:
141        self.Liquidate(symbol)
142long_count = len(self.long)
143short_count = len(self.short)
144
145for symbol in self.long:
146    if symbol in data and data[symbol]:
147        self.SetHoldings(symbol, 1 / long_count)
148for symbol in self.short:
149    if symbol in data and data[symbol]:
150        self.SetHoldings(symbol, -1 / short_count)
151       
152def Selection(self):
153if self.Time.month in [3,6,9,12]:
154    # clear long and short selection once every selection period, so that portoflio can be rabalanced monthly even without new selection
155    self.long.clear()
156    self.short.clear()
157    self.selection_flag = True
158
159self.rebalance_flag = True
160    
161# Custom fee model
162class CustomFeeModel(FeeModel):
163def GetOrderFee(self, parameters):
164fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
165return OrderFee(CashAmount(fee, "USD"))
166# Quandl short interest data.
167class NasdaqCustomColumns(NasdaqDataLink):
168def __init__(self) -> None:
169self.ValueColumnName = 'shortvolume'    # also 'TOTALVOLUME' is accesible