Technology Momentum

1from AlgorithmImports import *
2from dateutil.relativedelta import relativedelta
3import data_tools
4from typing import List, Dict
5# endregion
6class TechnologyMomentum(QCAlgorithm):
7def Initialize(self) -> None:
8self.SetStartDate(2010, 1, 1)
9self.SetCash(100_000)
10self.international_data: bool = False
11self.tickers: List[str] = [
12    'ABAX', 'ARAY', 'ACTU', 'ADBE', 'ADVS', 'AFFX', 'A', 'ALGN', 'AOSL', 'ALIN-A', 'AMZN', 'AMD', 'AAPL', 'AMAT',
13    'AMCC', 'ARBA', 'ARUN', 'T', 'ATML', 'AXTI', 'BRCD', 'CDNS', 'CALD', 'CAVM', 'CPHD', 'CSCO', 'C', 'CDXS', 'COHR',
14    'CPTS', 'CY', 'DEPO', 'DLGC', 'DIS', 'DLB', 'DWA', 'DSPG', 'EBAY', 'ELON', 'EA', 'EFII', 'EPOC', 'EQIX', 'EXAR',
15    'EXEL', 'EXTR', 'FB', 'FCS', 'FNSR', 'F', 'FORM', 'FTNT', 'GE', 'GM', 'GDHX', 'GILD', 'GOOGL', 'GSIT', 'HLIT',
16    'HPQ', 'HMC', 'IBM', 'IGTE', 'IKAN', 'IPXL', 'INFN', 'INFA', 'IDIT', 'ISSI', 'INTC', 'ISIL', 'IVAC', 'INTU',
17    'ISRG', 'INVN', 'IPA', 'IXYS', 'JDSU', 'JNPR', 'KEYN', 'KLAC', 'LRCX', 'LF', 'LLTC', 'LSI', 'LLC', 'MTSN',
18    'MXIM', 'MERU', 'MCRL', 'MSFT', 'MPWR', 'ONTO', 'NTUS', 'NPTN', 'NTAP', 'NFLX', 'NTGR', 'N', 'NSANY', 'NVLS',
19    'NVDA', 'OCLR', 'OCZ', 'OMCL', 'OVTI', 'ONXX', 'OPEN_old', 'OPWV', 'OPLK', 'OPXT', 'ORCL', 'P', 'PSEM', 'PLXT',
20    'PMCS', 'PLCM', 'POWI', 'QMCO', 'QNST', 'RMBS', 'MKTG', 'RVBD', 'ROVI', 'SABA', 'CRM', 'SNDK', 'SANM', 'SCLN',
21    'SREV', 'SHOR', 'SFLY', 'SIGM', 'SGI', 'SIMG', 'SLTM', 'CODE', 'SPWR', 'SMCI', 'NLOK', 'SYMM', 'SYNA', 'SNX',
22    'SNPS', 'TNAV', 'TSLA', 'XPER', 'THOR', 'TIBX', 'TIVO', 'TYO', 'TRID', 'TRMB', 'TWTR', 'UI', 'UCTT', 'ULTRACEMCO',
23    'VAR', 'PAY', 'VMW', 'VLTR', 'WMT', 'XLNX', 'YAHOO', 'DZSI', 'ZNGA'
24]
25self.years_period: int = 5
26self.min_prices: int = 15
27self.quantile: int = 10
28self.leverage: int = 5
29self.weights: Dict[Symbol, float] = {}
30self.count_by_class_by_date: Dict[datetime.date, Dict[str, int]] = {}
31self.data: Dict[str, data_tools.SymbolData] = {}
32self.classification_symbol_by_ticker: Dict[str, Symbol] = {}
33for ticker in self.tickers:
34    classification_symbol: Symbol = self.AddData(data_tools.QuantpediaPatentClassification, ticker, Resolution.Daily).Symbol
35    self.classification_symbol_by_ticker[ticker] = classification_symbol
36    self.data[ticker] = data_tools.SymbolData()
37market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
38self.selection_flag: bool = False
39self.UniverseSettings.Leverage = self.leverage
40self.UniverseSettings.Resolution = Resolution.Daily
41self.AddUniverse(self.FundamentalSelectionFunction)
42self.settings.daily_precise_end_time = False
43self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
44self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
45def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
46for security in changes.AddedSecurities:
47    security.SetFeeModel(data_tools.CustomFeeModel())
48def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
49# daily update prices required in stock's monthly return calculation
50for equity in fundamental:
51    ticker: str = equity.Symbol.Value
52    if ticker in self.data:
53        self.data[ticker].update_prices(equity.AdjustedPrice)
54if not self.selection_flag:
55    return Universe.Unchanged
56start_boundry_date: datetime.date = self.Time.date() - relativedelta(years=self.years_period)
57for _, symbol_obj in self.data.items():
58    # reset of prices and update of monthly return, which is required for each stock on monthly basis, to keep valid data
59    if symbol_obj.prices_ready(self.min_prices):
60        symbol_obj.update_monthly_return()
61    else:
62        symbol_obj.reset_monthly_return()
63    symbol_obj.reset_prices()
64    
65    # remove stock's patents classifications, which won't be needed anymore
66    symbol_obj.remove_needless_classifications(start_boundry_date)
67selected: List[Fundamental] = [
68    x for x in fundamental
69    if x.Symbol.Value in self.data
70    and x.MarketCap != 0
71    and self.data[x.Symbol.Value].monthly_return_ready() 
72    and self.data[x.Symbol.Value].classifications_ready()]
73if len(selected) == 0:
74    return Universe.Unchanged    
75dates_to_remove: datetime.date = []
76companies_patents_classes: List[str] = []
77start_boundry_date: datetime.date = self.Time.date() - relativedelta(years=self.years_period)
78for date, count_by_class in self.count_by_class_by_date.items():
79    if date < start_boundry_date:
80        dates_to_remove.append(date)
81        continue
82    for patent_class, count in count_by_class.items():
83        if patent_class not in companies_patents_classes:
84            companies_patents_classes.append(patent_class)
85# these days won't be needed anymore
86for date in dates_to_remove:
87    del self.count_by_class_by_date[date]
88monthly_return_by_stock: Dict[Fundamental, float] = {}
89vector_by_stock: Dict[Fundamental, np.ndarray] = {}
90for stock in selected:
91    ticker: str = stock.Symbol.Value
92    proportional_vector: List[int] = self.data[ticker].get_proportional_vector(companies_patents_classes)
93    vector_by_stock[stock] = np.array(proportional_vector)
94    monthly_return: float = self.data[ticker].get_monthly_return()
95    monthly_return_by_stock[stock] = monthly_return
96TECH_RET: Dict[Fundamental, float] = {}
97for stock1, vector1 in vector_by_stock.items():
98    TECH: float = 0
99    RET: float = 0
100    for stock2, vector2 in vector_by_stock.items():
101        if stock1 != stock2:
102            scala_product: float = vector1.dot(vector2)
103            if scala_product == 0:
104                continue
105            TECH += scala_product / ((scala_product ** 0.5) * (scala_product ** 0.5))
106            RET += monthly_return_by_stock[stock2]
107    if TECH != 0 and RET != 0:
108        TECH_RET[stock1] = (TECH * RET) / TECH
109if len(TECH_RET) < self.quantile:
110    return Universe.Unchanged
111quantile: int = int(len(TECH_RET) / self.quantile)
112sorted_by_TECH_RET: List[Fundamental] = [x[0] for x in sorted(TECH_RET.items(), key=lambda item: item[1])]
113long_leg: List[Fundamental] = sorted_by_TECH_RET[-quantile:]
114short_leg: List[Fundamental] = sorted_by_TECH_RET[:quantile]
115for i, portfolio in enumerate([long_leg, short_leg]):
116    mc_sum: float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
117    for stock in portfolio:
118        self.weights[stock.Symbol] = ((-1)**i) * stock.MarketCap / mc_sum
119return list(self.weights.keys())
120
121def OnData(self, data: Slice) -> None:
122curr_date: datetime.date = self.Time.date()
123custom_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.QuantpediaPatentClassification.get_last_update_date()
124latest_update_date: datetime.date = max(date for date in custom_data_last_update_date.values())
125stock_to_delete: List[str] = []
126# if self.time.date() > latest_update_date:
127#     self.liquidate()
128#     return
129for stock_ticker, classification_symbol in self.classification_symbol_by_ticker.items():
130    if self.Securities[stock_ticker].GetLastData() and self.Time.date() > custom_data_last_update_date[stock_ticker]:
131        self.Liquidate(stock_ticker)
132        stock_to_delete.append(stock_ticker)
133        continue
134    
135    if data.contains_key(classification_symbol) and data[classification_symbol]:
136        if curr_date not in self.count_by_class_by_date:
137            self.count_by_class_by_date[curr_date] = {}
138        symbol_obj: SymbolData = self.data[stock_ticker]
139        wanted_property: str = 'international' if self.international_data else 'u.s.'
140        patent_classes: List[str] = list(data[classification_symbol].GetProperty(wanted_property))
141        for patent_class in patent_classes:
142            if patent_class not in self.count_by_class_by_date[curr_date]:
143                self.count_by_class_by_date[curr_date][patent_class] = 0
144            self.count_by_class_by_date[curr_date][patent_class] += 1
145            symbol_obj.update_patents(curr_date, patent_class)
146for ticker in stock_to_delete:
147    self.classification_symbol_by_ticker.pop(ticker)
148# rebalance monthly
149if not self.selection_flag:
150    return
151self.selection_flag = False
152
153# trade execution
154portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weights.items() if data.contains_key(symbol) and data[symbol]]
155self.SetHoldings(portfolio, True)
156self.weights.clear()
157
158def Selection(self) -> None:
159self.selection_flag = True