pikers
/
piker
4
1
Fork 0
Code Issues 10 Pull Requests 14 Packages Projects Releases Wiki Activity

Push only new key value pairs over quote streams 

This is something I've been meaning to try for a while and will likely
make writing tick data to a db more straight forward (filling in NaN
values is more matter of fact) plus it should minimize bandwidth usage.
Note, it'll require stream consumers to be considerate of non-full
quotes arriving and thus using the first "full" quote message to fill
out dynamically formatted systems or displays.
its_happening
Tyler Goodlet 2020-05-31 22:36:47 -04:00
parent 2f2ff7cded
commit 3042d1eec6
1 changed files with 17 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
piker/brokers/data.py
View File

@ -130,17 +130,33 @@ async def stream_requests(
for quote in quotes:
symbol = quote['symbol']
last = _cache.setdefault(symbol, {})
# find all keys that have match to a new value compared
# to the last quote received
new = set(quote.items()) - set(last.items())
if new:
log.info(
f"New quote {quote['symbol']}:\n{new}")
_cache[symbol] = quote
# only ship diff updates and other required fields
payload['symbol'] = symbol
payload = {k: quote[k] for k, v in new}
# if there was volume likely the last size of
# shares traded is useful info and it's possible
# that the set difference from above will disregard
# a "size" value since the same # of shares were traded
size = quote.get('size')
if size and 'volume' in payload:
payload['size'] = size
# XXX: we append to a list for the options case where the
# subscription topic (key) is the same for all
# expiries even though this is uncessary for the
# stock case (different topic [i.e. symbol] for each
# quote).
new_quotes.setdefault(quote['key'], []).append(quote)
new_quotes.setdefault(quote['key'], []).append(payload)
else:
# log.debug(f"Delivering quotes:\n{quotes}")
for quote in quotes: