Sequenced Insider Trading

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from typing import List, Dict
4import pandas as pd
5from dateutil.relativedelta import relativedelta
6#endregion
7class SequencedInsiderTrading(QCAlgorithm):
8def Initialize(self) -> None:
9self.SetStartDate(2014, 1, 1)
10self.SetCash(100000)
11self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
12market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
13self.insider_data: Dict[Symbol, QuiverInsiderTrading] = {}
14self.routine_traders_stocks: List[str] = []
15self.leverage: int = 3
16self.min_consecutive_month_count: int = 2
17self.fundamental_count: int = 500
18self.fundamental_sorting_key = lambda x: x.DollarVolume
19self.selection_flag: bool = False
20self.settings.daily_precise_end_time = False
21self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
22self.UniverseSettings.Resolution = Resolution.Daily
23self.AddUniverse(self.FundamentalSelectionFunction)
24self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
25def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
26for security in changes.AddedSecurities:
27    security.SetFeeModel(CustomFeeModel())
28    security.SetLeverage(self.leverage)
29    symbol: Symbol = security.Symbol
30    dataset_symbol: Symbol = self.AddData(QuiverInsiderTrading, symbol).Symbol
31def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
32# monthly selection
33if not self.selection_flag:
34    return Universe.Unchanged
35selected: List[Fundamental] = [
36    x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and \
37    x.SecurityReference.ExchangeId in self.exchange_codes
38]
39
40if len(selected) > self.fundamental_count:
41    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
42
43selected: Dict[str, Symbol] = {x.Symbol.Value: x.Symbol for x in selected}
44aggregated_data: Dict[Tuple[datetime.date, str], float] = {}
45curr_month: int = self.Time
46insiders: Dict[str, List[Tuple[datetime.date, str, float]]] = {}
47# aggregate data on monthly format
48for name, data in self.insider_data.items():
49    if name not in insiders:
50        insiders[name] = []
51    for data_point in data:
52        date: str = data_point[0].strftime('%m-%Y')
53        key = (date, data_point[1])
54        if key in aggregated_data:
55            if data_point[2] is not None:
56                aggregated_data[key] += data_point[2]
57        else:
58            aggregated_data[key] = data_point[2]
59    insiders[name].append([(key[0], key[1], shares) for key, shares in aggregated_data.items()])
60    aggregated_data.clear()
61# check for routine insider trades
62months_to_check: List[str] = [(curr_month - relativedelta(months=i)).strftime('%m-%Y') for i in [1, 2, 13, 14, 25, 26, 37, 38]]
63for name, data in insiders.items():
64    ticker: str = data[0][0][1]
65    if len(data[0]) > self.min_consecutive_month_count:
66        if (data[0][-1][0] == months_to_check[0] and data[0][-2][0] == months_to_check[1] and \
67            all(i[0] != month for month in months_to_check[2:] for i in data[0])) and \
68            ticker in selected:
69            self.routine_traders_stocks.append(selected[ticker])
70
71return [x for x in selected.values()]
72def OnData(self, data: Slice) -> None:
73for insider_trades in data.Get(QuiverInsiderTrading).values():
74    for insider_trade in insider_trades:
75        if insider_trade.Name not in self.insider_data:
76            self.insider_data[insider_trade.Name] = []           
77        self.insider_data[insider_trade.Name].append((insider_trade.Time, insider_trade.Symbol.Value, insider_trade.Shares))
78
79# monthly rebalance
80if not self.selection_flag:
81    return
82self.selection_flag = False
83
84targets: List[PortfolioTarget] = []
85for symbol in self.routine_traders_stocks:
86    if symbol in data and data[symbol]:
87        targets.append(PortfolioTarget(symbol, 1 / len(self.routine_traders_stocks)))
88
89self.SetHoldings(targets, True)
90self.routine_traders_stocks.clear()
91def Selection(self) -> None:
92self.selection_flag = True
93# Custom fee model.
94class CustomFeeModel(FeeModel):
95def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
96fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
97return OrderFee(CashAmount(fee, "USD"))