-
Notifications
You must be signed in to change notification settings - Fork 345
Expand file tree
/
Copy pathsnapshot.py
More file actions
561 lines (493 loc) · 18.1 KB
/
snapshot.py
File metadata and controls
561 lines (493 loc) · 18.1 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
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
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
from typing import Optional, List, Dict
from .aggs import Agg
from .quotes import LastQuote
from .trades import LastTrade
from ...modelclass import modelclass
@modelclass
class MinuteSnapshot:
"""Most recent minute bar."""
accumulated_volume: Optional[float] = None
open: Optional[float] = None
high: Optional[float] = None
low: Optional[float] = None
close: Optional[float] = None
volume: Optional[float] = None
vwap: Optional[float] = None
otc: Optional[bool] = None
timestamp: Optional[int] = None
transactions: Optional[int] = None
fractional_volume: Optional[str] = None
fractional_accumulated_volume: Optional[str] = None
@staticmethod
def from_dict(d):
return MinuteSnapshot(
accumulated_volume=d.get("av"),
open=d.get("o"),
high=d.get("h"),
low=d.get("l"),
close=d.get("c"),
volume=d.get("v"),
vwap=d.get("vw"),
otc=d.get("otc"),
timestamp=d.get("t"),
transactions=d.get("n"),
fractional_volume=d.get("dv"),
fractional_accumulated_volume=d.get("dav"),
)
@modelclass
class IndicesSession:
"Contains data for the most recent daily bar in an options contract."
change: Optional[float] = None
change_percent: Optional[float] = None
close: Optional[float] = None
high: Optional[float] = None
low: Optional[float] = None
open: Optional[float] = None
previous_close: Optional[float] = None
@staticmethod
def from_dict(d):
return IndicesSession(**d)
@modelclass
class IndicesSnapshot:
value: Optional[float] = None
name: Optional[str] = None
type: Optional[str] = None
ticker: Optional[str] = None
market_status: Optional[str] = None
session: Optional[IndicesSession] = None
error: Optional[str] = None
message: Optional[str] = None
@staticmethod
def from_dict(d):
return IndicesSnapshot(
value=d.get("value", None),
name=d.get("name", None),
type=d.get("type", None),
ticker=d.get("ticker", None),
market_status=d.get("market_status", None),
session=(
None if "session" not in d else IndicesSession.from_dict(d["session"])
),
error=d.get("error", None),
message=d.get("message", None),
)
@modelclass
class TickerSnapshot:
"Contains the most up-to-date market data for all traded ticker symbols."
day: Optional[Agg] = None
last_quote: Optional[LastQuote] = None
last_trade: Optional[LastTrade] = None
min: Optional[MinuteSnapshot] = None
prev_day: Optional[Agg] = None
ticker: Optional[str] = None
todays_change: Optional[float] = None
todays_change_percent: Optional[float] = None
updated: Optional[int] = None
fair_market_value: Optional[float] = None
@staticmethod
def from_dict(d):
return TickerSnapshot(
day=None if "day" not in d else Agg.from_dict(d["day"]),
last_quote=(
None if "lastQuote" not in d else LastQuote.from_dict(d["lastQuote"])
),
last_trade=(
None if "lastTrade" not in d else LastTrade.from_dict(d["lastTrade"])
),
min=None if "min" not in d else MinuteSnapshot.from_dict(d["min"]),
prev_day=None if "prevDay" not in d else Agg.from_dict(d["prevDay"]),
ticker=d.get("ticker", None),
todays_change=d.get("todaysChange", None),
todays_change_percent=d.get("todaysChangePerc", None),
updated=d.get("updated", None),
fair_market_value=d.get("fmv", None),
)
@modelclass
class DayOptionContractSnapshot:
"Contains data for the most recent daily bar in an options contract."
change: Optional[float] = None
change_percent: Optional[float] = None
close: Optional[float] = None
high: Optional[float] = None
last_updated: Optional[int] = None
low: Optional[float] = None
open: Optional[float] = None
previous_close: Optional[float] = None
volume: Optional[float] = None
vwap: Optional[float] = None
@staticmethod
def from_dict(d):
return DayOptionContractSnapshot(**d)
@modelclass
class OptionDetails:
"Contains details for an options contract."
contract_type: Optional[str] = None
exercise_style: Optional[str] = None
expiration_date: Optional[str] = None
shares_per_contract: Optional[float] = None
strike_price: Optional[float] = None
ticker: Optional[str] = None
@staticmethod
def from_dict(d):
return OptionDetails(**d)
@modelclass
class LastQuoteOptionContractSnapshot:
"Contains data for the most recent quote in an options contract."
ask: Optional[float] = None
ask_size: Optional[float] = None
bid: Optional[float] = None
bid_size: Optional[float] = None
last_updated: Optional[int] = None
midpoint: Optional[float] = None
timeframe: Optional[str] = None
@staticmethod
def from_dict(d):
return LastQuoteOptionContractSnapshot(**d)
@modelclass
class LastTradeOptionContractSnapshot:
"Contains data for the most recent trade for an options contract."
price: Optional[float] = None
sip_timestamp: Optional[int] = None
size: Optional[int] = None
conditions: Optional[List[int]] = None
exchange: Optional[int] = None
timeframe: Optional[str] = None
@staticmethod
def from_dict(d):
return LastTradeOptionContractSnapshot(**d)
@modelclass
class Greeks:
"Contains data for the greeks in an options contract."
delta: Optional[float] = None
gamma: Optional[float] = None
theta: Optional[float] = None
vega: Optional[float] = None
@staticmethod
def from_dict(d):
return Greeks(**d)
@modelclass
class UnderlyingAsset:
"Contains data for the underlying stock in an options contract."
change_to_break_even: Optional[float] = None
last_updated: Optional[int] = None
price: Optional[float] = None
value: Optional[float] = None
ticker: Optional[str] = None
timeframe: Optional[str] = None
@staticmethod
def from_dict(d):
return UnderlyingAsset(**d)
@modelclass
class OptionContractSnapshot:
"Contains data for the snapshot of an option contract of a stock equity."
break_even_price: Optional[float] = None
day: Optional[DayOptionContractSnapshot] = None
details: Optional[OptionDetails] = None
greeks: Optional[Greeks] = None
implied_volatility: Optional[float] = None
last_quote: Optional[LastQuoteOptionContractSnapshot] = None
last_trade: Optional[LastTradeOptionContractSnapshot] = None
open_interest: Optional[float] = None
underlying_asset: Optional[UnderlyingAsset] = None
fair_market_value: Optional[float] = None
@staticmethod
def from_dict(d):
return OptionContractSnapshot(
break_even_price=d.get("break_even_price", None),
day=(
None
if "day" not in d
else DayOptionContractSnapshot.from_dict(d["day"])
),
details=(
None if "details" not in d else OptionDetails.from_dict(d["details"])
),
greeks=None if "greeks" not in d else Greeks.from_dict(d["greeks"]),
implied_volatility=d.get("implied_volatility", None),
last_quote=(
None
if "last_quote" not in d
else LastQuoteOptionContractSnapshot.from_dict(d["last_quote"])
),
last_trade=(
None
if "last_trade" not in d
else LastTradeOptionContractSnapshot.from_dict(d["last_trade"])
),
open_interest=d.get("open_interest", None),
underlying_asset=(
None
if "underlying_asset" not in d
else UnderlyingAsset.from_dict(d["underlying_asset"])
),
fair_market_value=d.get("fmv", None),
)
@modelclass
class OrderBookQuote:
"Contains data for a book bid or ask."
price: Optional[float] = None
exchange_shares: Optional[Dict[str, float]] = None
@staticmethod
def from_dict(d):
return OrderBookQuote(
d.get("p", None),
d.get("x", None),
)
@modelclass
class SnapshotTickerFullBook:
"Contains the current level 2 book of a single ticker. This is the combined book from all of the exchanges."
ticker: Optional[str] = None
bids: Optional[List[OrderBookQuote]] = None
asks: Optional[List[OrderBookQuote]] = None
bid_count: Optional[float] = None
ask_count: Optional[float] = None
spread: Optional[float] = None
updated: Optional[int] = None
@staticmethod
def from_dict(d):
return SnapshotTickerFullBook(
ticker=d.get("ticker", None),
bids=(
None
if "bids" not in d
else [OrderBookQuote.from_dict(o) for o in d["bids"]]
),
asks=(
None
if "asks" not in d
else [OrderBookQuote.from_dict(o) for o in d["asks"]]
),
bid_count=d.get("bidCount", None),
ask_count=d.get("askCount", None),
spread=d.get("spread", None),
updated=d.get("updated", None),
)
@modelclass
class UniversalSnapshotSession:
"""Contains data about the most recent trading session for an asset."""
price: Optional[float] = None
change: Optional[float] = None
change_percent: Optional[float] = None
early_trading_change: Optional[float] = None
early_trading_change_percent: Optional[float] = None
regular_trading_change: Optional[float] = None
regular_trading_change_percent: Optional[float] = None
late_trading_change: Optional[float] = None
late_trading_change_percent: Optional[float] = None
open: Optional[float] = None
close: Optional[float] = None
high: Optional[float] = None
low: Optional[float] = None
previous_close: Optional[float] = None
volume: Optional[float] = None
vwap: Optional[float] = None
last_updated: Optional[int] = None
fractional_volume: Optional[str] = None
@staticmethod
def from_dict(d):
return UniversalSnapshotSession(
price=d.get("price"),
change=d.get("change"),
change_percent=d.get("change_percent"),
early_trading_change=d.get("early_trading_change"),
early_trading_change_percent=d.get("early_trading_change_percent"),
regular_trading_change=d.get("regular_trading_change"),
regular_trading_change_percent=d.get("regular_trading_change_percent"),
late_trading_change=d.get("late_trading_change"),
late_trading_change_percent=d.get("late_trading_change_percent"),
open=d.get("open"),
close=d.get("close"),
high=d.get("high"),
low=d.get("low"),
previous_close=d.get("previous_close"),
volume=d.get("volume"),
vwap=d.get("vwap"),
last_updated=d.get("last_updated"),
fractional_volume=d.get("decimal_volume"),
)
@modelclass
class UniversalSnapshotLastQuote:
"""Contains the most recent quote for an asset."""
ask: Optional[float] = None
ask_size: Optional[float] = None
ask_exchange: Optional[int] = None
bid: Optional[float] = None
bid_size: Optional[float] = None
bid_exchange: Optional[int] = None
midpoint: Optional[float] = None
exchange: Optional[int] = None
timeframe: Optional[str] = None
last_updated: Optional[int] = None
@staticmethod
def from_dict(d):
return UniversalSnapshotLastQuote(
ask=d.get("ask"),
ask_size=d.get("ask_size"),
ask_exchange=d.get("ask_exchange"),
bid=d.get("bid"),
bid_size=d.get("bid_size"),
bid_exchange=d.get("bid_exchange"),
midpoint=d.get("midpoint"),
exchange=d.get("exchange"),
timeframe=d.get("timeframe"),
last_updated=d.get("last_updated"),
)
@modelclass
class UniversalSnapshotLastTrade:
"""Contains the most recent trade for an asset."""
id: Optional[int] = None
price: Optional[float] = None
size: Optional[int] = None
exchange: Optional[int] = None
conditions: Optional[List[int]] = None
timeframe: Optional[str] = None
last_updated: Optional[int] = None
participant_timestamp: Optional[int] = None
sip_timestamp: Optional[int] = None
fractional_size: Optional[str] = None
@staticmethod
def from_dict(d):
return UniversalSnapshotLastTrade(
id=d.get("id"),
price=d.get("price"),
size=d.get("size"),
exchange=d.get("exchange"),
conditions=d.get("conditions"),
timeframe=d.get("timeframe"),
last_updated=d.get("last_updated"),
participant_timestamp=d.get("participant_timestamp"),
sip_timestamp=d.get("sip_timestamp"),
fractional_size=d.get("decimal_size"),
)
@modelclass
class UniversalSnapshotLastMinute:
"""Contains the most recent minute-level aggregate for the asset."""
open: Optional[float] = None
close: Optional[float] = None
high: Optional[float] = None
low: Optional[float] = None
volume: Optional[float] = None
vwap: Optional[float] = None
transactions: Optional[int] = None
last_updated: Optional[int] = None
fractional_volume: Optional[str] = None
@staticmethod
def from_dict(d):
return UniversalSnapshotLastMinute(
open=d.get("open"),
close=d.get("close"),
high=d.get("high"),
low=d.get("low"),
volume=d.get("volume"),
vwap=d.get("vwap"),
transactions=d.get("transactions"),
last_updated=d.get("last_updated"),
fractional_volume=d.get("decimal_volume"),
)
@modelclass
class UniversalSnapshotUnderlyingAsset:
"""Contains data for the underlying stock in an options contract."""
ticker: Optional[str] = None
price: Optional[float] = None
value: Optional[float] = None
change_to_break_even: Optional[float] = None
timeframe: Optional[str] = None
last_updated: Optional[int] = None
@staticmethod
def from_dict(d):
return UniversalSnapshotUnderlyingAsset(
ticker=d.get("ticker"),
price=d.get("price"),
value=d.get("value"),
change_to_break_even=d.get("change_to_break_even"),
timeframe=d.get("timeframe"),
last_updated=d.get("last_updated"),
)
@modelclass
class UniversalSnapshotDetails:
"""Contains details for an options contract."""
contract_type: Optional[str] = None
exercise_style: Optional[str] = None
expiration_date: Optional[str] = None
shares_per_contract: Optional[float] = None
strike_price: Optional[float] = None
@staticmethod
def from_dict(d):
return UniversalSnapshotDetails(
contract_type=d.get("contract_type"),
exercise_style=d.get("exercise_style"),
expiration_date=d.get("expiration_date"),
shares_per_contract=d.get("shares_per_contract"),
strike_price=d.get("strike_price"),
)
@modelclass
class UniversalSnapshot:
"""Contains snapshot data for an asset (stocks, options, indices, fx, crypto)."""
ticker: Optional[str] = None
type: Optional[str] = None
session: Optional[UniversalSnapshotSession] = None
last_quote: Optional[UniversalSnapshotLastQuote] = None
last_trade: Optional[UniversalSnapshotLastTrade] = None
last_minute: Optional[UniversalSnapshotLastMinute] = None
greeks: Optional[Greeks] = None
underlying_asset: Optional[UniversalSnapshotUnderlyingAsset] = None
details: Optional[UniversalSnapshotDetails] = None
break_even_price: Optional[float] = None
implied_volatility: Optional[float] = None
open_interest: Optional[float] = None
market_status: Optional[str] = None
name: Optional[str] = None
fair_market_value: Optional[float] = None
error: Optional[str] = None
message: Optional[str] = None
value: Optional[float] = None
last_updated: Optional[int] = None
timeframe: Optional[str] = None
@staticmethod
def from_dict(d):
return UniversalSnapshot(
ticker=d.get("ticker"),
type=d.get("type"),
session=(
None
if "session" not in d
else UniversalSnapshotSession.from_dict(d["session"])
),
last_quote=(
None
if "last_quote" not in d
else UniversalSnapshotLastQuote.from_dict(d["last_quote"])
),
last_trade=(
None
if "last_trade" not in d
else UniversalSnapshotLastTrade.from_dict(d["last_trade"])
),
last_minute=(
None
if "last_minute" not in d
else UniversalSnapshotLastMinute.from_dict(d["last_minute"])
),
greeks=(None if "greeks" not in d else Greeks.from_dict(d["greeks"])),
underlying_asset=(
None
if "underlying_asset" not in d
else UniversalSnapshotUnderlyingAsset.from_dict(d["underlying_asset"])
),
details=(
None
if "details" not in d
else UniversalSnapshotDetails.from_dict(d["details"])
),
break_even_price=d.get("break_even_price"),
implied_volatility=d.get("implied_volatility"),
open_interest=d.get("open_interest"),
market_status=d.get("market_status"),
name=d.get("name"),
fair_market_value=d.get("fmv"),
error=d.get("error"),
message=d.get("message"),
value=d.get("value"),
last_updated=d.get("last_updated"),
timeframe=d.get("timeframe"),
)