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

Broadcast all tick types to subs, not just trades

cached_feeds
Tyler Goodlet 2021-03-30 10:56:19 -04:00
parent 5fc2aba3ed
commit 4f51ca74f4
1 changed files with 35 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
piker/data/__init__.py
View File

@ -206,7 +206,6 @@ async def allocate_persistent_feed(
bus.feeds[symbol] = (cs, init_msg, first_quote) bus.feeds[symbol] = (cs, init_msg, first_quote)
with cs: with cs:
if opened: if opened:
# start history backfill task # start history backfill task
# ``backfill_bars()`` is a required backend func # ``backfill_bars()`` is a required backend func
@ -246,43 +245,47 @@ async def allocate_persistent_feed(
# while another consumer is serviced.. # while another consumer is serviced..
# start writing the shm buffer with appropriate trade data # start writing the shm buffer with appropriate trade data
for tick in iterticks(quote, types=('trade', 'utrade',)): for tick in quote['ticks']:
last = tick['price']
# update last entry # write trade events to shm last OHLC sample
# benchmarked in the 4-5 us range if tick['type'] in ('trade', 'utrade'):
o, high, low, v = shm.array[-1][
['open', 'high', 'low', 'volume']
]
new_v = tick.get('size', 0) last = tick['price']
if v == 0 and new_v: # update last entry
# no trades for this bar yet so the open # benchmarked in the 4-5 us range
# is also the close/last trade price o, high, low, v = shm.array[-1][
o = last ['open', 'high', 'low', 'volume']
]
if sum_tick_vlm: new_v = tick.get('size', 0)
volume = v + new_v
else:
volume = v
shm.array[[ if v == 0 and new_v:
'open', # no trades for this bar yet so the open
'high', # is also the close/last trade price
'low', o = last
'close',
'volume',
]][-1] = (
o,
max(high, last),
min(low, last),
last,
volume,
)
for ctx in bus.subscribers[sym]: if sum_tick_vlm:
await ctx.send_yield({sym: quote}) volume = v + new_v
else:
volume = v
shm.array[[
'open',
'high',
'low',
'close',
'volume',
]][-1] = (
o,
max(high, last),
min(low, last),
last,
volume,
)
for ctx in bus.subscribers[sym]:
await ctx.send_yield({sym: quote})
@tractor.stream @tractor.stream