-
-
Notifications
You must be signed in to change notification settings - Fork 663
/
book_test.py
57 lines (43 loc) · 1.28 KB
/
book_test.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
'''
Copyright (C) 2017-2024 Bryant Moscon - bmoscon@gmail.com
Please see the LICENSE file for the terms and conditions
associated with this software.
'''
import time
from cryptofeed import FeedHandler
from cryptofeed.callback import BookCallback
from cryptofeed.defines import BID, ASK, L2_BOOK
from cryptofeed.exchanges import Bitmex
counter = 0
avg = 0
START = time.time()
STATS = 1000
async def book(feed, symbol, book, timestamp):
global counter
global avg
t = time.time()
counter += 1
bids = list(book[BID].keys())
asks = list(book[ASK].keys())
avg += (t - timestamp)
try:
assert (t - timestamp) < 2
assert bids[-1] < asks[0]
except Exception:
print("FAILED")
print("BID", bids[-1])
print("ASKS", asks[0])
print("DELTA", t - timestamp)
print("COUNTER", counter)
if counter % STATS == 0:
print("Checked", counter, "updates")
print("Runtime", t - START)
print("Current spread", asks[0] - bids[-1])
print("Average book update handle time", avg / counter)
print("\n")
def main():
f = FeedHandler()
f.add_feed(Bitmex(symbols=['BTC-USD'], channels=[L2_BOOK], callbacks={L2_BOOK: BookCallback(book)}))
f.run()
if __name__ == '__main__':
main()