Stock Momentum Volatility Switching Strategy

1from AlgorithmImports import *
2import pandas as pd
3import numpy as np
4from collections import deque
5import data_tools
6from numpy import isnan
7from pandas.core.frame import DataFrame
8from pandas.core.series import Series
9
10class SwitchingbetweenValueMomentum(QCAlgorithm):
11
12def Initialize(self) -> None:
13self.SetStartDate(2000, 1, 1)
14self.SetCash(100000)
15
16self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']	
17self.ticker_to_ignore: List[str] = ['TOPS', 'SSCC']
18
19self.data: Dict[Symbol, SymbolData] = {}
20self.performance_data: DataFrame = pd.DataFrame()
21
22self.period: int = 120
23self.data_period: int = 12
24self.month_period: int = 21
25self.leverage: int = 5
26self.quantile: int = 10
27self.min_traded_weight: float = 0.00001
28self.short_momentum_period: int = 3
29self.long_momentum_period: int = 12
30
31self.short_term_long: List[Symbol] = []
32self.short_term_short: List[Symbol] = []
33self.long_term_long: List[Symbol] = []
34self.long_term_short: List[Symbol] = []
35
36market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
37
38self.fundamental_count: int = 500
39self.fundamental_sorting_key = lambda x: x.DollarVolume
40
41self.selection_flag: bool = False
42self.UniverseSettings.Resolution = Resolution.Daily
43self.AddUniverse(self.FundamentalSelectionFunction)
44self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
45self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
46
47def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
48for security in changes.AddedSecurities:
49    security.SetFeeModel(data_tools.CustomFeeModel())
50    security.SetLeverage(self.leverage)
51    
52def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
53if not self.selection_flag:
54    return Universe.Unchanged
55
56# update the rolling window every month
57for stock in fundamental:
58    symbol: Symbol = stock.Symbol
59    
60    # store monthly price
61    if symbol in self.data:
62        self.data[symbol].update(stock.AdjustedPrice)
63
64selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
65                            and x.SecurityReference.ExchangeId in self.exchange_codes and x.Symbol.Value not in self.ticker_to_ignore]
66
67if len(selected) > self.fundamental_count:
68    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
69
70short_term_momentum: Dict[Symbol, float] = {}
71long_term_momentum: Dict[Symbol, float] = {}
72
73# warmup price rolling windows
74for stock in selected:
75    symbol: Symbol = stock.Symbol
76
77    if symbol not in self.data:
78        self.data[symbol] = data_tools.SymbolData(self.data_period)
79        history: DataFrame = self.History(symbol, self.data_period * self.month_period, Resolution.Daily)
80        if history.empty:
81            self.Log(f"Not enough data for {symbol} yet.")
82            continue
83        data: Series = history.loc[symbol]
84        monthly_data: Series = data.groupby(pd.Grouper(freq='MS')).last()
85        for time, row in monthly_data.iterrows():
86            self.data[symbol].update(row.close)
87    
88    if self.data[symbol].is_ready():
89        short_term_momentum[symbol] = self.data[symbol].momentum(self.short_momentum_period)
90        long_term_momentum[symbol] = self.data[symbol].momentum(self.long_momentum_period)
91
92if len(short_term_momentum) >= self.quantile and len(long_term_momentum) >= self.quantile:
93    # sorting by long term and short term momentum
94    sorted_short_term_momentum: List[Symbol] = sorted(short_term_momentum, key = short_term_momentum.get, reverse=True)
95    sorted_long_term_momentum: List[Symbol] = sorted(long_term_momentum, key = long_term_momentum.get, reverse=True)
96    quantile: int = int(len(sorted_short_term_momentum) / self.quantile)
97
98    self.short_term_long = sorted_short_term_momentum[:quantile]
99    self.short_term_short = sorted_short_term_momentum[-quantile:]
100    self.long_term_long = sorted_long_term_momentum[:quantile]
101    self.long_term_short = sorted_long_term_momentum[-quantile:]
102
103return self.short_term_long + self.short_term_short + self.long_term_long + self.long_term_short
104
105def OnData(self, data: Slice) -> None:
106if not self.selection_flag:
107    return
108self.selection_flag = False
109
110if len(set(self.short_term_long + self.short_term_short)) == 0 or len(set(self.long_term_long + self.long_term_short)) == 0: 
111    return
112
113portfolio_list: List[List[Symbol]] = [self.short_term_long, self.short_term_short, self.long_term_long, self.long_term_short]
114
115# optimization process
116price_data: List[Dict[Symbol, List[float]]] = [{ symbol : self.data[symbol].get_prices() for symbol in portfolio if symbol in data and data[symbol]} for portfolio in portfolio_list]
117returns_df_list: List[DataFrame] = [pd.DataFrame(portfolio_prices, columns=portfolio_prices.keys()).pct_change().dropna() for portfolio_prices in price_data]
118
119# store factors' performance
120df_returns: DataFrame = pd.concat([(returns_df_list[0].sum(axis=1) - returns_df_list[1].sum(axis=1)), (returns_df_list[2].sum(axis=1) - returns_df_list[3].sum(axis=1))], axis=1)
121self.performance_data = pd.concat([self.performance_data, df_returns], axis=0)
122
123if len(self.performance_data.index) < self.period:
124    return
125
126self.performance_data = self.performance_data[-self.period:]
127optimiztion = data_tools.PortfolioOptimization(self.performance_data, 0, df_returns.shape[1])
128opt_weight = optimiztion.opt_portfolio()
129
130if isnan(sum(opt_weight)):
131    return
132
133trade_quantities: Dict[Symbol, float] = {}
134
135for i, term in enumerate([[self.short_term_long, self.short_term_short], [self.long_term_long, self.long_term_short]]):
136    for n, portfolio in enumerate(term):
137        w: float = opt_weight[i]
138        for symbol in portfolio:
139            if w > self.min_traded_weight:
140                if symbol in data and data[symbol]:
141                    quantity: float = ((self.Portfolio.TotalPortfolioValue / len(portfolio)) * w) // data[symbol].Price
142                    if symbol not in trade_quantities:
143                        trade_quantities[symbol] = 0
144                    trade_quantities[symbol] += ((-1) ** n) * quantity
145
146# trade execution
147stocks_invested: List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
148for symbol in stocks_invested:
149    if symbol not in trade_quantities:
150        self.Liquidate(symbol)
151
152for symbol, new_quantity in trade_quantities.items():
153    quantity:float = new_quantity - self.Portfolio[symbol].Quantity
154    if abs(quantity) >= 1.:
155        self.MarketOrder(symbol, quantity)
156
157def Selection(self) -> None:
158self.selection_flag = True