pikers
/
piker
3
0
Fork 0
Code Issues 2 Pull Requests 5 Packages Projects Releases Wiki Activity

Add multi-broker streaming test using both `binance` and `kraken`

agg_feedz
Tyler Goodlet 2022-11-11 17:27:02 -05:00
parent 7bbe86d6fb
commit e348968113
1 changed files with 50 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
tests/test_feeds.py
View File

@ -6,7 +6,7 @@ from collections import Counter
from pprint import pprint from pprint import pprint
import pytest import pytest
import tractor # import tractor
import trio import trio
from piker import ( from piker import (
open_piker_runtime, open_piker_runtime,
@ -21,12 +21,18 @@ from piker.data._source import (
@pytest.mark.parametrize( @pytest.mark.parametrize(
'fqsns', 'fqsns',
[ [
# binance
(100, {'btcusdt.binance', 'ethusdt.binance'}), (100, {'btcusdt.binance', 'ethusdt.binance'}),
(50, {'xbteur.kraken', 'xbtusd.kraken'}),
# kraken
(20, {'xbteur.kraken', 'xbtusd.kraken'}),
# binance + kraken
(200, {'btcusdt.binance', 'xbtusd.kraken'}),
], ],
ids=lambda param: f'quotes={param[0]}@fqsns={param[1]}', ids=lambda param: f'quotes={param[0]}@fqsns={param[1]}',
) )
def test_basic_rt_feed( def test_multi_fqsn_feed(
fqsns: set[str], fqsns: set[str],
): ):
''' '''
@ -41,10 +47,6 @@ def test_basic_rt_feed(
brokername, key, suffix = unpack_fqsn(fqsn) brokername, key, suffix = unpack_fqsn(fqsn)
brokers.add(brokername) brokers.add(brokername)
# NOTE: we only have single broker-backed multi-symbol streams
# currently.
assert len(brokers) == 1
async def main(): async def main():
async with ( async with (
open_piker_runtime( open_piker_runtime(
@ -52,9 +54,9 @@ def test_basic_rt_feed(
# XXX tractor BUG: this doesn't translate through to the # XXX tractor BUG: this doesn't translate through to the
# ``tractor._state._runtimevars``... # ``tractor._state._runtimevars``...
registry_addr=('127.0.0.1', 6666), # registry_addr=('127.0.0.1', 6666),
# debug_mode=True, debug_mode=True,
), ),
open_feed( open_feed(
fqsns, fqsns,
@ -73,55 +75,54 @@ def test_basic_rt_feed(
ohlcv: ShmArray = flume.rt_shm ohlcv: ShmArray = flume.rt_shm
hist_ohlcv: ShmArray = flume.hist_shm hist_ohlcv: ShmArray = flume.hist_shm
# stream some ticks and ensure we see data from both symbol async with feed.open_multi_stream(brokers) as stream:
# subscriptions.
stream = feed.streams[brokername]
# pull the first startup quotes, one for each fqsn, and # pull the first startup quotes, one for each fqsn, and
# ensure they match each flume's startup quote value. # ensure they match each flume's startup quote value.
fqsns_copy = fqsns.copy() fqsns_copy = fqsns.copy()
for _ in range(1): with trio.fail_after(0.5):
first_quotes = await stream.receive() for _ in range(1):
for fqsn, quote in first_quotes.items(): first_quotes = await stream.receive()
for fqsn, quote in first_quotes.items():
# XXX: TODO: WTF apparently this error will get # XXX: TODO: WTF apparently this error will get
# supressed and only show up in the teardown # supressed and only show up in the teardown
# excgroup if we don't have the fix from # excgroup if we don't have the fix from
# <tractorbugurl> # <tractorbugurl>
# assert 0 # assert 0
fqsns_copy.remove(fqsn) fqsns_copy.remove(fqsn)
flume = feed.flumes[fqsn] flume = feed.flumes[fqsn]
assert quote['last'] == flume.first_quote['last'] assert quote['last'] == flume.first_quote['last']
cntr = Counter() cntr = Counter()
async for quotes in stream: with trio.fail_after(3):
for fqsn, quote in quotes.items(): async for quotes in stream:
cntr[fqsn] += 1 for fqsn, quote in quotes.items():
cntr[fqsn] += 1
# await tractor.breakpoint() # await tractor.breakpoint()
flume = feed.flumes[fqsn] flume = feed.flumes[fqsn]
ohlcv: ShmArray = flume.rt_shm ohlcv: ShmArray = flume.rt_shm
hist_ohlcv: ShmArray = flume.hist_shm hist_ohlcv: ShmArray = flume.hist_shm
# print quote msg, rt and history # print quote msg, rt and history
# buffer values on console. # buffer values on console.
rt_row = ohlcv.array[-1] rt_row = ohlcv.array[-1]
hist_row = hist_ohlcv.array[-1] hist_row = hist_ohlcv.array[-1]
# last = quote['last'] # last = quote['last']
# assert last == rt_row['close'] # assert last == rt_row['close']
# assert last == hist_row['close'] # assert last == hist_row['close']
pprint( pprint(
f'{fqsn}: {quote}\n' f'{fqsn}: {quote}\n'
f'rt_ohlc: {rt_row}\n' f'rt_ohlc: {rt_row}\n'
f'hist_ohlc: {hist_row}\n' f'hist_ohlc: {hist_row}\n'
) )
if cntr.total() >= 100: if cntr.total() >= max_quotes:
break break
# await tractor.breakpoint()
assert set(cntr.keys()) == fqsns assert set(cntr.keys()) == fqsns
trio.run(main) trio.run(main)