Co-Skewness Enhanced Momentum

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from typing import List, Dict
4import numpy as np
5# endregion
6
7class CoSkewnessEnhancedMomentum(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2000, 1, 1)
11self.SetCash(100000) 
12
13self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
14self.tickers_to_ignore:List[str] = ['KELYB']
15
16self.data:Dict[Symbol, SymbolData] = {}
17self.CS_wml:List[float] = []
18self.six_month_std:List[float] = []
19self.long:list[Symbol] = []
20self.short:list[Symbol] = []
21self.weight:float = 0.
22self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
23
24self.quantile:int = 10
25self.leverage:int = 5
26self.month_period:int = 21
27self.period:int = 60
28self.six_month_period:int = 6 * self.month_period
29self.estimation_period:int = 6
30self.leverage_cap:float = 3.
31
32self.fundamental_count:int = 500
33self.selection_flag:bool = False
34self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
35self.UniverseSettings.Resolution = Resolution.Daily
36self.AddUniverse(self.FundamentalSelectionFunction)
37self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
38
39def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
40for security in changes.AddedSecurities:
41    security.SetLeverage(self.leverage)
42    security.SetFeeModel(CustomFeeModel())
43
44def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
45# update the rolling window every day
46for stock in fundamental:
47    symbol:Symbol = stock.Symbol
48
49    # store monthly price
50    if symbol in self.data:
51        self.data[symbol].update_data(stock.AdjustedPrice, self.Time.month)
52
53if not self.selection_flag:
54    return Universe.Unchanged
55
56selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value not in self.tickers_to_ignore and x.MarketCap != 0 \
57                            and x.SecurityReference.ExchangeId in self.exchange_codes]
58# selected:List[Fundamental] = [x
59#     for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value not in self.tickers_to_ignore \
60#     and x.MarketCap != 0 \
61#     and x.SecurityReference.ExchangeId in self.exchange_codes],
62#     key = lambda x: x.DollarVolume, reverse = True)[:self.fundamental_count]]
63
64if len(selected) > self.fundamental_count:
65    selected = sorted(selected, key=lambda x: x.DollarVolume, reverse=True)[:self.fundamental_count]
66
67selected:Dict[Symbol, Fundamental] = {x.Symbol: x for x in selected}
68
69# warmup price rolling windows
70for symbol in list(selected.keys()) + [self.market]:
71    if symbol in self.data:
72        continue
73    
74    self.data[symbol] = SymbolData(self.six_month_period, self.period)
75    history:DataFrame = self.History(symbol, self.period * self.month_period, Resolution.Daily)
76    if history.empty:
77        self.Log(f"Not enough data for {symbol} yet.")
78        continue
79    closes:pd.Series = history.loc[symbol].close
80    for time, close in closes.items():
81        self.data[symbol].update_data(close, time.date().month)
82
83if len(selected) != 0:
84    if self.data[self.market].is_ready():
85        stock_momentum:Dict[Symbol, float] = {symbol: value.get_momentum() for symbol, value in self.data.items() if value.is_ready() and symbol in selected}
86        if len(stock_momentum) >= self.quantile:
87            sorted_momentum:List[Symbol] = sorted(stock_momentum, key=stock_momentum.get, reverse=True)
88            quantile:int = len(sorted_momentum) // self.quantile
89            self.long:List[Symbol] = sorted_momentum[:quantile]
90            self.short:List[Symbol] = sorted_momentum[-quantile:]
91
92        # mean calculation of monthly returns of selected portfolio
93        winners_performance:np.ndarray = np.array([self.data[x].get_monthly_returns() for x in self.long if self.data[x].is_ready()], dtype=float)
94        losers_performance:np.ndarray = np.array([self.data[x].get_monthly_returns() for x in self.short if self.data[x].is_ready()], dtype=float)
95        winners_mean_performance:np.ndarray = np.mean(winners_performance, axis=0)
96        losers_mean_performance:np.ndarray = np.mean(losers_performance, axis=0)
97        
98        market_monthly_returns:np.ndarray = np.array(self.data[self.market].get_monthly_returns(), dtype=float)
99
100        # standard deviation calculation of daily returns over 6 months
101        winners_daily_returns:np.ndarray = np.array([self.data[x].get_daily_returns() for x in self.long], dtype=float)
102        losers_daily_returns:np.ndarray = np.array([self.data[x].get_daily_returns() for x in self.short], dtype=float)
103        winners_daily_returns:float = np.mean(winners_daily_returns, axis=0)
104        losers_daily_returns:float = np.mean(losers_daily_returns, axis=0)
105        six_month_std:float = np.std(winners_daily_returns - losers_daily_returns)
106        self.six_month_std.append(six_month_std)
107        
108        # CS WML calculation
109        std_wml = np.std(winners_mean_performance - losers_mean_performance)
110        std_market = np.std(market_monthly_returns)
111        CS_wml = (np.cov(winners_mean_performance, market_monthly_returns ** 2)[0][1] - np.cov(losers_mean_performance, market_monthly_returns ** 2)[0][1]) / (std_wml * (std_market ** 2))
112        self.CS_wml.append(CS_wml)
113    
114    # weight calculation
115    if len(self.CS_wml) >= self.estimation_period and len(self.six_month_std) >= self.estimation_period:
116        last_CS_wml:float = self.CS_wml[-1]
117        last_std:float = self.six_month_std[-1]
118        std_target:float = np.mean(self.six_month_std)
119        CS_wml_target:float = np.mean(self.CS_wml)
120        CS_range:float = max(self.CS_wml) - min(self.CS_wml)
121
122        self.weight:float = (1 + ((CS_wml_target - last_CS_wml) / CS_range)) * (std_target / last_std)
123        self.weight = min([self.leverage_cap, max([-self.leverage_cap, self.weight])])
124
125return self.long + self.short
126
127def OnData(self, data:Slice) -> None:
128# monthly rebalance
129if not self.selection_flag:
130    return
131self.selection_flag = False
132
133if self.weight > 0.:
134    targets:List[PortfolioTarget] = []
135    for i, portfolio in enumerate([self.long, self.short]):
136        for symbol in portfolio:
137            if symbol in data and data[symbol]:
138                targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio) * self.weight))
139    
140    self.SetHoldings(targets, True)
141else:
142    self.Liquidate()
143
144self.long.clear()
145self.short.clear()
146self.weight = 0.
147
148def Selection(self) -> None:
149self.selection_flag = True
150
151class SymbolData():
152def __init__(self, daily_period:int, monthly_period:int) -> None:
153self._month:int = -1
154self._previous_price:float|None = None
155self._daily_price:List[float] = []
156self._monthly_return:RollingWindow = RollingWindow[float](monthly_period)
157self._daily_returns:RollingWindow = RollingWindow[float](daily_period)
158
159def update_data(self, price:float, month:int) -> None:
160if self._previous_price is None:
161    self._previous_price = price
162    return
163
164if self._month != month and len(self._daily_price) != 0:
165    self.update_monthly_data()
166
167self._month = month
168daily_return:float = (price - self._previous_price) / self._previous_price
169self._daily_returns.Add(daily_return)
170self._previous_price = price
171self._daily_price.append(price)
172
173def update_monthly_data(self) -> None:
174monthly_return:float = (self._daily_price[-1] - self._daily_price[0]) / self._daily_price[0]
175self._monthly_return.Add(monthly_return)
176self._daily_price.clear()
177
178def get_momentum(self) -> float:
179if self._monthly_return[11] == 0:
180    return 0.
181
182momentum:float = self._monthly_return[0] / self._monthly_return[11] - 1
183return momentum
184
185def get_daily_returns(self) -> float:
186return list(self._daily_returns)
187
188def get_monthly_returns(self) -> float:
189return list(self._monthly_return)
190
191def is_ready(self) -> bool:
192return self._monthly_return.IsReady and self._daily_returns.IsReady and self._previous_price is not None
193
194# custom fee model
195class CustomFeeModel(FeeModel):
196def GetOrderFee(self, parameters):
197fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
198return OrderFee(CashAmount(fee, "USD"))