forked from QuantConnect/Lean
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathRsiAlphaModel.py
More file actions
140 lines (113 loc) · 5.55 KB
/
Copy pathRsiAlphaModel.py
File metadata and controls
140 lines (113 loc) · 5.55 KB
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
# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
# Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from clr import AddReference
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
AddReference("QuantConnect.Indicators")
AddReference("QuantConnect.Logging")
AddReference("QuantConnect.Common")
from QuantConnect import *
from QuantConnect.Indicators import *
from QuantConnect.Logging import Log
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Alphas import *
from datetime import timedelta
from enum import Enum
class RsiAlphaModel(AlphaModel):
'''Uses Wilder's RSI to create insights.
Using default settings, a cross over below 30 or above 70 will trigger a new insight.'''
def __init__(self,
period = 14,
resolution = Resolution.Daily):
'''Initializes a new instance of the RsiAlphaModel class
Args:
period: The RSI indicator period'''
self.period = period
self.resolution = resolution
self.insightPeriod = Time.Multiply(Extensions.ToTimeSpan(resolution), period)
self.symbolDataBySymbol ={}
resolutionString = Extensions.GetEnumString(resolution, Resolution)
self.Name = '{}({},{})'.format(self.__class__.__name__, period, resolutionString)
def Update(self, algorithm, data):
'''Updates this alpha model with the latest data from the algorithm.
This is called each time the algorithm receives data for subscribed securities
Args:
algorithm: The algorithm instance
data: The new data available
Returns:
The new insights generated'''
insights = []
for symbol, symbolData in self.symbolDataBySymbol.items():
rsi = symbolData.RSI
previous_state = symbolData.State
state = self.GetState(rsi, previous_state)
if state != previous_state and rsi.IsReady:
if state == State.TrippedLow:
insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Up))
if state == State.TrippedHigh:
insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Down))
symbolData.State = state
return insights
def OnSecuritiesChanged(self, algorithm, changes):
'''Cleans out old security data and initializes the RSI for any newly added securities.
Event fired each time the we add/remove securities from the data feed
Args:
algorithm: The algorithm instance that experienced the change in securities
changes: The security additions and removals from the algorithm'''
# clean up data for removed securities
symbols = [ x.Symbol for x in changes.RemovedSecurities ]
if len(symbols) > 0:
for subscription in algorithm.SubscriptionManager.Subscriptions:
if subscription.Symbol in symbols:
self.symbolDataBySymbol.pop(subscription.Symbol, None)
subscription.Consolidators.Clear()
# initialize data for added securities
addedSymbols = [ x.Symbol for x in changes.AddedSecurities if x.Symbol not in self.symbolDataBySymbol]
if len(addedSymbols) == 0: return
history = algorithm.History(addedSymbols, self.period, self.resolution)
for symbol in addedSymbols:
rsi = algorithm.RSI(symbol, self.period, MovingAverageType.Wilders, self.resolution)
if not history.empty:
ticker = SymbolCache.GetTicker(symbol)
if ticker not in history.index.levels[0]:
Log.Trace(f'RsiAlphaModel.OnSecuritiesChanged: {ticker} not found in history data frame.')
continue
for tuple in history.loc[ticker].itertuples():
rsi.Update(tuple.Index, tuple.close)
self.symbolDataBySymbol[symbol] = SymbolData(symbol, rsi)
def GetState(self, rsi, previous):
''' Determines the new state. This is basically cross-over detection logic that
includes considerations for bouncing using the configured bounce tolerance.'''
if rsi.Current.Value > 70:
return State.TrippedHigh
if rsi.Current.Value < 30:
return State.TrippedLow
if previous == State.TrippedLow:
if rsi.Current.Value > 35:
return State.Middle
if previous == State.TrippedHigh:
if rsi.Current.Value < 65:
return State.Middle
return previous
class SymbolData:
'''Contains data specific to a symbol required by this model'''
def __init__(self, symbol, rsi):
self.Symbol = symbol
self.RSI = rsi
self.State = State.Middle
class State(Enum):
'''Defines the state. This is used to prevent signal spamming and aid in bounce detection.'''
TrippedLow = 0
Middle = 1
TrippedHigh = 2