Expected Skewness and Momentum in Stocks

1from AlgorithmImports import *
2import numpy as np
3#endregion
4class ExpectedSkewnessMomentum(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2010, 1, 1)
7self.SetCash(100000)
8market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
9
10self.fundamental_count:int = 1000
11self.fundamental_sorting_key = lambda x: x.DollarVolume
12
13# Monthly close data.
14self.data:Dict[Symbol, SymbolData] = {}
15self.period:int = 12 * 21
16self.min_share_price:float = 1.
17self.quantile:int = 10
18self.leverage:int = 10
19self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
20
21self.long:List[Symbol] = []
22self.short:List[Symbol] = []
23self.selection_flag:bool = False
24self.UniverseSettings.Resolution = Resolution.Daily
25self.AddUniverse(self.FundamentalSelectionFunction)
26self.settings.daily_precise_end_time = False
27self.settings.minimum_order_margin_portfolio_percentage = 0.
28self.schedule.on(self.date_rules.month_start(market),
29                self.time_rules.after_market_open(market),
30                self.selection)
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34    security.SetLeverage(self.leverage)
35    
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
37# Update the rolling window every day.
38for stock in fundamental:
39    symbol:Symbol = stock.Symbol
40    # Store monthly price.
41    if symbol in self.data:
42        self.data[symbol].update(stock.AdjustedPrice)
43if not self.selection_flag:
44    return Universe.Unchanged
45
46selected:List[Fundamental] = [
47    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchange_codes and \
48    x.Price > self.min_share_price and x.MarketCap != 0
49]
50
51if len(selected) > self.fundamental_count:
52    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
53    
54performance_max_return:Dict[Symbol, Tuple[float, float]] = {}
55# Warmup price rolling windows.
56for stock in selected:
57    symbol:Symbol = stock.Symbol
58    if symbol not in self.data:
59        self.data[symbol] = SymbolData(self.period)
60        history = self.History(symbol, self.period, Resolution.Daily)
61        if history.empty:
62            self.Log(f"Not enough data for {symbol} yet")
63            continue
64        closes = history.loc[symbol].close
65        for time, close in closes.items():
66            self.data[symbol].update(close)
67    
68    if self.data[symbol].is_ready():
69        performance_max_return[symbol] = (self.data[symbol].performance(), self.data[symbol].max_performance_last_month())
70
71if len(performance_max_return) >= self.quantile * 2:
72    sorted_by_max_perf:List = sorted(performance_max_return.items(), key = lambda x: x[1][1], reverse = True)
73    quantile:int = int(len(sorted_by_max_perf) / self.quantile)
74    high_by_daily_perf:List = [x for x in sorted_by_max_perf[:quantile]]
75    low_by_daily_perf:List = [x for x in sorted_by_max_perf[-quantile:]]
76
77    # Most negatively skewed winners.
78    sorted_by_performance:List = sorted(low_by_daily_perf, key = lambda x: x[1][0], reverse = True)
79    quantile = int(len(sorted_by_performance) / self.quantile)
80    self.long = [x[0] for x in sorted_by_performance[:quantile]]
81    # Most positively skewed losers.
82    sorted_by_performance:List = sorted(high_by_daily_perf, key = lambda x: x[1][0], reverse = True)
83    quantile = int(len(sorted_by_performance) / self.quantile)
84    self.short = [x[0] for x in sorted_by_performance[-quantile:]]
85
86return self.long + self.short
87
88def OnData(self, data: Slice) -> None:
89if not self.selection_flag:
90    return
91self.selection_flag = False
92
93# order execution
94targets:List[PortfolioTarget] = []
95for i, portfolio in enumerate([self.long, self.short]):
96    for symbol in portfolio:
97        if symbol in data and data[symbol]:
98            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
99		
100self.SetHoldings(targets, True)
101self.long.clear()
102self.short.clear()
103def selection(self) -> None:
104self.selection_flag = True
105class SymbolData():
106def __init__(self, period: int):
107self._price:RollingWindow = RollingWindow[float](period)
108
109def update(self, price: float) -> None:
110self._price.Add(price)
111
112def is_ready(self) -> bool:
113return self._price.IsReady
114
115# Yearly performance, one month skipped.
116def performance(self) -> float:
117closes:List[float] = list(self._price)[21:]
118return (closes[0] / closes[-1] - 1)
119
120def max_performance_last_month(self) -> float:
121closes:np.ndarray = np.array(list(self._price)[:21])
122daily_returns:np.ndarray = closes[:-1] / closes[1:] - 1
123return max(daily_returns)
124
125# Custom fee model.
126class CustomFeeModel(FeeModel):
127def GetOrderFee(self, parameters):
128fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
129return OrderFee(CashAmount(fee, "USD"))