ETF Flows Predict Subsequent ETF Performance

1from AlgorithmImports import *
2from typing import List, Dict
3from data_tools import CustomFeeModel, QuantpediaSharesOutstandingETFs, SymbolData
4# endregion
5
6class ETFFlowsPredictSubsequentETFPerformance(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2010, 1, 1)
10self.SetCash(100000)
11
12self.leverage:int = 5
13
14self.market_close_minute:int = 0
15self.market_close_hour:int = 16
16
17self.market_open_minute:int = 31
18self.market_open_hour:int = 9
19
20self.max_missing_days:int = 5
21
22self.open_trades:List[List[Symbol, float]] = []
23self.data:Dict[Symbol, SymbolData] = {}
24
25csv_file:str = self.Download('data.quantpedia.com/backtesting_data/equity/etf_shares_outstanding.csv')
26lines:List[str] = csv_file.split('\r\n')
27
28tickers:List[str] = lines[0].split(';')[1:]
29
30for ticker in tickers:
31    data = self.AddEquity(ticker, Resolution.Minute)
32    data.SetFeeModel(CustomFeeModel())
33    data.SetLeverage(self.leverage)
34
35    self.data[ticker] = SymbolData(data.Symbol)
36
37self.etf_shares_outstanding:Symbol = self.AddData(QuantpediaSharesOutstandingETFs,
38    'etf_shares_outstanding', Resolution.Daily).Symbol
39
40def OnData(self, data: Slice):
41if self.etf_shares_outstanding in data and data[self.etf_shares_outstanding]:
42    shares_outstanding:Dict[str, float] = data[self.etf_shares_outstanding].GetProperty('shares_outstanding')
43
44    curr_date:datetime.date = self.Time.date()
45
46    for ticker, shares_outstanding in shares_outstanding.items():
47        self.data[ticker].update_shares_outstanding(curr_date, shares_outstanding)
48
49if self.Time.minute == self.market_close_minute and self.Time.hour == self.market_close_hour:
50    # on market close calculate flow values and create MarketOnOper orders
51    long_leg:List[str] = []
52    short_leg:List[str] = []
53    curr_date:datetime.date = self.Time.date()
54    
55    for ticker, symbol_data in self.data.items():
56        symbol:Symbol = symbol_data.get_symbol()
57
58        if not symbol_data.data_still_coming(curr_date, self.max_missing_days):
59            # make sure shares outstanding data are still coming
60            symbol_data.reset()
61
62        elif symbol_data.shares_outstanding_ready() and symbol in data and data[symbol]:
63            # trade only ETFs, which have price and shares outstanding
64            price:float = data[symbol].Value
65            symbol_data.update_price(price)
66            symbol_data.update_market_cap()
67
68            flow_value:float = symbol_data.get_flow()
69            
70            if flow_value > 0:
71                long_leg.append(ticker)
72            elif flow_value < 0:
73                short_leg.append(ticker)
74
75    weight:float = self.Portfolio.TotalPortfolioValue / 2 \
76        if len(long_leg) != 0 and len(short_leg) != 0 else self.Portfolio.TotalPortfolioValue
77
78    # trade execution
79    total_long_cap:float = sum(map(lambda ticker: self.data[ticker].get_market_cap(), long_leg))
80    for ticker in long_leg:
81        symbol:Symbol = self.data[ticker].get_symbol()
82        quantity:float = -self.get_quantity(ticker, weight, total_long_cap)
83
84        self.MarketOnOpenOrder(symbol, quantity)
85        self.open_trades.append((symbol, -quantity))
86
87    total_short_cap:float = sum(map(lambda ticker: self.data[ticker].get_market_cap(), short_leg))
88    for ticker in short_leg:
89        symbol:Symobl = self.data[ticker].get_symbol()
90        quantity:float = self.get_quantity(ticker, weight, total_short_cap)
91
92        self.MarketOnOpenOrder(symbol, quantity)
93        self.open_trades.append((symbol, -quantity))
94
95if self.Time.minute == self.market_open_minute and self.Time.hour == self.market_open_hour:
96    # create MarketOnClose order to liquidate opened trades
97    for symbol, quantity in self.open_trades:
98        if self.Portfolio[symbol].Invested:
99            self.MarketOnCloseOrder(symbol, quantity)
100
101    self.open_trades.clear()
102
103def get_quantity(self, ticker:str, weight:float, total_cap:float) -> float:
104price:float = self.data[ticker].get_price()
105market_cap:float = self.data[ticker].get_market_cap()
106
107quantity:float = np.floor((weight * (market_cap / total_cap)) / price)
108
109return quantity