Buy-Side Competition and Momentum Profits

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict, Set
4from itertools import combinations
5from pandas.core.frame import DataFrame
6from pandas.core.series import Series
7import numpy as np
8# endregion
9class BuySideCompetitionandMomentumProfits(QCAlgorithm):
10def Initialize(self):
11self.SetStartDate(2010, 1, 1)
12self.SetCash(100000)
13self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
14self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15self.competition_quantile: int = 3
16self.momentum_quantile: int = 5
17self.leverage: int = 5
18self.month_period: int = 21
19self.period: int = 12
20self.competition_period: int = 9
21self.target_granularity: float = 0.08858
22self.min_share_price: int = 1
23self.fundamental_count: int = 1000
24self.fundamental_sorting_key = lambda x: x.DollarVolume
25self.data: Dict[Symbol, SymbolData] = {}
26self.weight: Dict[Symbol, float] = {} 
27self.stock_competition: Dict[str, RollingWindow] = {}
28self.holdings_by_fund: Dict[str, FundHoldings] = {}
29self.ticker_universe: Set = set()  # every ticker stored in hedge fund holdings data
30self.funds_tickers: Dict[str, dict[datetime.date, list]] = {}
31hedge_fund_file_content: str = self.Download('data.quantpedia.com/backtesting_data/equity/hedge_fund_holdings/hedge_funds_holdings.json')
32hedge_funds_data: List[Dict[str, Dict[str, str]]] = json.loads(hedge_fund_file_content)
33for hedge_fund_data in hedge_funds_data:
34    hedge_fund_names: List[str] = list(hedge_fund_data.keys())
35    hedge_fund_names.remove('date')
36    date: datetime.date = datetime.strptime(hedge_fund_data['date'], '%d.%m.%Y').date()
37    for hedge_fund_name in hedge_fund_names:
38        if hedge_fund_name not in self.holdings_by_fund:
39            self.holdings_by_fund[hedge_fund_name] = data_tools.FundHoldings(hedge_fund_name)
40        holding_list: List[StockHolding] = []
41        holdings: List[dict] = hedge_fund_data[hedge_fund_name]['stocks']
42        for holding in holdings:
43            ticker: str = holding['ticker']
44            number_of_shares: int = int(holding['#_of_shares'])
45            weight: float = float(holding['weight'])
46            self.ticker_universe.add(ticker)
47            if ticker not in self.funds_tickers:
48                # initialize dictionary for stock's ticker
49                self.funds_tickers[ticker] = {}
50            if date not in self.funds_tickers[ticker]:
51                # initialize list, where will be all funds, which hold this stock in this date
52                self.funds_tickers[ticker][date] = []
53            
54            # add fund with stock weight in that fund to list == tuple (hedge_fund_name, weight)
55            self.funds_tickers[ticker][date].append((hedge_fund_name, weight))
56            holding_list.append(data_tools.StockHolding(ticker, number_of_shares, weight))
57        
58        self.holdings_by_fund[hedge_fund_name].holdings_by_date[date] = holding_list
59self.last_date: datetime.date = max([max(x.holdings_by_date) for x in self.holdings_by_fund.values()])
60self.selection_flag: bool = False
61self.settings.daily_precise_end_time = False
62self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
63self.UniverseSettings.Resolution = Resolution.Daily
64self.AddUniverse(self.FundamentalSelectionFunction)
65self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
66def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
67for security in changes.AddedSecurities:
68    security.SetFeeModel(data_tools.CustomFeeModel())
69    security.SetLeverage(self.leverage)
70
71for security in changes.RemovedSecurities:
72    if security.Symbol in self.data:
73        self.data.pop(security.Symbol)
74def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
75# monthly selection
76if not self.selection_flag:
77    return Universe.Unchanged
78# update the rolling window every month
79for stock in fundamental:
80    symbol: Symbol = stock.Symbol
81    if symbol in self.data:
82        self.data[symbol].update_data(stock.AdjustedPrice)
83selected: List[Fundamental] = [
84    x for x in fundamental 
85    if x.HasFundamentalData 
86    and x.Market == 'usa' 
87    and x.MarketCap != 0 
88    and x.Price >= self.min_share_price 
89    and x.Symbol.Value in self.ticker_universe 
90    and x.AssetClassification.MorningstarSectorCode != MorningstarSectorCode.FinancialServices 
91    and x.SecurityReference.ExchangeId in self.exchange_codes
92]
93if len(selected) > self.fundamental_count:
94    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
95
96# warmup price rolling windows
97for stock in selected:
98    symbol: Symbol = stock.Symbol
99    if symbol in self.data:
100        continue
101    
102    self.data[symbol] = data_tools.SymbolData(self.period)
103    history: DataFrame = self.History(symbol, self.period * self.month_period, Resolution.Daily)
104    if history.empty:
105        self.Log(f"Not enough data for {symbol} yet.")
106        continue
107    closes: Series = history.loc[symbol].close.groupby(pd.Grouper(freq='M')).last()
108    for time, close in closes.items():
109        self.data[symbol].update_data(close)
110selected_dict: Dict[Symbol, Fundamental] = {x.Symbol.Value: x for x in selected if self.data[x.Symbol].is_ready()}
111stock_z_score: Dict[str, float] = {}
112fund_z_score: Dict[str, float] = {}
113if len(selected_dict) != 0:
114    cross_mean: float = np.mean([sym_data.get_momentum() for sym, sym_data in self.data.items() if sym.Value in selected_dict and sym_data.is_ready()])
115    cross_std: float = np.std([sym_data.get_momentum() for sym, sym_data in self.data.items() if sym.Value in selected_dict and sym_data.is_ready()])
116    
117    # calculate z-score of stocks
118    stock_z_score: Dict[str, float] = {ticker: (self.data[selected_dict[ticker].Symbol].get_momentum() - cross_mean) / cross_std for ticker in self.ticker_universe \
119        if ticker in selected_dict and self.data[selected_dict[ticker].Symbol].is_ready()}
120    # calculate z-score of funds
121    for fund, fund_data in self.holdings_by_fund.items():
122        last_date: List[datetime.date] = [x for x in list(fund_data.holdings_by_date.keys()) if x < self.Time.date()]
123        if len(last_date) == 0:
124            continue
125        
126        last_date: datetime.date = max(last_date)
127        fund_z_score[fund] = sum([(x.weight / 100) * stock_z_score[x.ticker] for x in fund_data.holdings_by_date[last_date] if x.ticker in selected_dict])
128
129    if len(fund_z_score) == 0:
130        return Universe.Unchanged
131    normalization_constant: float = max(list(fund_z_score.values()))
132    # stock level competitions
133    COMP: Dict[Symbol, float] = {}
134    # calculate fund-level competition and stock-level competition
135    for ticker, date in self.funds_tickers.items():
136        if ticker not in selected_dict:
137            continue
138        last_dates: List[datetime.date] = [x for x in date.keys() if x < self.Time.date()]
139        last_date: datetime.date = max(last_dates) if len(last_dates) > 0 else None
140        if last_date is None:
141            continue
142        
143        if len(date[last_date]) > 1:
144            fund_competition: List[float] = []
145            for fund, weight in date[last_date]:
146                fund_competition.append(sum([normalization_constant - abs(fund_z_score[fund] - fund_z_score[x]) for x, w in date[last_date] \
147                    if x != fund and abs(fund_z_score[fund] - fund_z_score[x]) <= self.target_granularity]))
148            if len(fund_competition) > 2 and sum(fund_competition) != 0:
149                COMP[selected_dict[ticker].Symbol] = np.mean(fund_competition)
150                # if ticker not in self.stock_competition:
151                #     self.stock_competition[ticker] = RollingWindow[float](self.competition_period)
152                # self.stock_competition[ticker].Add(np.mean(fund_competition))  
153                # if self.stock_competition[ticker].IsReady:
154                #     COMP[selected_dict[ticker].Symbol] = np.mean(list(self.stock_competition[ticker])[::-1][:5])
155    
156    # remove keys with duplicate COMP values for the sake of not random backtest
157    COMP_no_repl: Dict[float, Symbol] = {}
158    for key, val in COMP.items():
159        COMP_no_repl.setdefault(val, key)
160    COMP_no_repl: Dict[Symbol, float] = dict((v, k) for k, v in COMP_no_repl.items())
161        
162    # sort and divide to quantiles
163    if len(COMP_no_repl) >= self.competition_quantile * self.momentum_quantile:
164        sorted_stock_level_competitions: List[Symbol] = sorted(COMP_no_repl, key=COMP_no_repl.get)
165        quantile: int = int(len(sorted_stock_level_competitions) / self.competition_quantile)
166        lowest_competition: List[Symbol] = sorted_stock_level_competitions[:quantile]
167        # get momemtum for second sorting
168        lowest_competitions: Dict[Symbol, float] = {sym: sym_data.get_momentum() for sym, sym_data in self.data.items() if sym in lowest_competition and sym.Value in selected_dict}
169        sorted_lowest_competitions: List[Symbol] = sorted(lowest_competitions, key=lowest_competitions.get, reverse=True)
170        quantile: int = int(len(sorted_lowest_competitions) / self.momentum_quantile)
171        long: List[Symbol] = sorted_lowest_competitions[:quantile]
172        short: List[Symbol] = sorted_lowest_competitions[-quantile:]
173        # calculate weights based on values
174        for i, portfolio in enumerate([long, short]):
175            mc_sum: float = sum(list(map(lambda symbol: selected_dict[symbol.Value].MarketCap, portfolio)))
176            for symbol in portfolio:
177                self.weight[symbol] = ((-1)**i) * selected_dict[symbol.Value].MarketCap / mc_sum
178return list(self.weight.keys())
179def OnData(self, data: Slice) -> None:
180# rebalance monthly
181if not self.selection_flag:
182    return
183self.selection_flag = False
184# trade execution
185portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
186
187self.SetHoldings(portfolio, True)
188self.weight.clear()
189def Selection(self) -> None:
190if self.Time.date() > self.last_date:
191    self.Liquidate()
192    return
193self.selection_flag = True