Estimating Hedge Funds’ Returns Out of Sample

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4from pandas.core.frame import DataFrame
5import numpy as np
6import statsmodels.api as sm
7# endregion
8
9class EstimatingHedgeFundsReturnsOutofSample(QCAlgorithm):
10
11def Initialize(self):
12self.SetStartDate(2011, 1, 1)
13self.SetCash(100000)
14
15self.leverage:int = 5
16
17self.max_missing_days:int = 90
18self.selection_month:int = 1
19self.t_stats_value_threshold:int = 2
20
21self.period:int = 36
22self.month_period:int = 21 
23self.quantile:int = 5
24
25self.traded_portfolio_portion:Dict[Symbol, float] = {}
26self.data:Dict[Symbol, SymbolData] = {}
27self.fund_performance:Dict[str, RollingWindow] = {}
28
29self.holdings_by_fund:Dict[str, FundHoldings] = {}
30self.ticker_universe:set = set()  # every ticker stored in hedge fund holdings data
31
32self.funds_tickers:Dict[str, Dict[datetime.date, list]] = {}
33
34hedge_fund_file_content:str = self.Download('data.quantpedia.com/backtesting_data/equity/hedge_fund_holdings/hedge_funds_holdings.json')
35hedge_funds_data:Dict = json.loads(hedge_fund_file_content)
36
37for index, hedge_fund_data in enumerate(hedge_funds_data):
38    hedge_fund_names:list[str] = list(hedge_fund_data.keys())
39    hedge_fund_names.remove('date')
40    date:datetime.date = datetime.strptime(hedge_fund_data['date'], '%d.%m.%Y').date()
41
42    for hedge_fund_name in hedge_fund_names:
43        if hedge_fund_name not in self.holdings_by_fund:
44            self.holdings_by_fund[hedge_fund_name] = data_tools.FundHoldings(hedge_fund_name)
45
46        holding_list:list[StockHolding] = []
47        holdings:list[Dict] = hedge_fund_data[hedge_fund_name]['stocks']
48        for holding in holdings:
49            ticker:str = holding['ticker']
50            number_of_shares:int = int(holding['#_of_shares'])
51            weight:float = float(holding['weight'])
52
53            self.ticker_universe.add(ticker)
54
55            if ticker not in self.funds_tickers:
56                # initialize dictionary for stock's ticker
57                self.funds_tickers[ticker] = {}
58
59            if date not in self.funds_tickers[ticker]:
60                # initialize list, where will be all funds, which hold this stock in this date
61                self.funds_tickers[ticker][date] = []
62            
63            # add fund with stock weight in that fund to list == tuple (hedge_fund_name, weight)
64            self.funds_tickers[ticker][date].append((hedge_fund_name, weight))
65
66            holding_list.append(data_tools.StockHolding(ticker, number_of_shares, weight))
67        
68        self.holdings_by_fund[hedge_fund_name].update_holdings_by_date(date, holding_list)
69
70self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
71self.vix:Symbol = self.AddEquity("SVXY", Resolution.Daily).Symbol
72
73# universe selection
74self.selection_flag:bool = False
75self.rebalance_flag:bool = False
76self.UniverseSettings.Resolution = Resolution.Daily
77self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
78self.UniverseSettings.Resolution = Resolution.Daily
79self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
80self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
81
82def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
83for security in changes.AddedSecurities:
84    security.SetFeeModel(data_tools.CustomFeeModel())
85    security.SetLeverage(self.leverage)
86
87for security in changes.RemovedSecurities:
88    if security.Symbol in self.data:
89        self.data.pop(security.Symbol)
90
91def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
92# monthly selection
93if not self.selection_flag:
94    return Universe.Unchanged
95self.selection_flag = False
96
97# update the price every month
98for stock in coarse:
99    symbol:Symbol = stock.Symbol
100    ticker:str = symbol.Value
101
102    if ticker in self.ticker_universe:
103        if symbol in self.data:
104            self.data[symbol].update_data(stock.AdjustedPrice)
105
106if self.vix in self.data:
107    if not self.Securities[self.vix].Price == 0:
108        self.data[self.vix].update_data(self.Securities[self.vix].Price)
109
110# selected = [x.Symbol for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice >= 1 and x.Symbol.Value in self.ticker_universe]
111selected:List[Symbol] = [x.Symbol
112    for x in sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice >= 1 and x.Symbol.Value in self.ticker_universe],
113        key = lambda x: x.DollarVolume, reverse = True)]
114
115# warmup price rolling windows
116for symbol in selected + [self.vix]:
117    if symbol in self.data:
118        continue
119
120    if symbol.Value in self.ticker_universe or symbol == self.vix:
121        self.data[symbol] = data_tools.SymbolData(self.period)
122        history:DataFrame = self.History(symbol, self.period * self.month_period, Resolution.Daily)
123        if history.empty:
124            self.Log(f"Not enough data for {symbol} yet.")
125            continue
126        closes:pd.Series = history.loc[symbol].close.groupby(pd.Grouper(freq='M')).last()
127        for time, close in closes.iteritems():
128            self.data[symbol].update_data(close)
129
130return [x for x in selected if self.data[x].is_ready()]
131
132def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
133fine = [x for x in fine if x.MarketCap != 0 and x.AssetClassification.MorningstarSectorCode != MorningstarSectorCode.FinancialServices and \
134        (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')]
135
136fine:Dict[Symbol, FineFundamental] = {x.Symbol.Value: x for x in fine}
137
138if len(fine) != 0:
139    performance_by_stock:Dict[str, float] = {ticker: self.data[fine[ticker].Symbol].get_last_return() for ticker in self.ticker_universe \
140        if ticker in fine and self.data[fine[ticker].Symbol].is_ready()}
141
142    for fund, fund_data in self.holdings_by_fund.items():
143        last_date = fund_data.get_latest_date(self.Time.date())
144        if last_date is None:
145            continue
146
147        fund_performance:float = sum([(x.weight / 100) * performance_by_stock[x.ticker] for x in fund_data.holdings_by_date[last_date] if x.ticker in fine])
148
149        if fund not in self.fund_performance:
150            self.fund_performance[fund] = RollingWindow[float](self.period)
151        self.fund_performance[fund].Add(fund_performance)
152
153if not self.rebalance_flag:
154    return Universe.Unchanged
155# self.rebalance_flag = False
156
157if not self.data[self.vix].is_ready():
158    return Universe.Unchanged
159
160vix_returns:np.ndarray = np.array(self.data[self.vix].get_returns())
161selected_funds:Dict[str, FundHoldings] = {}
162
163# run regression on every fund
164for fund, fund_data in self.holdings_by_fund.items():
165    if fund not in self.fund_performance:
166        continue
167    if not self.fund_performance[fund].IsReady:
168        continue
169
170    x:np.ndarray = vix_returns
171    y:np.ndarray = np.array(list(self.fund_performance[fund])[::-1])
172
173    model = self.multiple_linear_regression(x, y)
174
175    # check alpha and t stats values
176    if model.params[0] > 0 and abs(model.tvalues[0]) >= self.t_stats_value_threshold and abs(model.tvalues[1]) >= 2:
177        selected_funds[fund] = fund_data
178
179# calculate portions on stocks
180if len(selected_funds) != 0:
181    for fund, fund_data in selected_funds.items():
182        last_date = fund_data.get_latest_date(self.Time.date())
183
184        if last_date is None or (self.Time.date() - last_date).days >= self.max_missing_days:
185            continue
186        for data in fund_data.holdings_by_date[last_date]:
187            if data.ticker not in fine:
188                continue
189            portion:float = ((self.Portfolio.TotalPortfolioValue / len(selected_funds)) * (data.weight / 100))
190            if data.ticker not in self.traded_portfolio_portion:
191                self.traded_portfolio_portion[fine[data.ticker].Symbol] = 0
192            self.traded_portfolio_portion[fine[data.ticker].Symbol] += portion
193            
194return list(self.traded_portfolio_portion.keys())
195
196def OnData(self, data: Slice):
197if not self.rebalance_flag:
198    return
199self.rebalance_flag = False
200
201# trade execution
202stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
203for symbol in stocks_invested:
204    if symbol not in self.traded_portfolio_portion:
205        self.Liquidate(symbol)
206
207for symbol, portion in self.traded_portfolio_portion.items():
208    if symbol in data and data[symbol]:
209        quantity:float = (portion // data[symbol].Price) - self.Portfolio[symbol].Quantity
210        self.MarketOrder(symbol, quantity)
211
212self.traded_portfolio_portion.clear()
213
214def Selection(self):
215if self.Time.month != self.selection_month:
216    self.selection_flag = True
217    return
218self.rebalance_flag = True
219self.selection_flag = True
220
221def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
222x = sm.add_constant(x, has_constant='add')
223result = sm.OLS(endog=y, exog=x).fit()
224return result