forked from QuantConnect/Lean
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathDataConsolidationAlgorithm.py
More file actions
189 lines (149 loc) · 9.78 KB
/
DataConsolidationAlgorithm.py
File metadata and controls
189 lines (149 loc) · 9.78 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
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
# 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("System")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Common")
from System import *
from QuantConnect import *
from QuantConnect.Algorithm import *
from QuantConnect.Data.Consolidators import *
from datetime import timedelta
### <summary>
### Example algorithm giving an introduction into using IDataConsolidators.
### This is an advanced QC concept and requires a certain level of comfort using C# and its event system.
###
### What is an IDataConsolidator?
### IDataConsolidator is a plugin point that can be used to transform your data more easily.
### In this example we show one of the simplest consolidators, the TradeBarConsolidator.
### This type is capable of taking a timespan to indicate how long each bar should be, or an
### integer to indicate how many bars should be aggregated into one.
###
### When a new 'consolidated' piece of data is produced by the IDataConsolidator, an event is fired
### with the argument of the new data.
### </summary>
### <meta name="tag" content="using data" />
### <meta name="tag" content="consolidating data" />
class DataConsolidationAlgorithm(QCAlgorithm):
def Initialize(self):
'''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.'''
self.SetStartDate(DateTime(2013, 10, 7, 9, 30, 0)) #Set Start Date
self.SetEndDate(self.StartDate + timedelta(60)) #Set End Date
# Find more symbols here: http://quantconnect.com/data
self.AddEquity("SPY")
self.AddForex("EURUSD", Resolution.Hour)
# define our 30 minute trade bar consolidator. we can
# access the 30 minute bar from the DataConsolidated events
thirtyMinuteConsolidator = TradeBarConsolidator(timedelta(minutes=30))
# attach our event handler. the event handler is a function that will
# be called each time we produce a new consolidated piece of data.
thirtyMinuteConsolidator.DataConsolidated += self.ThirtyMinuteBarHandler
# this call adds our 30 minute consolidator to
# the manager to receive updates from the engine
self.SubscriptionManager.AddConsolidator("SPY", thirtyMinuteConsolidator)
# here we'll define a slightly more complex consolidator. what we're trying to produce is
# a 3 day bar. Now we could just use a single TradeBarConsolidator like above and pass in
# TimeSpan.FromDays(3), but in reality that's not what we want. For time spans of longer than
# a day we'll get incorrect results around weekends and such. What we really want are tradeable
# days. So we'll create a daily consolidator, and then wrap it with a 3 count consolidator.
# first define a one day trade bar -- this produces a consolidated piece of data after a day has passed
oneDayConsolidator = TradeBarConsolidator(timedelta(1))
# next define our 3 count trade bar -- this produces a consolidated piece of data after it sees 3 pieces of data
threeCountConsolidator = TradeBarConsolidator(3)
# here we combine them to make a new, 3 day trade bar. The SequentialConsolidator allows composition of
# consolidators. It takes the consolidated output of one consolidator (in this case, the oneDayConsolidator)
# and pipes it through to the threeCountConsolidator. His output will be a 3 day bar.
three_oneDayBar = SequentialConsolidator(oneDayConsolidator, threeCountConsolidator)
# attach our handler
three_oneDayBar.DataConsolidated += self.ThreeDayBarConsolidatedHandler
# this call adds our 3 day to the manager to receive updates from the engine
self.SubscriptionManager.AddConsolidator("SPY", three_oneDayBar)
# Custom monthly consolidator
customMonthlyConsolidator = TradeBarConsolidator(self.CustomMonthly)
customMonthlyConsolidator.DataConsolidated += self.CustomMonthlyHandler
self.SubscriptionManager.AddConsolidator("SPY", customMonthlyConsolidator)
# API convenience method for easily receiving consolidated data
self.Consolidate("SPY", timedelta(minutes=45), self.FortyFiveMinuteBarHandler)
self.Consolidate("SPY", Resolution.Hour, self.HourBarHandler)
self.Consolidate("EURUSD", Resolution.Daily, self.DailyEurUsdBarHandler)
# API convenience method for easily receiving weekly-consolidated data
self.Consolidate("SPY", Calendar.Weekly, self.CalendarTradeBarHandler)
self.Consolidate("EURUSD", Calendar.Weekly, self.CalendarQuoteBarHandler)
# API convenience method for easily receiving monthly-consolidated data
self.Consolidate("SPY", Calendar.Monthly, self.CalendarTradeBarHandler);
self.Consolidate("EURUSD", Calendar.Monthly, self.CalendarQuoteBarHandler);
# API convenience method for easily receiving quarterly-consolidated data
self.Consolidate("SPY", Calendar.Quarterly, self.CalendarTradeBarHandler)
self.Consolidate("EURUSD", Calendar.Quarterly, self.CalendarQuoteBarHandler)
# API convenience method for easily receiving yearly-consolidated data
self.Consolidate("SPY", Calendar.Yearly, self.CalendarTradeBarHandler);
self.Consolidate("EURUSD", Calendar.Yearly, self.CalendarQuoteBarHandler);
# some securities may have trade and quote data available, so we can choose it based on TickType:
#self.Consolidate("BTCUSD", Resolution.Hour, TickType.Trade, self.HourBarHandler) # to get TradeBar
#self.Consolidate("BTCUSD", Resolution.Hour, TickType.Quote, self.HourBarHandler) # to get QuoteBar (default)
self.consolidatedHour = False
self.consolidated45Minute = False
self.__last = None
def OnData(self, data):
'''We need to declare this method'''
pass
def OnEndOfDay(self):
# close up shop each day and reset our 'last' value so we start tomorrow fresh
self.Liquidate("SPY")
self.__last = None
def ThirtyMinuteBarHandler(self, sender, consolidated):
'''This is our event handler for our 30 minute trade bar defined above in Initialize(). So each time the
consolidator produces a new 30 minute bar, this function will be called automatically. The 'sender' parameter
will be the instance of the IDataConsolidator that invoked the event, but you'll almost never need that!'''
if self.__last is not None and consolidated.Close > self.__last.Close:
self.Log(f"{consolidated.Time} >> SPY >> LONG >> 100 >> {self.Portfolio['SPY'].Quantity}")
self.Order("SPY", 100)
elif self.__last is not None and consolidated.Close < self.__last.Close:
self.Log(f"{consolidated.Time} >> SPY >> SHORT >> 100 >> {self.Portfolio['SPY'].Quantity}")
self.Order("SPY", -100)
self.__last = consolidated
def ThreeDayBarConsolidatedHandler(self, sender, consolidated):
''' This is our event handler for our 3 day trade bar defined above in Initialize(). So each time the
consolidator produces a new 3 day bar, this function will be called automatically. The 'sender' parameter
will be the instance of the IDataConsolidator that invoked the event, but you'll almost never need that!'''
self.Log(f"{consolidated.Time} >> Plotting!")
self.Plot(consolidated.Symbol.Value, "3HourBar", consolidated.Close)
def FortyFiveMinuteBarHandler(self, consolidated):
''' This is our event handler for our 45 minute consolidated defined using the Consolidate method'''
self.consolidated45Minute = True
self.Log(f"{consolidated.EndTime} >> FortyFiveMinuteBarHandler >> {consolidated.Close}")
def HourBarHandler(self, consolidated):
'''This is our event handler for our one hour consolidated defined using the Consolidate method'''
self.consolidatedHour = True
self.Log(f"{consolidated.EndTime} >> FortyFiveMinuteBarHandler >> {consolidated.Close}")
def DailyEurUsdBarHandler(self, consolidated):
'''This is our event handler for our daily consolidated defined using the Consolidate method'''
self.Log(f"{consolidated.EndTime} EURUSD Daily consolidated.")
def CalendarTradeBarHandler(self, tradeBar):
self.Log(f'{self.Time} :: {tradeBar.Time} {tradeBar.Close}')
def CalendarQuoteBarHandler(self, quoteBar):
self.Log(f'{self.Time} :: {quoteBar.Time} {quoteBar.Close}')
def CustomMonthly(self, dt):
'''Custom Monthly Func'''
start = dt.replace(day=1).date()
end = dt.replace(day=28) + timedelta(4)
end = (end - timedelta(end.day-1)).date()
return CalendarInfo(start, end - start)
def CustomMonthlyHandler(self, sender, consolidated):
'''This is our event handler Custom Monthly function'''
self.Log(f"{consolidated.Time} >> CustomMonthlyHandler >> {consolidated.Close}")
def OnEndOfAlgorithm(self):
if not self.consolidatedHour:
raise Exception("Expected hourly consolidator to be fired.")
if not self.consolidated45Minute:
raise Exception("Expected 45-minute consolidator to be fired.")