-
Notifications
You must be signed in to change notification settings - Fork 500
/
momentum-and-reversal-combined-with-volatility-effect-in-stocks.py
205 lines (161 loc) · 8.69 KB
/
momentum-and-reversal-combined-with-volatility-effect-in-stocks.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
# https://quantpedia.com/strategies/momentum-and-reversal-combined-with-volatility-effect-in-stocks/
#
# The investment universe consists of NYSE, AMEX, and NASDAQ stocks with prices higher than $5 per share. At the beginning of each month,
# the sample is divided into equal halves, at the size median, and only larger stocks are used. Then each month, realized returns and realized
# (annualized) volatilities are calculated for each stock for the past six months. One week (seven calendar days) prior to the beginning of
# each month is skipped to avoid biases due to microstructures. Stocks are then sorted into quintiles based on their realized past returns
# and past volatility. The investor goes long on stocks from the highest performing quintile from the highest volatility group and short on
# stocks from the lowest-performing quintile from the highest volatility group. Stocks are equally weighted and held for six months
# (therefore, 1/6 of the portfolio is rebalanced every month).
#
# QC implementation changes:
# - Instead of all listed stock, we select 1000 most liquid stocks from QC filtered stock universe (~8000 stocks) due to time complexity issues tied to whole universe filtering.
import numpy as np
from AlgorithmImports import *
class MomentumReversalCombinedWithVolatilityEffectinStocks(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2002, 1, 1)
self.SetCash(100000)
self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
# EW Tranching.
self.holding_period = 6
self.managed_queue = []
# Daily price data.
self.data = {}
self.period = 6 * 21
self.coarse_count = 1000
self.selection_flag = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(10)
def CoarseSelectionFunction(self, coarse):
# Update the rolling window every day.
for stock in coarse:
symbol = stock.Symbol
# Store monthly price.
if symbol in self.data:
self.data[symbol].update(stock.AdjustedPrice)
self.data[symbol].LastPrice = stock.AdjustedPrice
if not self.selection_flag:
return Universe.Unchanged
# selected = [x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5]
selected = sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5], \
key = lambda x: x.DollarVolume, reverse = True)[:self.coarse_count]
# Warmup price rolling windows.
for stock in selected:
symbol = stock.Symbol
if symbol in self.data:
continue
self.data[symbol] = SymbolData(symbol, self.period)
history = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes = history.loc[symbol].close
for time, close in closes.iteritems():
self.data[symbol].update(close)
self.data[symbol].LastPrice = close
return [x.Symbol for x in selected if self.data[x.Symbol].is_ready()]
def FineSelectionFunction(self, fine):
fine = [x for x in fine if x.MarketCap != 0 and \
((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE"))]
# if len(fine) > self.coarse_count:
# sorted_by_market_cap = sorted(fine, key = lambda x: x.MarketCap, reverse=True)
# top_by_market_cap = sorted_by_market_cap[:self.coarse_count]
# else:
# top_by_market_cap = fine
sorted_by_market_cap = sorted(fine, key = lambda x: x.MarketCap, reverse=True)
half = int(len(sorted_by_market_cap) / 2)
top_by_market_cap = [x.Symbol for x in sorted_by_market_cap][:half]
# Performance and volatility tuple.
perf_volatility = {}
for symbol in top_by_market_cap:
performance = self.data[symbol].performance()
annualized_volatility = self.data[symbol].volatility()
perf_volatility[symbol] = (performance, annualized_volatility)
long = []
short = []
if len(perf_volatility) >= 5:
sorted_by_perf = sorted(perf_volatility.items(), key = lambda x: x[1][0], reverse = True)
quintile = int(len(sorted_by_perf) / 5)
top_by_perf = [x[0] for x in sorted_by_perf[:quintile]]
low_by_perf = [x[0] for x in sorted_by_perf[-quintile:]]
sorted_by_vol = sorted(perf_volatility.items(), key = lambda x: x[1][1], reverse = True)
quintile = int(len(sorted_by_vol) / 5)
top_by_vol = [x[0] for x in sorted_by_vol[:quintile]]
low_by_vol = [x[0] for x in sorted_by_vol[-quintile:]]
long = [x for x in top_by_perf if x in top_by_vol]
short = [x for x in low_by_perf if x in top_by_vol]
if len(long) != 0:
long_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
# symbol/quantity collection
long_symbol_q = [(x, np.ceil(long_w / self.data[x].LastPrice)) for x in long]
else:
long_symbol_q = []
if len(short) != 0:
short_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
# symbol/quantity collection
short_symbol_q = [(x, -np.ceil(short_w / self.data[x].LastPrice)) for x in short]
else:
short_symbol_q = []
self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
return long + short
def OnData(self, data):
if not self.selection_flag:
return
self.selection_flag = False
remove_item = None
# Rebalance portfolio
for item in self.managed_queue:
if item.holding_period == self.holding_period:
for symbol, quantity in item.symbol_q:
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
self.MarketOrder(symbol, -quantity)
remove_item = item
# Trade execution
if item.holding_period == 0:
open_symbol_q = []
for symbol, quantity in item.symbol_q:
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
self.MarketOrder(symbol, quantity)
open_symbol_q.append((symbol, quantity))
# Only opened orders will be closed
item.symbol_q = open_symbol_q
item.holding_period += 1
# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
if remove_item:
self.managed_queue.remove(remove_item)
def Selection(self):
self.selection_flag = True
class RebalanceQueueItem():
def __init__(self, symbol_q):
# symbol/quantity collections
self.symbol_q = symbol_q
self.holding_period = 0
class SymbolData():
def __init__(self, symbol, period):
self.Symbol = symbol
self.Price = RollingWindow[float](period)
self.LastPrice = 0
def update(self, value):
self.Price.Add(value)
def is_ready(self):
return self.Price.IsReady
def update(self, close):
self.Price.Add(close)
def volatility(self):
closes = np.array([x for x in self.Price][5:]) # Skip last week.
daily_returns = closes[:-1] / closes[1:] - 1
return np.std(daily_returns) * np.sqrt(252 / (len(closes)))
def performance(self):
closes = [x for x in self.Price][5:] # Skip last week.
return (closes[0] / closes[-1] - 1)
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))