Large Price Changes combined with Analyst Revisions

1from AlgorithmImports import *
2from pandas.tseries.offsets import BDay
3from typing import List, Dict
4import data_tools
5# endregion
6class LargePriceChangesCombinedWithAnalystRevisions(QCAlgorithm):
7def Initialize(self) -> None:
8self.SetStartDate(2010, 1, 1)       # estimize dataset starts in 2011
9self.SetCash(100_000)
10
11self.years_period: int = 3
12self.low_high_percentage: int = 30
13self.min_values: int = 15
14self.period: int = 45    # need n values for mean volumes calculation
15self.volume_percentage: float = 1.1
16self.return_increase: float = 0.05
17self.days_for_revision: int = 5
18self.leverage: int = 5
19self.min_share_price: int = 5
20self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
21self.data: Dict[Symbol, SymbolData] = {}
22self.weights: Dict[Symbol, float] = {}
23self.already_subscribed: Dict[Symbol] = []
24self.estimates: Dict[str, Dict[datetime.date, List[str]]] = {}
25self.analysts_data: Dict[str, Dict[datetime.date, float]] = {}
26market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
27self.fundamental_sorting_key = lambda x: x.DollarVolume
28self.fundamental_count: int = 500
29self.rebalance_flag: bool = False
30self.selection_flag: bool = False
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
35def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
36for security in changes.AddedSecurities:
37    security.SetFeeModel(data_tools.CustomFeeModel())
38    security.SetLeverage(self.leverage)
39def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
40curr_date: datetime.date = self.Time.date()
41# daily update of prices and volumes
42for equity in fundamental:
43    symbol: Symbol = equity.Symbol
44    if symbol in self.data:
45        self.data[symbol].update(curr_date, equity.AdjustedPrice, equity.Volume)
46# monthly selection
47if not self.selection_flag:
48    return Universe.Unchanged
49self.selection_flag = False
50self.rebalance_flag = True
51
52selected: List[Fundamental] = [
53    x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and x.Market == 'usa' and x.Price > self.min_share_price and \
54    x.SecurityReference.ExchangeId in self.exchange_codes
55]
56
57if len(selected) > self.fundamental_count:
58    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
59selected_stocks: set = set()
60for stock in selected:
61    symbol: Symbol = stock.Symbol
62    ticker: str = symbol.Value
63    # check if stock is already subscribed
64    if symbol not in self.data:
65        self.data[symbol] = data_tools.SymbolData(self.period)
66        self.AddData(EstimizeEstimate, symbol)
67    if not self.data[symbol].is_ready(self.min_values):
68        continue
69    if ticker not in self.estimates:
70        continue
71    
72    large_swing_dates: List[datetime.date] = self.data[symbol].get_large_swing_dates(self.volume_percentage, self.return_increase)
73    # iterate through each large swing date and check if any analyst increased estimated EPS within self.days_for_revision days
74    for date in large_swing_dates:
75        for i in range(1, self.days_for_revision + 1, 1):
76            future_date: datetime.date = (date + BDay(i)).date()
77            if future_date not in self.estimates[ticker]:
78                continue
79            analyst_ids: List[str] = self.estimates[ticker][future_date]
80            for analyst_id in analyst_ids:
81                estimate_dates: List[datetime.date] = list(self.analysts_data[analyst_id].keys())
82                estimate_dates.reverse()
83                est_after_swing: float = self.analysts_data[analyst_id][future_date]
84                latest_date_before_swing: datetime.date = next((est_date for est_date in estimate_dates if est_date < date), None)
85                # check if analyst increased his/her EPS estimate value
86                if latest_date_before_swing != None and (est_after_swing > self.analysts_data[analyst_id][latest_date_before_swing]):
87                    selected_stocks.add(symbol)
88                    break
89            
90            # stock were already selected, no need to check any more dates
91            if selected_stocks in selected_stocks:
92                break
93# reset monthly data
94for symbol, symbol_obj in self.data.items():
95    symbol_obj.reset_monthly_data()
96long_length: int = len(selected_stocks)
97for symbol in selected_stocks:
98    self.weights[symbol] = 1 / long_length
99return list(selected_stocks)
100
101def OnData(self, data: Slice) -> None:
102estimate = data.Get(EstimizeEstimate)
103for symbol, value in estimate.items():
104    ticker: str = symbol.Value
105    if ticker not in self.estimates:
106        self.estimates[ticker] = {}
107    created_at: datetime.date = value.CreatedAt.date()
108    if created_at not in self.estimates[ticker]:
109        self.estimates[ticker][created_at] = []
110    analyst_id: str = value.AnalystId
111    self.estimates[ticker][created_at].append(analyst_id)
112    if analyst_id not in self.analysts_data:
113        self.analysts_data[analyst_id] = {}
114    self.analysts_data[analyst_id][created_at] = value.Eps
115# rebalance when selection was made
116if not self.rebalance_flag:
117    return
118self.rebalance_flag = False
119# reset monthly data
120for _, symbol_obj in self.data.items():
121    symbol_obj.reset_monthly_data()
122# trade execution
123portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weights.items() if symbol in data and data[symbol]]
124self.SetHoldings(portfolio, True)
125
126self.weights.clear()
127def Selection(self) -> None:
128self.selection_flag = True