Dividend Announcement Effect

1import trade_manager
2from AlgorithmImports import *
3from typing import Dict, List
4import data_tools
5class DividendAnnouncementEffect(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9
10self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
11self.universe_assets:List[Symbol] = []
12self.fundamental_sorting_key = lambda x: x.DollarVolume
13self.fundamental_count:int = 500
14self.leverage:int = 10
15
16self.data:Dict[Symbol, data_tools.SymbolData] = {}
17
18# self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
19spy_data = self.AddData(data_tools.QuantpediaFutures, 'CME_ES1', Resolution.Daily)
20spy_data.SetLeverage(self.leverage)
21self.symbol:Symbol = spy_data.Symbol
22
23self.period:int = 12
24self.holding_period:int = 12 * 30
25self.long_count:int = 40
26self.short_count:int = 0
27
28# 40 symbols long, one year of holding.
29self.trade_manager:TradeManager = trade_manager.TradeManager(self, self.long_count, self.short_count, self.holding_period)
30
31self.selection_flag:bool = False
32self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
33self.settings.daily_precise_end_time = False
34self.UniverseSettings.Resolution = Resolution.Daily
35self.AddUniverse(self.FundamentalSelectionFunction)
36self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.At(0, 0), self.Selection)
37
38def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
39for security in changes.AddedSecurities:
40    security.SetFeeModel(data_tools.CustomFeeModel())
41    security.SetLeverage(self.leverage)
42
43for security in changes.RemovedSecurities:
44    symbol:Symbol = security.Symbol
45    if symbol in self.data:
46        del self.data[symbol]
47    
48def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
49if not self.selection_flag:
50    return Universe.Unchanged
51
52selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchange_codes]
53
54if len(selected) > self.fundamental_count:
55    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
56self.universe_assets = [x.Symbol for x in selected]
57for symbol in self.universe_assets:
58    if symbol not in self.data:
59        self.data[symbol] = data_tools.SymbolData(self.period)
60
61return self.universe_assets
62def OnData(self, data: Slice):
63if self.Securities[self.symbol].GetLastData() and self.Time.date() >= data_tools.QuantpediaFutures.get_last_update_date()[self.symbol]:
64    self.Liquidate()
65    return
66# Liquidate opened symbols after one year.
67self.trade_manager.TryLiquidate()
68
69# Update dividends data
70for kvp in data.Dividends:
71    div_ticker:str = kvp.Key
72    if div_ticker in self.data:
73        # Storing dividends data
74        month_year:str = self.Time.strftime("%m-%Y")
75        self.data[div_ticker].update_dividends(month_year)
76
77# If only SPY is invested, we liquidate it        
78if self.Portfolio.Count == 1 and self.Portfolio[self.symbol].Invested is True:
79    self.Liquidate(self.symbol)
80
81if not self.selection_flag:
82    return
83self.selection_flag = False
84
85# Check stocks for trades
86for symbol in self.universe_assets:
87    if symbol in self.data:
88        if symbol in data and data[symbol]:
89            if self.data[symbol].is_included_long_enough():
90                current_date:datetime.date = self.Time.date()
91                
92                # Trade execution
93                if self.data[symbol].check_dividends_paying(current_date) and self.Securities[symbol].Invested is False:
94                    # We short SPY if we invest into at least one stock from universe.
95                    if not self.Portfolio[self.symbol].Invested:
96                        self.SetHoldings(self.symbol, -1)
97                    
98                    self.trade_manager.Add(symbol, True)
99                
100            self.data[symbol].increment_inclusion_period()
101        
102def Selection(self) -> None:
103self.selection_flag = True