Patent-to-Market Equity Factor

1from AlgorithmImports import *
2from enum import Enum
3from dateutil.relativedelta import relativedelta
4from pandas.tseries.offsets import BDay
5from collections import deque
6from typing import List, Dict
7#endregion
8class PortfolioWeighting(Enum):
9EQUALLY_WEIGHTED = 1
10VALUE_WEIGHTED = 2
11INVERSE_VOLATILITY_WEIGHTED = 3
12class PatentToMarketEquityFactor(QCAlgorithm):
13def Initialize(self) -> None:
14self.SetStartDate(2005, 1, 1)
15self.SetCash(100_000)
16
17# parameters
18self.reaction_period_after_patent: int = 2  # check for reaction of n days after patent grant
19self.d_period_after_patent: int = self.reaction_period_after_patent + 1 # n of needed daily prices for performance after patent grant calculation
20self.d_volatility_period: int = 60       # daily volatility calculation period
21self.m_cumulative_period: int = 12       # calculate CPM value using n-month cumulative patent performance history
22self.m_rebalance_period: int = 12        # rebalance once a n months
23self.quantile: int = 10                  # portfolio percentile selection (3-tercile; 4-quartile; 5-quintile; 10-decile and so on)
24self.leverage: int = 20
25self.portfolio_weighting: PortfolioWeighting = PortfolioWeighting.EQUALLY_WEIGHTED
26# assign larger daily period if volatility weighting is set
27if self.portfolio_weighting == PortfolioWeighting.INVERSE_VOLATILITY_WEIGHTED:
28    self.max_period: int = max(self.d_volatility_period, self.d_period_after_patent)
29else:
30    self.max_period: int = self.d_period_after_patent
31self.required_exchanges: List[str] = ['NYS', 'NAS', 'ASE']
32self.CMPs: Dict[str, float] = {}                                              # recent CPM value storage
33self.weights: Dict[Symbol, float] = {}                                        # recent portfolio selection traded weights
34self.patent_dates: Dict[datetime.datetime, list[str]] = {}                    # storing list of stocks keyed by their patent date
35self.market_moves: Dict[str, list[tuple(float, datetime.datetime.date)]] = {} # storing all market moves in one year keyed by stock's ticker
36
37# Source: https://companyprofiles.justia.com/companies
38csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/patents.csv')
39lines: List[str] = csv_string_file.split('\r\n')
40
41# select header, then exclude 'date'
42tickers: List[str] = lines[0].split(';')[1:]
43
44# store RollingWindow object keyed by stock ticker 
45self.prices: Dict[str, deque] = { ticker : deque(maxlen=self.max_period) for ticker in tickers }
46
47for line in lines[1:]:
48    if line == '':
49        continue
50    
51    line_split: List[str] = line.split(';')
52    date: datetime.date = datetime.strptime(line_split[0], "%d.%m.%Y").date()
53    
54    # initialize empty list for stock's tickers, which have patent in current date
55    self.patent_dates[date] = []
56    
57    length: int = len(line_split)
58    
59    for index in range(1, length):
60        # store stock's ticker into list, when stock has patent in current date
61        if line_split[index] != '0.0' and line_split[index] != '0':
62            self.patent_dates[date].append(tickers[index - 1])
63        
64self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
65
66# add market to prices dictionary
67self.prices[self.market.Value] = deque(maxlen=self.max_period)
68self.symbol_by_ticker:dict[str, Symbol] = {}
69self.month_counter: int = 0
70self.selection_flag: bool = False
71self.settings.daily_precise_end_time = False
72self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
73self.UniverseSettings.Resolution = Resolution.Daily
74self.AddUniverse(self.FundamentalSelectionFunction)
75self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
76def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
77for security in changes.AddedSecurities:
78    security.SetFeeModel(CustomFeeModel())
79    security.SetLeverage(self.leverage)
80def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
81# update daily prices
82for stock in fundamental:
83    ticker:str = stock.Symbol.Value
84    
85    if ticker in self.prices:
86        self.symbol_by_ticker[ticker] = stock.Symbol
87        if stock.AdjustedPrice != 0:
88            self.prices[ticker].append((self.Time.date(), stock.AdjustedPrice))
89
90days_before: datetime.datetime = (self.Time - BDay(self.reaction_period_after_patent)).date()
91# check if there was any patent granted in d_period_after_patent days before todays date
92# market has to have price data ready
93if days_before in self.patent_dates and len(self.prices[self.market.Value]) == self.prices[self.market.Value].maxlen:
94    if self.prices[self.market.Value][-self.d_period_after_patent][0] == days_before:
95        # calculate market's return for last d_period_after_patent days
96        market_return: float = self.prices[self.market.Value][-1][1] / self.prices[self.market.Value][-self.d_period_after_patent][1] - 1
97        tickers: List[str] = self.patent_dates[days_before]
98        
99        # calc market moves
100        for ticker in tickers:
101            # if not self.prices[ticker].IsReady:
102            if len(self.prices[ticker]) != self.prices[ticker].maxlen:
103                continue
104            
105            if self.prices[ticker][-self.d_period_after_patent][0] == days_before:
106                # calc stock's return for last d_period_after_patent days
107                stock_return: float = self.prices[ticker][-1][1] / self.prices[ticker][-self.d_period_after_patent][1] - 1
108                
109                # calc excess market move value
110                market_move_value: float = stock_return - market_return
111                
112                if ticker not in self.market_moves:
113                    self.market_moves[ticker] = []
114                self.market_moves[ticker].append((days_before, market_move_value))
115
116# rebalance yearly    
117if not self.selection_flag:
118    return Universe.Unchanged
119
120# select stocks, which has at least one market move value
121selected: List[Fundamental] = [
122    x for x in fundamental 
123    if x.MarketCap != 0 
124    and x.SecurityReference.ExchangeId in self.required_exchanges 
125    and x.CompanyReference.IsREIT != 1
126    and x.Symbol.Value in self.market_moves
127    ]
128
129PMT:dict[Fundamental, float] = {}   # stores stock's PMT value keyed by stock's object
130volatility:dict[Symbol, float] = {}     # stores volatility values for each symbol in current selection
131
132for stock in selected:
133    symbol: Symbol = stock.Symbol
134    ticker: str = symbol.Value
135    market_cap: float = stock.MarketCap
136    
137    # fetch only market moves stored within cumulative period window
138    sum_market_move: float = sum([x[1] for x in self.market_moves[ticker] if x[0] >= (self.Time - relativedelta(months=self.m_cumulative_period)).date()])
139    
140    # in case there isn't last_CMP use formula: CMP = MP / (g + gama), otherwise use formula: # CMP = (1 - gama) * last_CMP + MP
141    curr_CMP_value: float = 0.85 * self.CMPs[ticker] + sum_market_move if ticker in self.CMPs else sum_market_move / (0.20 + 0.15)
142    
143    # store new current CMP value keyed by stock's ticker
144    self.CMPs[ticker] = curr_CMP_value
145    
146    # calc stock's PMT value
147    PMT_value: float = curr_CMP_value / market_cap
148    
149    # store stock's PMT value keyed by stock's object
150    PMT[stock] = PMT_value
151    
152    # volatility calculation - self.d_volatility_period
153    daily_prices: np.ndarray = np.array([x[1] for x in self.prices[ticker]][-self.d_volatility_period:])
154    daily_returns: np.ndarray =  daily_prices[1:] / daily_prices[:-1] - 1
155    volatility[symbol] = np.std(daily_returns) * np.sqrt(252) # annualized volatility
156    
157# make sure, there are enough stocks for selection
158if len(PMT) < self.quantile:
159    return Universe.Unchanged
160
161# make percentile selection
162quantile: int = int(len(PMT) / self.quantile)
163sorted_by_PMT: List[Fundamental] = [x[0] for x in sorted(PMT.items(), key=lambda item: item[1])]
164
165# long highest decile
166long: List[Fundamental] = sorted_by_PMT[-quantile:]
167
168# short lowest decile
169short: List[Fundamental] = sorted_by_PMT[:quantile]
170
171# portfolio weighting
172# calculate weights for long and short portfolio part
173if self.portfolio_weighting == PortfolioWeighting.EQUALLY_WEIGHTED:
174    for i, portfolio in enumerate([long, short]):
175        for stock in portfolio:
176            self.weights[stock.Symbol] = ((-1) ** i) / len(portfolio)
177elif self.portfolio_weighting == PortfolioWeighting.VALUE_WEIGHTED:
178    for i, portfolio in enumerate([long, short]):
179        mc_sum: float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
180        for stock in portfolio:
181            self.weights[stock.Symbol] = ((-1)**i) * stock.MarketCap / mc_sum
182
183elif self.portfolio_weighting == PortfolioWeighting.INVERSE_VOLATILITY_WEIGHTED:
184    for i, portfolio in enumerate([long, short]):
185        inv_vol_sum: float = sum(list(map(lambda stock: 1 / volatility[stock.Symbol], portfolio)))
186        for stock in portfolio:
187            self.weights[stock.Symbol] = ((-1)**i) * volatility[stock.Symbol] / inv_vol_sum
188
189# return stocks symbols
190return list(self.weights.keys())
191
192def OnData(self, data: Slice) -> None:
193# wait for selection flag to be set
194if not self.selection_flag:
195    return
196self.selection_flag = False
197
198# trade execution
199portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weights.items() if symbol in data and data[symbol]]
200self.SetHoldings(portfolio, True)
201self.weights.clear()
202
203def Selection(self) -> None:
204# wait for self.m_cumulative_period months to elapse from the start of the algorithm before first selection. It gives the chance to self.market_moves to potentially fill up.
205if self.Time.date() < (self.StartDate + relativedelta(months=self.m_cumulative_period)).date():
206    return
207# rebalance once a rebalance period
208if self.month_counter % self.m_rebalance_period == 0:
209    self.selection_flag = True
210self.month_counter += 1
211
212# Custom fee model.
213class CustomFeeModel(FeeModel):
214def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
215fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
216return OrderFee(CashAmount(fee, "USD"))