Directional Momentum

1from AlgorithmImports import *
2import pandas as pd
3import data_tools
4from typing import List, Dict
5import statsmodels.api as sm
6from dateutil.relativedelta import relativedelta
7from collections import deque
8import numpy as np
9from pandas.core.frame import DataFrame
10from pandas.core.series import Series
11# endregion
12
13class DirectionalMomentum(QCAlgorithm):
14
15def Initialize(self):
16self.SetStartDate(2000, 1, 1)
17self.SetCash(100000)
18
19self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']    
20
21self.data: Dict[Symbol, data_tools.SymbolData] = {}
22self.weight: Dict[Symbol, float] = {}
23
24self.warmup_period: int = 12
25self.period: int = 60
26self.month_period: int = 21
27self.leverage: int = 5
28self.quantile: int = 10
29
30self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
31
32self.fundamental_count: int = 500
33self.fundamental_sorting_key = lambda x: x.DollarVolume
34
35self.selection_flag: bool = False
36self.UniverseSettings.Resolution = Resolution.Daily
37self.AddUniverse(self.FundamentalSelectionFunction)
38self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
39self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
40
41def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
42for security in changes.AddedSecurities:
43    security.SetFeeModel(data_tools.CustomFeeModel())
44    security.SetLeverage(self.leverage)
45    
46def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
47# update the data every day
48for stock in fundamental:
49    symbol: Symbol = stock.Symbol
50    
51    # store monthly price
52    if symbol in self.data:
53        self.data[symbol].update_daily_return(self.Time.date(), stock.AdjustedPrice)
54        if self.selection_flag:
55            self.data[symbol].update_monthly_return(stock.AdjustedPrice)
56    
57if not self.selection_flag:
58    return Universe.Unchanged
59
60selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
61                          and x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchange_codes]
62
63if len(selected) > self.fundamental_count:
64    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
65
66selected_dict: Dict[Symbol, Fundamental] = {x.Symbol: x for x in selected}
67
68# warmup price rolling windows
69for stock in selected + [self.market]:
70    if stock == self.market:
71        symbol: Symbol = stock
72    else:
73        symbol: Symbol = stock.Symbol
74
75    if symbol not in self.data:
76        self.data[symbol] = data_tools.SymbolData(self.period * self.month_period, self.warmup_period * self.month_period, self.period)
77        history: DataFrame = self.History(symbol, self.period * self.month_period, Resolution.Daily)
78        if history.empty:
79            self.Log(f"Not enough data for {symbol} yet.")
80            continue
81        data: Series = history.loc[symbol]
82        monthly_data: Series = data.groupby(pd.Grouper(freq='MS')).last()
83        for time, row in data.iterrows():
84            self.data[symbol].update_daily_return(time.date(), row.close)
85        for time, row in monthly_data.iterrows():
86            self.data[symbol].update_monthly_return(row.close)
87
88if len(selected) != 0:
89
90    returns: Dict[Symbol, List[float]] = {symbol: data.get_returns_idiosyncratic_regression() for symbol, data in self.data.items() if data.is_ready_for_first_regression() and symbol in list(selected_dict.keys()) and symbol != self.market}
91    stock_returns: List[List[float]] = list(zip(*[[i for i in x] for x in returns.values()]))
92
93    if len(returns) != 0:
94        # get idiosyncratic variance
95        x: np.ndarray = np.array(self.data[self.market].get_returns_idiosyncratic_regression())
96        y: np.ndarray = stock_returns[-self.month_period:]
97        model = self.multiple_linear_regression(x, y)
98        variance: float = np.std(model.resid, axis=0) ** 2
99
100        for i, symbol in enumerate(returns):
101            self.data[symbol].update_variance(variance[i])
102
103        reg_data_dict: Dict[Symbol,List[Tuple[datetime.date, float]]] = {symbol: self.data[symbol].get_variance_values() for symbol in list(returns.keys()) if self.data[symbol].is_ready_for_regression()}
104    
105        if len(reg_data_dict) != 0:
106            reg_data: List[List[float]] = list(zip (*[[i for i in x] for x in reg_data_dict.values()]))
107            df_stocks: DataFrame = pd.DataFrame(index=self.data[self.market].get_dates())
108
109            returns: Dict[Symbol, List[float]] = {symbol: data.get_data() for symbol, data in self.data.items() if data.is_ready_for_regression() and symbol in list(selected_dict.keys()) and symbol != self.market}
110
111            # fill dataframe with data
112            for symbol, data in returns.items():
113                symbol_df: DataFrame = pd.DataFrame(data, columns=['date', 'return'])
114                symbol_df['date'] = pd.to_datetime(symbol_df['date'])
115                symbol_df = symbol_df.set_index('date')
116                symbol_df = symbol_df.loc[~symbol_df.index.duplicated(keep='first')]
117
118                df_stocks[symbol] = symbol_df['return']
119
120            df_stocks = df_stocks.ffill().fillna(0)
121
122            # get current positive and negative runs
123            df_reg: DataFrame = self.longest_consecutive_run_for_stocks(df_stocks)[-self.period:]
124
125            independent_variable: np.ndarray = np.column_stack((reg_data, self.data[self.market].get_monthly_returns(), df_reg.values))
126
127            # # run stock regression
128            x: np.ndarray = independent_variable[:-1]
129            y: np.ndarray = np.where(df_stocks.groupby(pd.Grouper(freq='M')).sum()[-self.period:].values[1:] < 0, 0, 1) # change to binary linear probability model
130            model = self.multiple_linear_regression(x, y)
131            predicted_y: np.ndarray = model.predict(sm.add_constant(independent_variable[-1:], has_constant='add'))[0]
132
133            PS: Dict[Symbol, float] = {self.Symbol(symbol): predicted_y[i] for i, symbol in enumerate(list(df_stocks.columns))}
134
135            # sort and divide to quantiles
136            if len(PS) >= self.quantile:
137                sorted_PS: List[Symbol] = sorted(PS, key=PS.get, reverse=True)
138                quantile: int = int(len(sorted_PS) / self.quantile)
139                long: List[Symbol] = sorted_PS[:quantile]
140                short: List[Symbol] = sorted_PS[-quantile:]
141
142            # calculate weights
143            for i, portfolio in enumerate([long, short]):
144                mc_sum: float = sum(list(map(lambda symbol: selected_dict[symbol].MarketCap, portfolio)))
145                for symbol in portfolio:
146                    self.weight[symbol] = ((-1)**i) * selected_dict[symbol].MarketCap / mc_sum
147
148return list(self.weight.keys())
149
150def OnData(self, data: Slice):
151if not self.selection_flag:
152    return
153self.selection_flag = False
154
155# trade execution
156portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
157self.SetHoldings(portfolio, True)
158
159self.weight.clear()
160
161def Selection(self) -> None:
162self.selection_flag = True
163
164def multiple_linear_regression(self, x: np.ndarray, y: np.ndarray):
165# x:np.ndarray = np.array(x).T
166x = sm.add_constant(x, prepend=True)
167result = sm.OLS(endog=y, exog=x).fit()
168return result
169
170def longest_consecutive_run_for_stocks(self, df, threshold=0) -> DataFrame:
171df_daily: DataFrame = df
172df_daily.index = pd.to_datetime(df_daily.index)
173df_weekly: DataFrame = df_daily.groupby(pd.Grouper(freq='W')).sum()
174df_monthly: Dataframe = df_daily.groupby(pd.Grouper(freq='M')).sum()
175df_list: List[DataFrame] = [df_daily, df_weekly, df_monthly]
176
177df_end: DataFrame = pd.DataFrame()
178
179for df in df_list:
180    positive_df: DataFrame = df.applymap(lambda x: 1 if x > 0 else 0).apply(lambda col: col.groupby((col != col.shift()).cumsum()).cumsum()).groupby(pd.Grouper(freq='M')).last()
181    negative_df: DataFrame = df.applymap(lambda x: 1 if x < 0 else 0).apply(lambda col: col.groupby((col != col.shift()).cumsum()).cumsum()).groupby(pd.Grouper(freq='M')).last()
182
183    df_end: DataFrame = pd.concat([df_end, positive_df, negative_df], axis=1, ignore_index=True)
184
185return df_end