ETF的创建/赎回活动与收益可预测性

1from AlgorithmImports import *
2from dateutil.relativedelta import relativedelta
3#endregion
4class ETFCreationRedemptionActivityAndReturnPredictability(QCAlgorithm):
5def Initialize(self) -> None:
6self.SetStartDate(2006, 1, 1) # etf data starts since 2006
7self.SetCash(100_000)
8
9self.spread_period: int = 21 # need n bid ask spread values for each ETF
10self.leverage: int = 5
11self.quantile: int = 5
12
13self.data: Dict[Symbol, SymbolData] = {}
14self.shares_outstanding: Dict[str, float] = {} 
15
16self.long: List[Symbol] = []
17self.short: List[Symbol] = []
18
19self.last_custom_data_date: datetime.date = datetime(1, 1, 1).date()
20market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
21
22csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/equity/etf_shares_outstanding.csv')
23lines: List[str] = csv_string_file.split('\r\n')
24
25# retrieve tickers from csv header
26self.etf_tickers: List[str] = lines[0].split(';')[1:]
27
28for etf_ticker in self.etf_tickers:
29    self.shares_outstanding[etf_ticker] = {}
30
31for line in lines[1:]: # iterate through each line except header
32    if line == '':
33        continue
34    
35    split: List[str] = line.split(';')
36    date: datetime.date = datetime.strptime(split[0], "%Y-%m-%d").date()
37    shares_outstanding_values: float = split[1:]
38    total_shares_outstanding_values: int = len(shares_outstanding_values)
39    
40    if date > self.last_custom_data_date:
41        self.last_custom_data_date = date
42    # load share outstanding data and index them by year, month then day
43    for index in range(1, total_shares_outstanding_values, 1):
44        # get share outstanding value for specific ticker
45        shares_outstanding_value: float = float(shares_outstanding_values[index])
46        
47        # make sure stored value won't be zero
48        if shares_outstanding_value == 0:
49            continue
50        
51        # get etf ticker, which belongs to current share outstanding value
52        etf_ticker: str = self.etf_tickers[index]
53        
54        if date not in self.shares_outstanding[etf_ticker]:
55            self.shares_outstanding[etf_ticker][date] = shares_outstanding_value
56
57self.selection_flag: bool = False
58self.UniverseSettings.Leverage = self.leverage
59self.UniverseSettings.Resolution = Resolution.Daily
60self.AddUniverse(self.FundamentalSelectionFunction)
61self.settings.daily_precise_end_time = False
62self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
63self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.BeforeMarketClose(market, 0), self.Selection)
64def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
65for security in changes.AddedSecurities:
66    security.SetFeeModel(CustomFeeModel())
67def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
68# rebalance monthly    
69if not self.selection_flag:
70    return Universe.Unchanged
71
72# filter ETFs universe
73selected: List[Fundamental] = [x for x in fundamental if x.Symbol.Value in self.etf_tickers]
74
75share_spread: Dict[Symbol, float] = {}
76bid_ask_avg_spread: Dict[Symbol, float] = {}
77
78for etf in selected:
79    etf_symbol: Symbol = etf.Symbol
80    
81    # initialize SymbolData object for new etf symbol in data
82    if etf_symbol not in self.data:
83        self.data[etf_symbol] = SymbolData(self.spread_period)
84    
85    # make sure, data are ready    
86    if not self.data[etf_symbol].are_data_ready():
87        continue
88    
89    share_spread_value: float = self.GetShareSpread(etf_symbol.Value)
90    
91    # make sure there is valid share spread value
92    if share_spread_value == None:
93        continue
94    
95    bid_ask_avg_spread_value: float = self.data[etf_symbol].bid_ask_avg_spread()
96    
97    # store spreads
98    share_spread[etf_symbol] = share_spread_value
99    bid_ask_avg_spread[etf_symbol] = bid_ask_avg_spread_value
100    
101# there has to be enough ETFs for quintile selections   
102if len(share_spread) < self.quantile or len(bid_ask_avg_spread) < self.quantile:
103    return list(self.data.keys())
104    
105# quintile selections
106quintile: int = int(len(share_spread) / self.quantile)
107sorted_by_share_spread: List[Symbol] = [x[0] for x in sorted(share_spread.items(), key=lambda item: item[1])]
108sorted_by_bid_ask_avg_spread: List[Symbol] = [x[0] for x in sorted(bid_ask_avg_spread.items(), key=lambda item: item[1])]
109
110highest_share_spread: List[Symbol] = sorted_by_share_spread[-quintile:]
111lowest_share_spread: List[Symbol] = sorted_by_share_spread[:quintile]
112
113# highest_bid_ask_avg_spread = sorted_by_bid_ask_avg_spread[-quintile:]
114lowest_bid_ask_avg_spread: List[Symbol] = sorted_by_bid_ask_avg_spread[:quintile]
115
116# Investor goes long portfolio with the lowest ShareChange (quintile with the highest redemption activity) and simultaneously with the lowest bid-ask spread.
117self.long = [x for x in lowest_share_spread if x in lowest_bid_ask_avg_spread]
118# Investor goes short portfolio with the highest ShareChange (quintile with the highest creation activity) and simultaneously with the lowest bid-ask spread.
119self.short = [x for x in highest_share_spread if x in lowest_bid_ask_avg_spread]
120
121return list(self.data.keys())
122def OnData(self, slice: Slice) -> None:
123if self.time.date() > self.last_custom_data_date:
124    self.liquidate()
125    return
126    
127for etf_symbol in self.data:
128    if slice.contains_key(etf_symbol) and slice[etf_symbol]:
129        bid: float = self.Securities[etf_symbol].BidPrice
130        ask: float = self.Securities[etf_symbol].AskPrice
131        
132        self.data[etf_symbol].update_bid_ask_spread(bid, ask)
133
134# rebalance monthly
135if not self.selection_flag:
136    return
137self.selection_flag = False
138
139# trade execution
140targets: List[PortfolioTarget] = []
141for i, portfolio in enumerate([self.long, self.short]):
142    for symbol in portfolio:
143        if slice.contains_key(symbol) and slice[symbol]:
144            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
145
146self.SetHoldings(targets, True)
147self.long.clear()
148self.short.clear()
149
150def GetShareSpread(self, etf_ticker: str) -> float:
151current_date: datetime.date = self.Time.date()
152prev_month_date: datetime.date = current_date - relativedelta(months=1)
153
154# get indexed ETF's share oustanding from dictionary
155etf_shares_outstanding: float = self.shares_outstanding[etf_ticker]
156
157# make sure share outstanding values are ready for each day
158if current_date not in etf_shares_outstanding or prev_month_date not in etf_shares_outstanding:
159    return None
160    
161# get current share oustanding value
162curr_shares_outstanding: float = etf_shares_outstanding[current_date]
163
164# get prev month share outstanding value
165prev_shares_outstanding: float = etf_shares_outstanding[prev_month_date]
166
167# return share spread
168return (curr_shares_outstanding / prev_shares_outstanding) - 1
169
170def Selection(self) -> None:
171self.selection_flag = True
172
173class SymbolData():
174def __init__(self, spread_period: int) -> None:
175self.bid_ask_spread: RollingWindow = RollingWindow[float](spread_period)
176
177def update_bid_ask_spread(self, bid: float, ask: float) -> None:
178midpoint: float = (bid + ask) / 2
179spread: float = (ask - bid) / midpoint
180self.bid_ask_spread.Add(spread)
181
182def are_data_ready(self) -> bool:
183return self.bid_ask_spread.IsReady
184
185def bid_ask_avg_spread(self) -> float:
186bid_ask_spreads: List[float] = [x for x in self.bid_ask_spread]
187bid_ask_avg_spread: float = sum(bid_ask_spreads) / len(bid_ask_spreads)
188return bid_ask_avg_spread
189
190# Custom fee model
191class CustomFeeModel(FeeModel):
192def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
193fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
194return OrderFee(CashAmount(fee, "USD"))