Post Earnings Announcement Drift in China

1from AlgorithmImports import *
2from typing import Dict, List
3from data_tools import ChineseStocks, ChineseIncomeStatement, QuantpediaCSI500, SymbolData, \
4CustomFeeModel, WaitingStock, ActiveStock
5# endregion
6
7class PostEarningsAnnouncementDriftInChina(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2015, 1, 1)   # Chinese data starts in 2015
11self.SetCash(100000)
12
13self.short_flag:bool = True
14
15self.leverage:int = 5
16
17self.threshold:float = 0.03
18
19self.wait_period:int = 3
20self.holding_period:int = 60
21
22self.active_universe:Dict[Symbol, ActiveStock] = {}     # symbols of stocks, which are actively traded
23self.waiting_universe:Dict[Symbol, WaitingStock] = {}   # symbols of stocks, which overnight perf was greater than 3%
24self.potential_trades:Dict[Symbol, datetime.date] = {}  # symbols of stocks with value of their annoucement date(income statement date)
25
26self.data:dict[Symbol, SymbolData] = {}
27
28self.top_size_symbol_count:int = 300
29ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/chinese_stocks/large_cap_500.csv')
30tickers:List[str] = ticker_file_str.split('\r\n')[:self.top_size_symbol_count]
31
32for t in tickers:
33    data = self.AddData(ChineseStocks, t, Resolution.Daily)
34    data.SetFeeModel(CustomFeeModel())
35    data.SetLeverage(self.leverage)
36
37    stock_symbol:Symbol = data.Symbol
38    data = self.AddData(ChineseIncomeStatement, t, Resolution.Daily)
39
40    self.data[stock_symbol] = SymbolData(data.Symbol)
41
42if self.short_flag:
43    data = self.AddData(QuantpediaCSI500, 'CSI_500', Resolution.Daily)
44    data.SetFeeModel(CustomFeeModel())
45    data.SetLeverage(self.leverage)
46
47    self.benchmark = data.Symbol
48
49def OnData(self, data: Slice):
50curr_date:datetime.date = self.Time.date()
51
52# store daily data
53for symbol, symbol_data in self.data.items():
54    income_statement_symbol:Symbol = symbol_data.get_income_statement_symbol()
55
56    if data.ContainsKey(symbol):
57        price_data:dict[str, str] = data[symbol].GetProperty('price_data')
58        # valid price data
59        if data[symbol].Value != 0. and price_data:
60            close_price:float = data[symbol].Value
61            open_price:float = price_data['openPrice']
62            symbol_data.update_prices(curr_date, close_price, open_price)
63
64            mc:float = float(price_data['marketValue'])
65            symbol_data.update_market_cap(mc)
66
67    # check if stock has income statement day(announcement day)
68    if data.ContainsKey(income_statement_symbol):
69        self.potential_trades[symbol] = curr_date
70
71potential_trades_to_remove:List[Symbol] = []
72for symbol, announcement_day in self.potential_trades.items():
73    # make sure overnight perf is calculated day after annoucement day
74    if curr_date == announcement_day:
75        continue
76    
77    symbol_data:SymbolData = self.data[symbol]
78    # make sure prices for overnight perf calculations are ready
79    if symbol_data.prices_ready():
80        prev_close_price:float = symbol_data.get_prev_close_price()
81        open_price:float = symbol_data.get_open_price()
82        perf:float = (open_price - prev_close_price) / prev_close_price # overnight perf
83        
84        if perf >= self.threshold:
85            # add stock's symbol to universe of stocks, which will be traded after self.wait_period days since curr_date
86            self.waiting_universe[symbol] = WaitingStock(curr_date, symbol, self.wait_period)
87
88    # stock's symbols has to be removed from universe of potential trades each time
89    potential_trades_to_remove.append(symbol)
90
91for symbol in potential_trades_to_remove:
92    del self.potential_trades[symbol]
93
94trade_flag:bool = False
95waiting_stocks_to_remove:List[Symbol] = []
96
97for symbol, waiting_stock_obj in self.waiting_universe.items():
98    # stocks in waiting universe have to wait self.waiting_period before they will be trade
99    # this condition makes sure these days will be fulfilled
100    if curr_date != waiting_stock_obj.get_date():
101        waiting_stock_obj.decrease_waiting_period(curr_date)
102
103    if waiting_stock_obj.waiting_period_expired():
104        # when self.waiting_period days pass, stock's symbol will be traded
105        # and added to universe of actively trading stocks
106        trade_flag = True
107        waiting_stocks_to_remove.append(symbol)
108        self.active_universe[symbol] = ActiveStock(curr_date, symbol, self.holding_period)
109
110for symbol in waiting_stocks_to_remove:
111    del self.waiting_universe[symbol]
112
113new_active_universe:Dict[Symbol, ActiveStock] = {}
114for symbol, active_stock_obj in self.active_universe.items():
115    # stocks in active universe are hold for self.holding_period days
116    # this condition makes sure these holding days will be fulfilled
117    if curr_date != active_stock_obj.get_date():
118        active_stock_obj.decrease_holding_period(curr_date)
119
120    if active_stock_obj.holding_period_expired():
121        # liquidate stock, when it was holded for self.holding_period days
122        self.Liquidate(symbol)
123        # change trade_flag to True, because liquidated stock was removed from active universe
124        trade_flag = True
125    else:
126        new_active_universe[symbol] = active_stock_obj
127
128self.active_universe = new_active_universe
129# trade only, when there has been change in active universe
130# when trade_flag is true, the signal for trade is met
131if trade_flag:
132    total_cap:float = sum(list(map(lambda symbol: self.data[symbol].get_market_cap(), self.active_universe)))
133    for symbol in self.active_universe:
134        self.SetHoldings(symbol, self.data[symbol].get_market_cap() / total_cap)
135
136    if self.short_flag:
137        if not self.Securities[self.benchmark].Invested and len(self.active_universe) != 0:
138            self.SetHoldings(self.benchmark, -1)
139        elif self.Securities[self.benchmark].Invested:
140            self.Liquidate(self.benchmark)