Trading on the Dividend Paydate

1from AlgorithmImports import *
2from datetime import datetime
3from pandas.tseries.offsets import BDay
4from typing import Dict, List
5import json
6#endregion
7class TradingDividendPaydate(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2012, 9, 18)
10self.SetCash(100000)    
11symbol:Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
12self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
13# Store drip tickers.
14# Source: http://www.dripdatabase.com/DRIP_Directory_AtoZ.aspx
15csv_string_file:str = self.Download('data.quantpedia.com/backtesting_data/economic/drip_tickers.csv')
16lines:str = csv_string_file.split('\r\n')
17self.drip_tickers:List[str] = [x for x in lines[1:]]
18
19# dividend data
20self.dividend_data:Dict = {}  # dict of dicts indexed by paydate date
21
22# Data source: https://www.nasdaq.com/market-activity/dividends
23dividend_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/dividend_dates.json')
24dividend_data_json:Dict[str] = json.loads(dividend_data)
25    
26for obj in dividend_data_json:
27    ex_div_date:datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
28    
29    for stock_data in obj['stocks']:
30        ticker:str = stock_data['ticker']
31        payday:datetime.date = datetime.strptime(stock_data['PayDate'], '%m/%d/%Y').date()
32        if payday not in self.dividend_data:
33            self.dividend_data[payday] = {}    
34        record_date:Union[datetime.date, None] = datetime.strptime(stock_data['RecordDate'], '%m/%d/%Y').date() if 'RecordDate' in stock_data else None
35        dividend_value:float = stock_data['Div']
36        ann_dividend_value:float = stock_data['AnnDiv']
37        announcement_date:Union[datetime.date, None] = datetime.strptime(stock_data['AnnounceDate'], '%m/%d/%Y').date() if 'AnnounceDate' in stock_data else None
38        # store ticker dividend info to current ex-div date
39        self.dividend_data[payday][ticker] = DividendInfo(ticker, ex_div_date, payday, record_date, dividend_value, ann_dividend_value, announcement_date)
40self.active_universe:List[Symbol] = []   # selected stock universe
41self.selection_flag:bool = False
42self.UniverseSettings.Resolution = Resolution.Minute
43self.AddUniverse(self.FundamentalSelectionFunction)
44self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
45self.settings.daily_precise_end_time = False
46self.Schedule.On(self.DateRules.MonthEnd(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
47self.Schedule.On(self.DateRules.EveryDay(symbol), self.TimeRules.BeforeMarketClose(symbol, 16), self.Rebalance)
48def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
49for security in changes.AddedSecurities:
50    security.SetFeeModel(CustomFeeModel())
51def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
52if not self.selection_flag:
53    return Universe.Unchanged
54
55self.selection_flag = False
56selection:List[Fundamental] = [x for x in fundamental if x.Symbol.Value in self.drip_tickers and x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchange_codes]
57# sorting by market cap
58sorted_by_market_cap = sorted(selection, key = lambda x: x.MarketCap, reverse = True)
59half = int(len(sorted_by_market_cap) / 2)
60
61# pick lower half
62self.active_universe = [x.Symbol for x in sorted_by_market_cap[-half:]]
63
64# pick upper half
65# self.active_universe = [x.Symbol for x in sorted_by_market_cap[:half]]
66
67return self.active_universe
68
69def Rebalance(self) -> None:
70# close opened positions
71stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
72for symbol in stocks_invested:
73    q_invested:int = self.Portfolio[symbol].Quantity
74    self.MarketOnCloseOrder(symbol, -q_invested)
75day_to_check = (self.Time.date() + BDay(1)).date()
76# there are stocks with payday next business day
77if day_to_check in self.dividend_data:
78    payday_tickers = list(self.dividend_data[day_to_check].keys())
79    long = []
80    for symbol in self.active_universe:
81        if symbol.Value in payday_tickers:
82            long.append(symbol) 
83    
84    if len(long) != 0:
85        portfolio_value = self.Portfolio.MarginRemaining / len(long)
86        for symbol in long:
87            price = self.Securities[symbol].Price
88            if price != 0:
89                q = portfolio_value / price
90                self.MarketOnCloseOrder(symbol, q)
91def Selection(self) -> None:
92if self.Time.month % 3 == 0:
93    self.selection_flag = True
94# custom fee model
95class CustomFeeModel(FeeModel):
96def GetOrderFee(self, parameters):
97fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
98return OrderFee(CashAmount(fee, "USD"))
99class DividendInfo():
100def __init__(
101    self, 
102    ticker:str, 
103    ex_div_date:datetime.date,
104    payday:datetime.date, 
105    record_date:Union[datetime.date, None],
106    dividend_value:float,
107    ann_dividend_value:float,
108    announcement_date:datetime.date
109):
110self.ticker:str = ticker
111self.ex_div_date:datetime.date = ex_div_date
112self.payday:datetime.date = payday
113self.record_date:Union[datetime.date, None] = record_date
114self.dividend_value:float = dividend_value
115self.ann_dividend_value:float = ann_dividend_value
116self.announcement_date:Union[datetime.date, None] = announcement_date