Realized Skewness Predicts Equity Returns

1from AlgorithmImports import *
2from collections import deque
3from scipy.stats import skew
4import numpy as np
5from pandas.core.frame import DataFrame
6#endregion
7class RealizedSkewnessPredictsEquityReturns(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2010, 1, 1)
11self.SetCash(100000)
12market:Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
13# 5-minute price data.
14self.data:Dict[Symbol, deque] = {}
15self.period:int = 5 * 78
16self.rebalance_month:int = 12
17self.leverage:int = 5
18self.quantile:int = 10
19self.min_share_price:float = 5.
20
21self.fundamental_count:int = 100
22self.fundamental_sorting_key = lambda x: x.MarketCap
23# Yearly selected universe with symbol and market cap data.
24self.selected_universe:List[Fundamental] = []
25
26self.month:int = 12
27self.days:int = 5
28self.selection_flag:bool = False
29self.UniverseSettings.Resolution = Resolution.Minute
30self.AddUniverse(self.FundamentalSelectionFunction)
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
32self.settings.daily_precise_end_time = False
33self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
34
35def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
36for security in changes.AddedSecurities:
37    symbol = security.Symbol
38    security.SetFeeModel(CustomFeeModel())
39    security.SetLeverage(self.leverage)
40    
41    if symbol not in self.data:
42        history:DataFrame = self.History(symbol, self.period * 5, Resolution.Minute)
43        if len(history) == self.period and 'close' in history:
44            closes_1M:List[float] = list(history['close'])
45            closes_5M:List[float] = closes_1M[::5]
46            self.data[symbol] = deque(closes_5M, maxlen = self.period)
47
48# Remove old stocks from selected universe data.
49for security in changes.RemovedSecurities:
50    if security.Symbol in self.data:
51        del self.data[security.Symbol]
52def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
53if not self.selection_flag: 
54    return Universe.Unchanged
55self.selection_flag = False
56selected:List[Fundamental] = [
57    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price >= self.min_share_price
58]
59if len(selected) > self.fundamental_count:
60    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
61
62self.selected_universe = selected
63
64return list(map(lambda x: x.Symbol, selected))
65
66def OnData(self, data: Slice) -> None:
67if self.Time.minute % 5 == 0:
68    # Store 5 minute data.
69    for symbol in self.data:
70        if symbol in data and data[symbol]:
71            price = data[symbol].Value
72            self.data[symbol].append(price)
73if not (self.Time.hour == 16 and self.Time.minute == 0):
74    return
75if self.days == 5:
76    aggregate_skewness:Dict[Symbol, float] = {}
77    for stock in self.selected_universe:
78        symbol:Symbol = stock.Symbol
79        # 5 Minute data is ready.
80        if symbol in self.data and len(self.data[symbol]) == self.data[symbol].maxlen:
81            closes_5M:np.ndarray = np.array(self.data[symbol])
82            returns_5M:np.ndarray = (closes_5M[1:] - closes_5M[:-1]) / closes_5M[:-1]
83            skewness:float = skew(returns_5M)
84            aggregate_skewness[stock] = skewness
85                    
86    if len(aggregate_skewness) != 0:
87        # Aggregate skewness sorting.
88        sorted_by_aggregate_skewness:List = sorted(aggregate_skewness.items(), key = lambda x: x[1], reverse = True)
89        quantile:int = int(len(sorted_by_aggregate_skewness) / self.quantile)
90        long:List[Fundamental] = [x[0] for x in sorted_by_aggregate_skewness[-quantile:]]
91        short:List[Fundamental] = [x[0] for x in sorted_by_aggregate_skewness[:quantile]]
92    
93        weight:Dict[Symbol, float] = {}
94        # Market cap weighting.
95        for i, portfolio in enumerate([long, short]):
96            mc_sum:float = sum(map(lambda x: x.MarketCap, portfolio))
97            for stock in portfolio:
98                weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
99                
100        # Trade execution.
101        portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in weight.items() if symbol in data and data[symbol]]
102        self.SetHoldings(portfolio, True)
103
104self.days += 1
105if self.days > 5:
106    self.days = 1      
107
108def Selection(self) -> None:
109if self.month == self.rebalance_month:
110    self.selection_flag = True
111
112self.month += 1
113if self.month > 12:
114    self.month = 1
115# Custom fee model
116class CustomFeeModel(FeeModel):
117def GetOrderFee(self, parameters):
118fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
119return OrderFee(CashAmount(fee, "USD"))