forked from 0xboz/zipline_bundle
-
Notifications
You must be signed in to change notification settings - Fork 0
/
calendar_utils.py
326 lines (283 loc) · 10.7 KB
/
calendar_utils.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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
from .always_open import AlwaysOpenCalendar
from .errors import (
CalendarNameCollision,
CyclicCalendarAlias,
InvalidCalendarName,
)
from .exchange_calendar_bvmf import BVMFExchangeCalendar
from .exchange_calendar_cmes import CMESExchangeCalendar
from .exchange_calendar_iepa import IEPAExchangeCalendar
from .exchange_calendar_xams import XAMSExchangeCalendar
from .exchange_calendar_xasx import XASXExchangeCalendar
from .exchange_calendar_xbom import XBOMExchangeCalendar
from .exchange_calendar_xbru import XBRUExchangeCalendar
from .exchange_calendar_xcbf import XCBFExchangeCalendar
from .exchange_calendar_xcse import XCSEExchangeCalendar
from .exchange_calendar_xdub import XDUBExchangeCalendar
from .exchange_calendar_xfra import XFRAExchangeCalendar
from .exchange_calendar_xhel import XHELExchangeCalendar
from .exchange_calendar_xhkg import XHKGExchangeCalendar
from .exchange_calendar_xkrx import XKRXExchangeCalendar
from .exchange_calendar_xlis import XLISExchangeCalendar
from .exchange_calendar_xlon import XLONExchangeCalendar
from .exchange_calendar_xmad import XMADExchangeCalendar
from .exchange_calendar_xmil import XMILExchangeCalendar
from .exchange_calendar_xnys import XNYSExchangeCalendar
from .exchange_calendar_xnze import XNZEExchangeCalendar
from .exchange_calendar_xosl import XOSLExchangeCalendar
from .exchange_calendar_xpar import XPARExchangeCalendar
from .exchange_calendar_xses import XSESExchangeCalendar
from .exchange_calendar_xshg import XSHGExchangeCalendar
from .exchange_calendar_xsto import XSTOExchangeCalendar
from .exchange_calendar_xswx import XSWXExchangeCalendar
from .exchange_calendar_xtks import XTKSExchangeCalendar
from .exchange_calendar_xtse import XTSEExchangeCalendar
from .exchange_calendar_xwbo import XWBOExchangeCalendar
from .us_futures_calendar import QuantopianUSFuturesCalendar
from .weekday_calendar import WeekdayCalendar
from .exchange_calendar_binance import BinanceExchangeCalendar
_default_calendar_factories = {
# Exchange calendars.
'BVMF': BVMFExchangeCalendar,
'CMES': CMESExchangeCalendar,
'IEPA': IEPAExchangeCalendar,
'XAMS': XAMSExchangeCalendar,
'XASX': XASXExchangeCalendar,
'XBOM': XBOMExchangeCalendar,
'XBRU': XBRUExchangeCalendar,
'XCBF': XCBFExchangeCalendar,
'XCSE': XCSEExchangeCalendar,
'XDUB': XDUBExchangeCalendar,
'XFRA': XFRAExchangeCalendar,
'XHEL': XHELExchangeCalendar,
'XHKG': XHKGExchangeCalendar,
'XKRX': XKRXExchangeCalendar,
'XLIS': XLISExchangeCalendar,
'XLON': XLONExchangeCalendar,
'XMAD': XMADExchangeCalendar,
'XMIL': XMILExchangeCalendar,
'XNYS': XNYSExchangeCalendar,
'XNZE': XNZEExchangeCalendar,
'XOSL': XOSLExchangeCalendar,
'XPAR': XPARExchangeCalendar,
'XSES': XSESExchangeCalendar,
'XSHG': XSHGExchangeCalendar,
'XSTO': XSTOExchangeCalendar,
'XSWX': XSWXExchangeCalendar,
'XTKS': XTKSExchangeCalendar,
'XTSE': XTSEExchangeCalendar,
'XWBO': XWBOExchangeCalendar,
# Miscellaneous calendars.
'BINANCE': BinanceExchangeCalendar,
'us_futures': QuantopianUSFuturesCalendar,
'24/7': AlwaysOpenCalendar,
'24/5': WeekdayCalendar,
}
_default_calendar_aliases = {
'NYSE': 'XNYS',
'NASDAQ': 'XNYS',
'BATS': 'XNYS',
'FWB': 'XFRA',
'LSE': 'XLON',
'TSX': 'XTSE',
'BMF': 'BVMF',
'CME': 'CMES',
'CBOT': 'CMES',
'COMEX': 'CMES',
'NYMEX': 'CMES',
'ICE': 'IEPA',
'ICEUS': 'IEPA',
'NYFE': 'IEPA',
'CFE': 'XCBF',
'Binance': 'BINANCE',
}
default_calendar_names = sorted(_default_calendar_factories.keys())
class TradingCalendarDispatcher(object):
"""
A class for dispatching and caching trading calendars.
Methods of a global instance of this class are provided by
calendars.calendar_utils.
Parameters
----------
calendars : dict[str -> TradingCalendar]
Initial set of calendars.
calendar_factories : dict[str -> function]
Factories for lazy calendar creation.
aliases : dict[str -> str]
Calendar name aliases.
"""
def __init__(self, calendars, calendar_factories, aliases):
self._calendars = calendars
self._calendar_factories = dict(calendar_factories)
self._aliases = dict(aliases)
def get_calendar(self, name):
"""
Retrieves an instance of an TradingCalendar whose name is given.
Parameters
----------
name : str
The name of the TradingCalendar to be retrieved.
Returns
-------
calendar : calendars.TradingCalendar
The desired calendar.
"""
canonical_name = self.resolve_alias(name)
try:
return self._calendars[canonical_name]
except KeyError:
# We haven't loaded this calendar yet, so make a new one.
pass
try:
factory = self._calendar_factories[canonical_name]
except KeyError:
# We don't have a factory registered for this name. Barf.
raise InvalidCalendarName(calendar_name=name)
# Cache the calendar for future use.
calendar = self._calendars[canonical_name] = factory()
return calendar
def has_calendar(self, name):
"""
Do we have (or have the ability to make) a calendar with ``name``?
"""
return (
name in self._calendars
or name in self._calendar_factories
or name in self._aliases
)
def register_calendar(self, name, calendar, force=False):
"""
Registers a calendar for retrieval by the get_calendar method.
Parameters
----------
name: str
The key with which to register this calendar.
calendar: TradingCalendar
The calendar to be registered for retrieval.
force : bool, optional
If True, old calendars will be overwritten on a name collision.
If False, name collisions will raise an exception.
Default is False.
Raises
------
CalendarNameCollision
If a calendar is already registered with the given calendar's name.
"""
if force:
self.deregister_calendar(name)
if self.has_calendar(name):
raise CalendarNameCollision(calendar_name=name)
self._calendars[name] = calendar
def register_calendar_type(self, name, calendar_type, force=False):
"""
Registers a calendar by type.
This is useful for registering a new calendar to be lazily instantiated
at some future point in time.
Parameters
----------
name: str
The key with which to register this calendar.
calendar_type: type
The type of the calendar to register.
force : bool, optional
If True, old calendars will be overwritten on a name collision.
If False, name collisions will raise an exception.
Default is False.
Raises
------
CalendarNameCollision
If a calendar is already registered with the given calendar's name.
"""
if force:
self.deregister_calendar(name)
if self.has_calendar(name):
raise CalendarNameCollision(calendar_name=name)
self._calendar_factories[name] = calendar_type
def register_calendar_alias(self, alias, real_name, force=False):
"""
Register an alias for a calendar.
This is useful when multiple exchanges should share a calendar, or when
there are multiple ways to refer to the same exchange.
After calling ``register_alias('alias', 'real_name')``, subsequent
calls to ``get_calendar('alias')`` will return the same result as
``get_calendar('real_name')``.
Parameters
----------
alias : str
The name to be used to refer to a calendar.
real_name : str
The canonical name of the registered calendar.
force : bool, optional
If True, old calendars will be overwritten on a name collision.
If False, name collisions will raise an exception.
Default is False.
"""
if force:
self.deregister_calendar(alias)
if self.has_calendar(alias):
raise CalendarNameCollision(calendar_name=alias)
self._aliases[alias] = real_name
# Ensure that the new alias doesn't create a cycle, and back it out if
# we did.
try:
self.resolve_alias(alias)
except CyclicCalendarAlias:
del self._aliases[alias]
raise
def resolve_alias(self, name):
"""
Resolve a calendar alias for retrieval.
Parameters
----------
name : str
The name of the requested calendar.
Returns
-------
canonical_name : str
The real name of the calendar to create/return.
"""
seen = []
while name in self._aliases:
seen.append(name)
name = self._aliases[name]
# This is O(N ** 2), but if there's an alias chain longer than 2,
# something strange has happened.
if name in seen:
seen.append(name)
raise CyclicCalendarAlias(
cycle=" -> ".join(repr(k) for k in seen)
)
return name
def deregister_calendar(self, name):
"""
If a calendar is registered with the given name, it is de-registered.
Parameters
----------
cal_name : str
The name of the calendar to be deregistered.
"""
self._calendars.pop(name, None)
self._calendar_factories.pop(name, None)
self._aliases.pop(name, None)
def clear_calendars(self):
"""
Deregisters all current registered calendars
"""
self._calendars.clear()
self._calendar_factories.clear()
self._aliases.clear()
# We maintain a global calendar dispatcher so that users can just do
# `register_calendar('my_calendar', calendar) and then use `get_calendar`
# without having to thread around a dispatcher.
global_calendar_dispatcher = TradingCalendarDispatcher(
calendars={},
calendar_factories=_default_calendar_factories,
aliases=_default_calendar_aliases,
)
get_calendar = global_calendar_dispatcher.get_calendar
clear_calendars = global_calendar_dispatcher.clear_calendars
deregister_calendar = global_calendar_dispatcher.deregister_calendar
register_calendar = global_calendar_dispatcher.register_calendar
register_calendar_type = global_calendar_dispatcher.register_calendar_type
register_calendar_alias = global_calendar_dispatcher.register_calendar_alias
resolve_alias = global_calendar_dispatcher.resolve_alias