Categorization Effect in Stocks

1from AlgorithmImports import *
2from typing import List, Dict
3import pandas as pd
4import numpy as np
5# endregion
6class CategorizationEffectInStocks(QCAlgorithm):
7def Initialize(self) -> None:
8self.SetStartDate(2002, 1, 1)
9self.SetCash(100_000)
10self.UniverseSettings.Leverage = 5
11self.UniverseSettings.Resolution = Resolution.Daily
12self.AddUniverse(self.FundamentalSelectionFunction)
13self.Settings.MinimumOrderMarginPortfolioPercentage = 0.0
14self.settings.daily_precise_end_time = False
15
16self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
17self.week_period: int = 5
18self.quantile: int = 5
19self.selection_flag: bool = False
20
21self.prices: Dict[str, RollingWindow] = {}
22self.firm_similarity: Dict[int, Dict[str, Dict[str, float]]] = {}
23self.yearly_universes: Dict[int, List[str]] = {}
24self.long_symbols: List[Symbol] = []
25self.short_symbols: List[Symbol] = []
26csv: str = self.Download('data.quantpedia.com/backtesting_data/equity/industries/firm_similarity.csv')
27lines: List[str] = csv.split('\r\n')
28for line in lines[1:]: # Skip header
29    if line == '':
30        continue
31    line_split: List[str] = line.split(';')
32    year: int = int(line_split[0].split('-')[0])
33    if year not in self.firm_similarity:
34        self.firm_similarity[year]: Dict = {}
35    if year not in self.yearly_universes:
36        self.yearly_universes[year]: List = []
37    industry_comp_ticker: str = line_split[1]
38    if industry_comp_ticker not in self.firm_similarity[year]:
39        self.firm_similarity[year][industry_comp_ticker]: Dict = {}
40    if industry_comp_ticker not in self.yearly_universes[year]:
41        self.yearly_universes[year].append(industry_comp_ticker)
42    company_ticker: str = line_split[2]
43    score: float = float(line_split[3])
44    self.firm_similarity[year][industry_comp_ticker][company_ticker]: float = score
45    self.yearly_universes[year].append(company_ticker)
46market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
47self.Schedule.On(
48    self.DateRules.EveryDay(market), 
49    self.TimeRules.BeforeMarketClose(market), 
50    self.Selection)
51def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
52for security in changes.AddedSecurities:
53    security.SetFeeModel(CustomFeeModel())
54def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
55# Update daily prices
56for f in fundamental:
57    ticker: str = f.Symbol.Value
58    if ticker in self.prices:
59        self.prices[ticker].Add(f.Price)
60if not self.selection_flag:
61    return Universe.Unchanged
62prev_year: int = self.Time.year - 1
63
64if prev_year not in self.firm_similarity:
65    return Universe.Unchanged
66
67# Select universe
68curr_year_universe: List[str] = self.yearly_universes[prev_year]
69selected: List[Fundamental] = [f for f in fundamental if f.HasFundamentalData
70                        and f.Symbol.Value in curr_year_universe
71                        and f.SecurityReference.ExchangeId in self.exchange_codes
72                        and not np.isnan(f.AssetClassification.MorningstarIndustryGroupCode)]
73
74warmed_up_stocks: List[Fundamental] = []
75# Warm up stock prices
76for f in selected:
77    symbol: Symbol = f.Symbol
78    ticker: str = symbol.Value
79    if ticker not in self.prices:
80        self.prices[ticker] = RollingWindow[float](self.week_period)
81        history: pd.DataFrame = self.History(symbol, self.week_period, Resolution.Daily)
82        if history.empty:
83            continue
84        closes: pd.Series = history.loc[symbol].close
85        for _, close in closes.items():
86            self.prices[ticker].Add(close)
87    if self.prices[ticker].IsReady:
88        warmed_up_stocks.append(f)
89csv_industries: Dict[str, Dict[str, float]] = self.firm_similarity[prev_year]
90industry_code_ticker: Dict[str, str] = {}
91industries_groups: Dict[str, List[Symbol]] = {}
92industry_diff_by_symbol: Dict[Symbol, float] = {}
93symbol_by_ticker: Dict[str, Symbol] = {}
94# Create industries groups
95for f in warmed_up_stocks:            
96    symbol: Symbol = f.Symbol
97    ticker: str = symbol.Value
98    industry_group_code: str = f.AssetClassification.MorningstarIndustryGroupCode
99    if ticker in csv_industries:
100        # Create match of csv industry group with QC industry group
101        industry_code_ticker[industry_group_code]: str = ticker
102        # Create match between stock's ticker and it's symbol, because this stock can be traded 
103        symbol_by_ticker[ticker] = symbol
104    if industry_group_code not in industries_groups:
105        industries_groups[industry_group_code]: List[Symbol] = []
106    industries_groups[industry_group_code].append(f.Symbol)
107# Calculate difference between industry performances
108for industry_group_code, ticker in industry_code_ticker.items():
109    industry_universe: List[Symbol] = industries_groups[industry_group_code]
110    # Official industry performance calculation
111    official_industry_perf: float = np.mean([self.Performance(self.prices[symbol.Value]) 
112                                                for symbol in industry_universe])
113    
114    # Fundamental industry performance calculation
115    tickers_similarities: Dict[str, float] = self.firm_similarity[prev_year][ticker]
116    fundamental_industry_perf: float = sum([self.Performance(self.prices[stock_ticker]) * similarity_score 
117                                            for stock_ticker, similarity_score in tickers_similarities.items() 
118                                                if stock_ticker in self.prices and self.prices[stock_ticker].IsReady])
119    total_industry_score: float = sum(list(tickers_similarities.values()))
120    if fundamental_industry_perf != 0 and total_industry_score != 0:
121        fundamental_industry_perf /= total_industry_score
122        
123        trade_symbol: Symbol = symbol_by_ticker[ticker]
124        curr_industries_diff: float = official_industry_perf - fundamental_industry_perf
125        industry_diff_by_symbol[trade_symbol] = fundamental_industry_perf
126# Make sure there are enough stocks for selection
127if len(industry_diff_by_symbol) < self.quantile:
128    return Universe.Unchanged
129quantile: int = int(len(industry_diff_by_symbol) / self.quantile)
130sorted_by_diff: List[Symbol] = [x[0] for x in sorted(industry_diff_by_symbol.items(), key=lambda item: item[1])]
131self.long_symbols = sorted_by_diff[:quantile]
132self.short_symbols = sorted_by_diff[-quantile:]
133return self.long_symbols + self.short_symbols
134
135def OnData(self, slice: Slice) -> None:
136# Rebalance weekly
137if not self.selection_flag:
138    return
139self.selection_flag = False
140
141# Trade execution
142targets: List[PortfolioTarget] = []
143for i, portfolio in enumerate([self.long_symbols, self.short_symbols]):
144    for symbol in portfolio:
145        if slice.ContainsKey(symbol) and slice[symbol] is not None:
146            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
147self.SetHoldings(targets, True)
148self.long_symbols.clear()
149self.short_symbols.clear()
150def Performance(self, prices_roll_window: RollingWindow) -> float:
151return (prices_roll_window[0] / prices_roll_window[prices_roll_window.Count - 1]) - 1
152def Selection(self) -> None:
153if self.Time.weekday() == 0:
154    self.selection_flag = True
155class CustomFeeModel(FeeModel):
156def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
157fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
158return OrderFee(CashAmount(fee, "USD"))