Analyst Days

1from AlgorithmImports import *
2#endregion
3class AnalystDays(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2010, 1, 1)
6self.SetCash(100_000)
7
8self.equally_weighted_flag: bool = False     # False - VW; True - EW
9self.analyst_days: Dict[str, List[datetime.date]] = {}
10self.selected: List[Fundamental] = []
11self.tickers: List[str] = []
12self.market_cap:dict[Symbol, float] = {}
13   
14market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15self.spy_future_symbol: Symbol = market
16
17self.opened_long_positions_period: Dict[Symbol, int] = {}      # opened long stock postions with holding period
18self.holding_period: int = 20   # n days of holding period
19self.leverage: int = 10
20csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/analyst_days.csv')
21lines: List[str] = csv_string_file.split('\r\n')
22
23header: List[str] = lines[0].split(';')
24
25for ticker in header[1:]: # skip first column == date
26    self.tickers.append(ticker)
27for line in lines[1:]: # skip header line
28    if line == '':
29        continue
30    
31    line: List[str] = line.split(';')
32    
33    # convert string to date
34    str_date: str = line[0]
35    date: datetime.date = datetime.strptime(str_date, '%d.%m.%Y').date()
36    for index in range(1, len(line)): # skip date as a first value of the row
37        # retrive ticker from list created based on csv header
38        ticker: str = self.tickers[index - 1]
39        
40        if ticker not in self.analyst_days:
41            self.analyst_days[ticker] = []
42        
43        # if analyst_day_flag == 1, then this company had analyst days
44        analyst_day_flag: str = line[index]
45        
46        if analyst_day_flag == '1':
47            # company had analyst days
48            self.analyst_days[ticker].append(date)
49self.selection_flag: bool = False
50self.UniverseSettings.Leverage = self.leverage
51self.UniverseSettings.Resolution = Resolution.Minute
52self.AddUniverse(self.FundamentalSelectionFunction)
53self.settings.daily_precise_end_time = False
54self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
55self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market, 0), self.Selection)
56self.Schedule.On(self.DateRules.EveryDay(market), self.TimeRules.BeforeMarketClose(market, 1), self.ManageTrade)
57def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
58for security in changes.AddedSecurities:
59    security.SetFeeModel(CustomFeeModel())
60def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
61# rebalance monthly
62if not self.selection_flag:
63    return Universe.Unchanged
64self.selection_flag = False
65
66# select S&P500 stocks from fundamental based on ticker    
67self.selected = [
68    x for x in fundamental 
69    if x.MarketCap != 0 
70    and x.Symbol.Value in self.analyst_days
71]
72if self.equally_weighted_flag:
73    self.selected_values = list(map(lambda stock: stock.Symbol, self.selected))
74    return self.selected_values
75else:
76    self.market_cap = { stock.Symbol : stock.MarketCap for stock in self.selected }
77    return list(self.market_cap.keys())
78
79def ManageTrade(self):
80# liquidate opened symbols
81symbols_to_remove: List[Symbol] = []
82rebalance_flag: bool = False
83
84for symbol in self.opened_long_positions_period:
85    holding_period_remaining: int = self.opened_long_positions_period[symbol]
86    if holding_period_remaining == 0:
87        # remove stock from holdings
88        symbols_to_remove.append(symbol)
89    else:
90        # decrement remaining holding period
91        self.opened_long_positions_period[symbol] -= 1
92
93for symbol in symbols_to_remove:
94    self.Liquidate(symbol)
95    del self.opened_long_positions_period[symbol]
96    rebalance_flag = True   # rebalance is required
97today: datetime.date = self.Time.date()
98long: List[Symbol] = [] # storing symbols of stocks of those companies that had analyst day
99
100# check if selected companies had analyst day
101for stock in self.selected:
102    symbol: Symbol = stock.Symbol
103    ticker: str = symbol.Value
104    # make sure selected stock has analyst days data
105    if ticker not in self.analyst_days:
106        continue
107    
108    if today in self.analyst_days[ticker]:
109        # store stock in opened postions selection
110        self.opened_long_positions_period[symbol] = self.holding_period
111        rebalance_flag = True   # rebalance is required
112
113# rebalance is required
114if rebalance_flag:
115    # rebalance whole active selection
116    if self.equally_weighted_flag:
117        n: int = len(self.opened_long_positions_period)
118        if n != 0:
119            # rebalance whole trade selection
120            for symbol in self.opened_long_positions_period:
121                price = self.Securities[symbol].Price
122                if price != 0:
123                    self.SetHoldings(symbol, 1/n)
124    else:
125        total_cap: float = sum([self.market_cap[x] for x in self.opened_long_positions_period])
126        for symbol in self.opened_long_positions_period:
127            if symbol in self.market_cap:
128                self.SetHoldings(symbol, self.market_cap[symbol] / total_cap)
129
130def Selection(self) -> None:
131self.selection_flag = True
132
133# Custom fee model.
134class CustomFeeModel(FeeModel):
135def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
136fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
137return OrderFee(CashAmount(fee, "USD"))