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

Update history shm buffer in ohlc sampler loop

history_view
Tyler Goodlet 2022-09-01 11:27:39 -04:00
parent 49ccfdd673
commit 6e574835c8
1 changed files with 38 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
piker/data/_sampling.py
View File

@ -137,7 +137,7 @@ async def increment_ohlc_buffer(
# this copies non-std fields (eg. vwap) from the last datum # this copies non-std fields (eg. vwap) from the last datum
last[ last[
['time', 'volume', 'open', 'high', 'low', 'close'] ['time', 'volume', 'open', 'high', 'low', 'close']
][0] = (t + delay_s, 0, close, close, close, close) ][0] = (t + this_delay_s, 0, close, close, close, close)
# write to the buffer # write to the buffer
shm.push(last) shm.push(last)
@ -227,7 +227,8 @@ async def iter_ohlc_periods(
async def sample_and_broadcast( async def sample_and_broadcast(
bus: _FeedsBus, # noqa bus: _FeedsBus, # noqa
shm: ShmArray, rt_shm: ShmArray,
hist_shm: ShmArray,
quote_stream: trio.abc.ReceiveChannel, quote_stream: trio.abc.ReceiveChannel,
brokername: str, brokername: str,
sum_tick_vlm: bool = True, sum_tick_vlm: bool = True,
@ -263,41 +264,45 @@ async def sample_and_broadcast(
last = tick['price'] last = tick['price']
# update last entry # more compact inline-way to do this assignment
# benchmarked in the 4-5 us range # to both buffers?
o, high, low, v = shm.array[-1][ for shm in [rt_shm, hist_shm]:
['open', 'high', 'low', 'volume'] # update last entry
] # benchmarked in the 4-5 us range
# for shm in [rt_shm, hist_shm]:
o, high, low, v = shm.array[-1][
['open', 'high', 'low', 'volume']
]
new_v = tick.get('size', 0) new_v = tick.get('size', 0)
if v == 0 and new_v: if v == 0 and new_v:
# no trades for this bar yet so the open # no trades for this bar yet so the open
# is also the close/last trade price # is also the close/last trade price
o = last o = last
if sum_tick_vlm: if sum_tick_vlm:
volume = v + new_v volume = v + new_v
else: else:
# presume backend takes care of summing # presume backend takes care of summing
# it's own vlm # it's own vlm
volume = quote['volume'] volume = quote['volume']
shm.array[[ shm.array[[
'open', 'open',
'high', 'high',
'low', 'low',
'close', 'close',
'bar_wap', # can be optionally provided 'bar_wap', # can be optionally provided
'volume', 'volume',
]][-1] = ( ]][-1] = (
o, o,
max(high, last), max(high, last),
min(low, last), min(low, last),
last, last,
quote.get('bar_wap', 0), quote.get('bar_wap', 0),
volume, volume,
) )
# XXX: we need to be very cautious here that no # XXX: we need to be very cautious here that no
# context-channel is left lingering which doesn't have # context-channel is left lingering which doesn't have