gap_annotator: super fast (time-)gap detection API #75

Open
goodboy wants to merge 14 commits from gap_annotator into ib_venue_closures
21 changed files with 1148 additions and 1401 deletions

View File

@ -2,7 +2,7 @@
-------------- --------------
more or less the "everything broker" for traditional and international more or less the "everything broker" for traditional and international
markets. they are the "go to" provider for automatic retail trading markets. they are the "go to" provider for automatic retail trading
and we interface to their APIs using the `ib_insync` project. and we interface to their APIs using the `ib_async` project.
status status
****** ******

View File

@ -22,7 +22,7 @@ Sub-modules within break into the core functionalities:
- ``broker.py`` part for orders / trading endpoints - ``broker.py`` part for orders / trading endpoints
- ``feed.py`` for real-time data feed endpoints - ``feed.py`` for real-time data feed endpoints
- ``api.py`` for the core API machinery which is ``trio``-ized - ``api.py`` for the core API machinery which is ``trio``-ized
wrapping around ``ib_insync``. wrapping around `ib_async`.
""" """
from .api import ( from .api import (

View File

@ -111,7 +111,7 @@ def load_flex_trades(
) -> dict[str, Any]: ) -> dict[str, Any]:
from ib_insync import flexreport, util from ib_async import flexreport, util
conf = get_config() conf = get_config()
@ -154,8 +154,7 @@ def load_flex_trades(
trade_entries, trade_entries,
) )
ledger_dict: dict | None = None ledger_dict: dict|None
for acctid in trades_by_account: for acctid in trades_by_account:
trades_by_id = trades_by_account[acctid] trades_by_id = trades_by_account[acctid]

View File

@ -15,7 +15,8 @@
# along with this program. If not, see <https://www.gnu.org/licenses/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
''' '''
Core API client machinery; mostly sane/useful wrapping around `ib_insync`.. Core API client machinery; mostly sane/useful wrapping around
`ib_async`..
''' '''
from __future__ import annotations from __future__ import annotations
@ -57,7 +58,7 @@ from pendulum import (
Interval, Interval,
) )
from eventkit import Event from eventkit import Event
from ib_insync import ( from ib_async import (
client as ib_client, client as ib_client,
IB, IB,
Contract, Contract,
@ -143,7 +144,7 @@ _bar_sizes = {
_show_wap_in_history: bool = False _show_wap_in_history: bool = False
# overrides to sidestep pretty questionable design decisions in # overrides to sidestep pretty questionable design decisions in
# ``ib_insync``: # ``ib_async``:
class NonShittyWrapper(Wrapper): class NonShittyWrapper(Wrapper):
def tcpDataArrived(self): def tcpDataArrived(self):
"""Override time stamps to be floats for now. """Override time stamps to be floats for now.
@ -183,7 +184,7 @@ class NonShittyIB(IB):
''' '''
def __init__(self): def __init__(self):
# override `ib_insync` internal loggers so we can see wtf # override `ib_async` internal loggers so we can see wtf
# it's doing.. # it's doing..
self._logger = get_logger( self._logger = get_logger(
name=__name__, name=__name__,
@ -194,7 +195,7 @@ class NonShittyIB(IB):
self.wrapper = NonShittyWrapper(self) self.wrapper = NonShittyWrapper(self)
self.client = ib_client.Client(self.wrapper) self.client = ib_client.Client(self.wrapper)
self.client._logger = get_logger( self.client._logger = get_logger(
name='ib_insync.client', name='ib_async.client',
) )
# self.errorEvent += self._onError # self.errorEvent += self._onError
@ -779,13 +780,14 @@ class Client:
ContFuture(symbol, exchange=exchange) ContFuture(symbol, exchange=exchange)
))[0] ))[0]
else: else:
con = (await self.ib.qualifyContractsAsync( cons = (await self.ib.qualifyContractsAsync(
Future( Future(
symbol, symbol,
exchange=exchange, exchange=exchange,
lastTradeDateOrContractMonth=expiry, lastTradeDateOrContractMonth=expiry,
) )
))[0] ))
con = cons[0]
return con return con
@ -878,7 +880,7 @@ class Client:
currency='USD', currency='USD',
exchange='PAXOS', exchange='PAXOS',
) )
# XXX, on `ib_insync` when first tried this, # XXX, on `ib_async` when first tried this,
# > Error 10299, reqId 141: Expected what to show is # > Error 10299, reqId 141: Expected what to show is
# > AGGTRADES, please use that instead of TRADES., # > AGGTRADES, please use that instead of TRADES.,
# > contract: Crypto(conId=479624278, symbol='BTC', # > contract: Crypto(conId=479624278, symbol='BTC',
@ -992,7 +994,6 @@ class Client:
async def get_sym_details( async def get_sym_details(
self, self,
fqme: str, fqme: str,
) -> tuple[ ) -> tuple[
Contract, Contract,
ContractDetails, ContractDetails,
@ -1092,7 +1093,7 @@ class Client:
size: int, size: int,
account: str, # if blank the "default" tws account is used account: str, # if blank the "default" tws account is used
# XXX: by default 0 tells ``ib_insync`` methods that there is no # XXX: by default 0 tells ``ib_async`` methods that there is no
# existing order so ask the client to create a new one (which it # existing order so ask the client to create a new one (which it
# seems to do by allocating an int counter - collision prone..) # seems to do by allocating an int counter - collision prone..)
reqid: int = None, reqid: int = None,
@ -1281,7 +1282,7 @@ async def load_aio_clients(
port: int = None, port: int = None,
client_id: int = 6116, client_id: int = 6116,
# the API TCP in `ib_insync` connection can be flaky af so instead # the API TCP in `ib_async` connection can be flaky af so instead
# retry a few times to get the client going.. # retry a few times to get the client going..
connect_retries: int = 3, connect_retries: int = 3,
connect_timeout: float = 30, # in case a remote-host connect_timeout: float = 30, # in case a remote-host
@ -1289,7 +1290,7 @@ async def load_aio_clients(
) -> dict[str, Client]: ) -> dict[str, Client]:
''' '''
Return an ``ib_insync.IB`` instance wrapped in our client API. Return an ``ib_async.IB`` instance wrapped in our client API.
Client instances are cached for later use. Client instances are cached for later use.
@ -1631,6 +1632,7 @@ async def open_aio_client_method_relay(
) -> None: ) -> None:
# with tractor.devx.maybe_open_crash_handler() as _bxerr:
# sync with `open_client_proxy()` caller # sync with `open_client_proxy()` caller
chan.started_nowait(client) chan.started_nowait(client)
@ -1640,7 +1642,11 @@ async def open_aio_client_method_relay(
# relay all method requests to ``asyncio``-side client and deliver # relay all method requests to ``asyncio``-side client and deliver
# back results # back results
while not chan._to_trio._closed: # <- TODO, better check like `._web_bs`? while not chan._to_trio._closed: # <- TODO, better check like `._web_bs`?
msg: tuple[str, dict]|dict|None = await chan.get() msg: (
None
|tuple[str, dict]
|dict
) = await chan.get()
match msg: match msg:
case None: # termination sentinel case None: # termination sentinel
log.info('asyncio `Client` method-proxy SHUTDOWN!') log.info('asyncio `Client` method-proxy SHUTDOWN!')
@ -1742,7 +1748,7 @@ async def get_client(
) -> Client: ) -> Client:
''' '''
Init the ``ib_insync`` client in another actor and return Init the ``ib_async`` client in another actor and return
a method proxy to it. a method proxy to it.
''' '''

View File

@ -35,14 +35,14 @@ from trio_typing import TaskStatus
import tractor import tractor
from tractor.to_asyncio import LinkedTaskChannel from tractor.to_asyncio import LinkedTaskChannel
from tractor import trionics from tractor import trionics
from ib_insync.contract import ( from ib_async.contract import (
Contract, Contract,
) )
from ib_insync.order import ( from ib_async.order import (
Trade, Trade,
OrderStatus, OrderStatus,
) )
from ib_insync.objects import ( from ib_async.objects import (
Fill, Fill,
Execution, Execution,
CommissionReport, CommissionReport,
@ -181,7 +181,7 @@ async def handle_order_requests(
# validate # validate
order = BrokerdOrder(**request_msg) order = BrokerdOrder(**request_msg)
# XXX: by default 0 tells ``ib_insync`` methods that # XXX: by default 0 tells ``ib_async`` methods that
# there is no existing order so ask the client to create # there is no existing order so ask the client to create
# a new one (which it seems to do by allocating an int # a new one (which it seems to do by allocating an int
# counter - collision prone..) # counter - collision prone..)
@ -237,7 +237,7 @@ async def recv_trade_updates(
) -> None: ) -> None:
''' '''
Receive and relay order control and positioning related events Receive and relay order control and positioning related events
from `ib_insync`, pack as tuples and push over mem-chan to our from `ib_async`, pack as tuples and push over mem-chan to our
trio relay task for processing and relay to EMS. trio relay task for processing and relay to EMS.
''' '''
@ -303,7 +303,7 @@ async def recv_trade_updates(
# much more then a few more pnl fields.. # much more then a few more pnl fields..
# 'updatePortfolioEvent', # 'updatePortfolioEvent',
# XXX: these all seem to be weird ib_insync internal # XXX: these all seem to be weird ib_async internal
# events that we probably don't care that much about # events that we probably don't care that much about
# given the internal design is wonky af.. # given the internal design is wonky af..
# 'newOrderEvent', # 'newOrderEvent',
@ -499,7 +499,7 @@ async def open_trade_event_stream(
] = trio.TASK_STATUS_IGNORED, ] = trio.TASK_STATUS_IGNORED,
): ):
''' '''
Proxy wrapper for starting trade event stream from ib_insync Proxy wrapper for starting trade event stream from ib_async
which spawns an asyncio task that registers an internal closure which spawns an asyncio task that registers an internal closure
(`push_tradies()`) which in turn relays trading events through (`push_tradies()`) which in turn relays trading events through
a `tractor.to_asyncio.LinkedTaskChannel` which the parent a `tractor.to_asyncio.LinkedTaskChannel` which the parent

View File

@ -36,7 +36,7 @@ from typing import (
) )
from async_generator import aclosing from async_generator import aclosing
import ib_insync as ibis import ib_async as ibis
import numpy as np import numpy as np
from pendulum import ( from pendulum import (
now, now,
@ -100,7 +100,7 @@ tick_types = {
5: 'size', 5: 'size',
8: 'volume', 8: 'volume',
# ``ib_insync`` already packs these into # `ib_async` already packs these into
# quotes under the following fields. # quotes under the following fields.
55: 'trades_per_min', # `'tradeRate'` 55: 'trades_per_min', # `'tradeRate'`
56: 'vlm_per_min', # `'volumeRate'` 56: 'vlm_per_min', # `'volumeRate'`
@ -304,7 +304,7 @@ async def open_history_client(
# TODO: it seems like we can do async queries for ohlc # TODO: it seems like we can do async queries for ohlc
# but getting the order right still isn't working and I'm not # but getting the order right still isn't working and I'm not
# quite sure why.. needs some tinkering and probably # quite sure why.. needs some tinkering and probably
# a lookthrough of the `ib_insync` machinery, for eg. maybe # a lookthrough of the `ib_async` machinery, for eg. maybe
# we have to do the batch queries on the `asyncio` side? # we have to do the batch queries on the `asyncio` side?
yield ( yield (
get_hist, get_hist,
@ -1051,6 +1051,21 @@ def normalize(
# ticker.rtTime.timestamp) / 1000. # ticker.rtTime.timestamp) / 1000.
data.pop('rtTime') data.pop('rtTime')
# XXX, `ib_async` seems to set a
# `'timezone': datetime.timezone.utc` in this `dict`
# which is NOT IPC serializeable sin codec!
#
# pretty sure we don't need any of this field for now anyway?
data.pop('defaults')
if lts := data.get('lastTimeStamp'):
lts.replace(tzinfo=None)
log.warning(
f'Stripping `.tzinfo` from datetime\n'
f'{lts}\n'
)
# breakpoint()
return data return data
@ -1227,7 +1242,7 @@ async def stream_quotes(
): ):
# ?TODO? can we rm this - particularly for `ib_async`? # ?TODO? can we rm this - particularly for `ib_async`?
# ugh, clear ticks since we've consumed them # ugh, clear ticks since we've consumed them
# (ahem, ib_insync is stateful trash) # (ahem, ib_async is stateful trash)
# first_ticker.ticks = [] # first_ticker.ticks = []
# only on first entry at feed boot up # only on first entry at feed boot up

View File

@ -36,7 +36,7 @@ from pendulum import (
parse, parse,
from_timestamp, from_timestamp,
) )
from ib_insync import ( from ib_async import (
Contract, Contract,
Commodity, Commodity,
Fill, Fill,

View File

@ -30,7 +30,7 @@ from typing import (
) )
from rapidfuzz import process as fuzzy from rapidfuzz import process as fuzzy
import ib_insync as ibis import ib_async as ibis
import tractor import tractor
import trio import trio
@ -522,7 +522,11 @@ async def get_mkt_info(
if atype == 'commodity': if atype == 'commodity':
venue: str = 'cmdty' venue: str = 'cmdty'
else: else:
venue = con.primaryExchange or con.exchange venue: str = (
con.primaryExchange
or
con.exchange
)
price_tick: Decimal = Decimal(str(details.minTick)) price_tick: Decimal = Decimal(str(details.minTick))
ib_min_tick_gt_2: Decimal = Decimal('0.01') ib_min_tick_gt_2: Decimal = Decimal('0.01')

View File

@ -41,7 +41,7 @@ from pendulum import (
) )
if TYPE_CHECKING: if TYPE_CHECKING:
from ib_insync import ( from ib_async import (
TradingSession, TradingSession,
ContractDetails, ContractDetails,
) )
@ -236,7 +236,7 @@ def is_venue_closure(
# #
# NOTE, this was generated by @guille from a gpt5 prompt # NOTE, this was generated by @guille from a gpt5 prompt
# and was originally thot to be needed before learning about # and was originally thot to be needed before learning about
# `ib_insync.contract.ContractDetails._parseSessions()` and # `ib_async.contract.ContractDetails._parseSessions()` and
# it's downstream meths.. # it's downstream meths..
# #
# This is still likely useful to keep for now to parse the # This is still likely useful to keep for now to parse the

View File

@ -973,9 +973,6 @@ async def open_feed(
# assert flume.mkt.fqme == fqme # assert flume.mkt.fqme == fqme
feed.flumes[fqme] = flume feed.flumes[fqme] = flume
# TODO: do we need this?
flume.feed = feed
# attach and cache shm handles # attach and cache shm handles
rt_shm = flume.rt_shm rt_shm = flume.rt_shm
assert rt_shm assert rt_shm

View File

@ -22,9 +22,6 @@ real-time data processing data-structures.
""" """
from __future__ import annotations from __future__ import annotations
from typing import (
TYPE_CHECKING,
)
import tractor import tractor
import pendulum import pendulum
@ -38,9 +35,6 @@ from ._sharedmem import (
) )
from piker.accounting import MktPair from piker.accounting import MktPair
if TYPE_CHECKING:
from piker.data.feed import Feed
class Flume(Struct): class Flume(Struct):
''' '''
@ -80,10 +74,6 @@ class Flume(Struct):
izero_rt: int = 0 izero_rt: int = 0
throttle_rate: int | None = None throttle_rate: int | None = None
# TODO: do we need this really if we can pull the `Portal` from
# ``tractor``'s internals?
feed: Feed|None = None
@property @property
def rt_shm(self) -> ShmArray: def rt_shm(self) -> ShmArray:
@ -156,7 +146,6 @@ class Flume(Struct):
# will get instead some kind of msg-compat version # will get instead some kind of msg-compat version
# that it can load. # that it can load.
msg.pop('stream') msg.pop('stream')
msg.pop('feed')
msg.pop('_rt_shm') msg.pop('_rt_shm')
msg.pop('_hist_shm') msg.pop('_hist_shm')

View File

@ -294,6 +294,11 @@ def ldshm(
f'Something is wrong with time period for {shm}:\n{times}' f'Something is wrong with time period for {shm}:\n{times}'
) )
period_s: float = float(max(d1, d2, med)) period_s: float = float(max(d1, d2, med))
log.info(
f'Processing shm buffer:\n'
f' file: {shmfile.name}\n'
f' period: {period_s}s\n'
)
null_segs: tuple = tsp.get_null_segs( null_segs: tuple = tsp.get_null_segs(
frame=shm.array, frame=shm.array,

View File

@ -276,14 +276,41 @@ def get_null_segs(
absi_zdiff: np.ndarray = np.diff(absi_zeros) absi_zdiff: np.ndarray = np.diff(absi_zeros)
if zero_t.size < 2: if zero_t.size < 2:
try: idx: int = zero_t['index'][0]
breakpoint() idx_before: int = idx - 1
except RuntimeError: idx_after: int = idx + 1
# XXX, if greenback not active from index = frame['index']
# piker store ldshm cmd.. before_cond = idx_before <= index
log.exception( after_cond = index <= idx_after
"Can't debug single-sample null!\n" bars: np.ndarray = frame[
) before_cond
&
after_cond
]
time: np.ndarray = bars['time']
from pendulum import (
from_timestamp,
Interval,
)
gap: Interval = (
from_timestamp(time[-1])
-
from_timestamp(time[0])
)
log.warning(
f'Single OHLCV-bar null-segment detected??\n'
f'gap -> {gap}\n'
)
# ^^XXX, if you want to debug the above bar-gap^^
# try:
# breakpoint()
# except RuntimeError:
# # XXX, if greenback not active from
# # piker store ldshm cmd..
# log.exception(
# "Can't debug single-sample null!\n"
# )
return None return None

View File

@ -30,6 +30,11 @@ import tractor
from piker.data._formatters import BGM from piker.data._formatters import BGM
from piker.storage import log from piker.storage import log
from piker.toolz.profile import (
Profiler,
pg_profile_enabled,
ms_slower_then,
)
from piker.ui._style import get_fonts from piker.ui._style import get_fonts
if TYPE_CHECKING: if TYPE_CHECKING:
@ -92,12 +97,22 @@ async def markup_gaps(
# gap's duration. # gap's duration.
show_txt: bool = False, show_txt: bool = False,
# A/B comparison: render individual arrows alongside batch
# for visual comparison
show_individual_arrows: bool = False,
) -> dict[int, dict]: ) -> dict[int, dict]:
''' '''
Remote annotate time-gaps in a dt-fielded ts (normally OHLC) Remote annotate time-gaps in a dt-fielded ts (normally OHLC)
with rectangles. with rectangles.
''' '''
profiler = Profiler(
msg=f'markup_gaps() for {gaps.height} gaps',
disabled=False,
ms_threshold=0.0,
)
# XXX: force chart redraw FIRST to ensure PlotItem coordinate # XXX: force chart redraw FIRST to ensure PlotItem coordinate
# system is properly initialized before we position annotations! # system is properly initialized before we position annotations!
# Without this, annotations may be misaligned on first creation # Without this, annotations may be misaligned on first creation
@ -106,6 +121,19 @@ async def markup_gaps(
fqme=fqme, fqme=fqme,
timeframe=timeframe, timeframe=timeframe,
) )
profiler('first `.redraw()` before annot creation')
log.info(
f'markup_gaps() called:\n'
f' fqme: {fqme}\n'
f' timeframe: {timeframe}s\n'
f' gaps.height: {gaps.height}\n'
)
# collect all annotation specs for batch submission
rect_specs: list[dict] = []
arrow_specs: list[dict] = []
text_specs: list[dict] = []
aids: dict[int] = {} aids: dict[int] = {}
for i in range(gaps.height): for i in range(gaps.height):
@ -217,56 +245,38 @@ async def markup_gaps(
# 1: 'wine', # down-gap # 1: 'wine', # down-gap
# }[sgn] # }[sgn]
rect_kwargs: dict[str, Any] = dict( # collect rect spec (no fqme/timeframe, added by batch
fqme=fqme, # API)
timeframe=timeframe, rect_spec: dict[str, Any] = dict(
meth='set_view_pos',
start_pos=lc, start_pos=lc,
end_pos=ro, end_pos=ro,
color=color, color=color,
update_label=False,
start_time=start_time, start_time=start_time,
end_time=end_time, end_time=end_time,
) )
rect_specs.append(rect_spec)
# add up/down rects
aid: int|None = await actl.add_rect(**rect_kwargs)
if aid is None:
log.error(
f'Failed to add rect for,\n'
f'{rect_kwargs!r}\n'
f'\n'
f'Skipping to next gap!\n'
)
continue
assert aid
aids[aid] = rect_kwargs
direction: str = ( direction: str = (
'down' if down_gap 'down' if down_gap
else 'up' else 'up'
) )
# TODO! mk this a `msgspec.Struct` which we deserialize
# on the server side! # collect arrow spec
# XXX: send timestamp for server-side index lookup
# to ensure alignment with current shm state
gap_time: float = row['time'][0] gap_time: float = row['time'][0]
arrow_kwargs: dict[str, Any] = dict( arrow_spec: dict[str, Any] = dict(
fqme=fqme,
timeframe=timeframe,
x=iend, # fallback if timestamp lookup fails x=iend, # fallback if timestamp lookup fails
y=cls, y=cls,
time=gap_time, # for server-side index lookup time=gap_time, # for server-side index lookup
color=color, color=color,
alpha=169, alpha=169,
pointing=direction, pointing=direction,
# TODO: expose these as params to markup_gaps()?
headLen=10, headLen=10,
headWidth=2.222, headWidth=2.222,
pxMode=True, pxMode=True,
) )
arrow_specs.append(arrow_spec)
aid: int = await actl.add_arrow(
**arrow_kwargs
)
# add duration label to RHS of arrow # add duration label to RHS of arrow
if up_gap: if up_gap:
@ -278,15 +288,12 @@ async def markup_gaps(
assert flat assert flat
anchor = (0, 0) # up from bottom anchor = (0, 0) # up from bottom
# use a slightly smaller font for gap label txt. # collect text spec if enabled
font, small_font = get_fonts()
font_size: int = small_font.px_size - 1
assert isinstance(font_size, int)
if show_txt: if show_txt:
text_aid: int = await actl.add_text( font, small_font = get_fonts()
fqme=fqme, font_size: int = small_font.px_size - 1
timeframe=timeframe,
text_spec: dict[str, Any] = dict(
text=gap_label, text=gap_label,
x=iend + 1, # fallback if timestamp lookup fails x=iend + 1, # fallback if timestamp lookup fails
y=cls, y=cls,
@ -295,12 +302,46 @@ async def markup_gaps(
anchor=anchor, anchor=anchor,
font_size=font_size, font_size=font_size,
) )
aids[text_aid] = {'text': gap_label} text_specs.append(text_spec)
# tell chart to redraw all its # submit all annotations in single batch IPC msg
# graphics view layers Bo log.info(
f'Submitting batch annotations:\n'
f' rects: {len(rect_specs)}\n'
f' arrows: {len(arrow_specs)}\n'
f' texts: {len(text_specs)}\n'
)
profiler('built all annotation specs')
result: dict[str, list[int]] = await actl.add_batch(
fqme=fqme,
timeframe=timeframe,
rects=rect_specs,
arrows=arrow_specs,
texts=text_specs,
show_individual_arrows=show_individual_arrows,
)
profiler('batch `.add_batch()` IPC call complete')
# build aids dict from batch results
for aid in result['rects']:
aids[aid] = {'type': 'rect'}
for aid in result['arrows']:
aids[aid] = {'type': 'arrow'}
for aid in result['texts']:
aids[aid] = {'type': 'text'}
log.info(
f'Batch submission complete: {len(aids)} annotation(s) '
f'created'
)
profiler('built aids result dict')
# tell chart to redraw all its graphics view layers
await actl.redraw( await actl.redraw(
fqme=fqme, fqme=fqme,
timeframe=timeframe, timeframe=timeframe,
) )
profiler('final `.redraw()` after annot creation')
return aids return aids

View File

@ -24,8 +24,11 @@ from pyqtgraph import (
Point, Point,
functions as fn, functions as fn,
Color, Color,
GraphicsObject,
) )
from pyqtgraph.Qt import internals
import numpy as np import numpy as np
import pyqtgraph as pg
from piker.ui.qt import ( from piker.ui.qt import (
QtCore, QtCore,
@ -35,6 +38,10 @@ from piker.ui.qt import (
QRectF, QRectF,
QGraphicsPathItem, QGraphicsPathItem,
) )
from piker.ui._style import hcolor
from piker.log import get_logger
log = get_logger(__name__)
def mk_marker_path( def mk_marker_path(
@ -104,7 +111,7 @@ def mk_marker_path(
class LevelMarker(QGraphicsPathItem): class LevelMarker(QGraphicsPathItem):
''' '''
An arrow marker path graphich which redraws itself An arrow marker path graphic which redraws itself
to the specified view coordinate level on each paint cycle. to the specified view coordinate level on each paint cycle.
''' '''
@ -251,9 +258,9 @@ def qgo_draw_markers(
) -> float: ) -> float:
''' '''
Paint markers in ``pg.GraphicsItem`` style by first Paint markers in ``pg.GraphicsItem`` style by first removing the
removing the view transform for the painter, drawing the markers view transform for the painter, drawing the markers in scene
in scene coords, then restoring the view coords. coords, then restoring the view coords.
''' '''
# paint markers in native coordinate system # paint markers in native coordinate system
@ -295,3 +302,449 @@ def qgo_draw_markers(
p.setTransform(orig_tr) p.setTransform(orig_tr)
return max(sizes) return max(sizes)
class GapAnnotations(GraphicsObject):
'''
Batch-rendered gap annotations using Qt's efficient drawing
APIs.
Instead of creating individual `QGraphicsItem` instances per
gap (which is very slow for 1000+ gaps), this class stores all
gap rectangles and arrows in numpy-backed arrays and renders
them in single batch paint calls.
Performance: ~1000x faster than individual items for large gap
counts.
Based on patterns from:
- `pyqtgraph.BarGraphItem` (batch rect rendering)
- `pyqtgraph.ScatterPlotItem` (fragment rendering)
- `piker.ui._curve.FlowGraphic` (single path pattern)
'''
def __init__(
self,
gap_specs: list[dict],
array: np.ndarray|None = None,
color: str = 'dad_blue',
alpha: int = 169,
arrow_size: float = 10.0,
fqme: str|None = None,
timeframe: float|None = None,
) -> None:
'''
gap_specs: list of dicts with keys:
- start_pos: (x, y) tuple for left corner of rect
- end_pos: (x, y) tuple for right corner of rect
- arrow_x: x position for arrow
- arrow_y: y position for arrow
- pointing: 'up' or 'down' for arrow direction
- start_time: (optional) timestamp for repositioning
- end_time: (optional) timestamp for repositioning
array: optional OHLC numpy array for repositioning on
backfill updates (when abs-index changes)
fqme: symbol name for these gaps (for logging/debugging)
timeframe: period in seconds that these gaps were
detected on (used to skip reposition when
called with wrong timeframe's array)
'''
super().__init__()
self._gap_specs = gap_specs
self._array = array
self._fqme = fqme
self._timeframe = timeframe
n_gaps = len(gap_specs)
# shared pen/brush matching original SelectRect/ArrowItem style
base_color = pg.mkColor(hcolor(color))
# rect pen: base color, fully opaque for outline
self._rect_pen = pg.mkPen(base_color, width=1)
# rect brush: base color with alpha=66 (SelectRect default)
rect_fill = pg.mkColor(hcolor(color))
rect_fill.setAlpha(66)
self._rect_brush = pg.functions.mkBrush(rect_fill)
# arrow pen: same as rects
self._arrow_pen = pg.mkPen(base_color, width=1)
# arrow brush: base color with user-specified alpha (default 169)
arrow_fill = pg.mkColor(hcolor(color))
arrow_fill.setAlpha(alpha)
self._arrow_brush = pg.functions.mkBrush(arrow_fill)
# allocate rect array using Qt's efficient storage
self._rectarray = internals.PrimitiveArray(
QtCore.QRectF,
4,
)
self._rectarray.resize(n_gaps)
rect_memory = self._rectarray.ndarray()
# fill rect array from gap specs
for (
i,
spec,
) in enumerate(gap_specs):
(
start_x,
start_y,
) = spec['start_pos']
(
end_x,
end_y,
) = spec['end_pos']
# QRectF expects (x, y, width, height)
rect_memory[i, 0] = start_x
rect_memory[i, 1] = min(start_y, end_y)
rect_memory[i, 2] = end_x - start_x
rect_memory[i, 3] = abs(end_y - start_y)
# build single QPainterPath for all arrows
self._arrow_path = QtGui.QPainterPath()
self._arrow_size = arrow_size
for spec in gap_specs:
arrow_x = spec['arrow_x']
arrow_y = spec['arrow_y']
pointing = spec['pointing']
# create arrow polygon
if pointing == 'down':
# arrow points downward
arrow_poly = QtGui.QPolygonF([
QPointF(arrow_x, arrow_y), # tip
QPointF(
arrow_x - arrow_size/2,
arrow_y - arrow_size,
), # left
QPointF(
arrow_x + arrow_size/2,
arrow_y - arrow_size,
), # right
])
else: # up
# arrow points upward
arrow_poly = QtGui.QPolygonF([
QPointF(arrow_x, arrow_y), # tip
QPointF(
arrow_x - arrow_size/2,
arrow_y + arrow_size,
), # left
QPointF(
arrow_x + arrow_size/2,
arrow_y + arrow_size,
), # right
])
self._arrow_path.addPolygon(arrow_poly)
self._arrow_path.closeSubpath()
# cache bounding rect
self._br: QRectF|None = None
def boundingRect(self) -> QRectF:
'''
Compute bounding rect from rect array and arrow path.
'''
if self._br is not None:
return self._br
# get rect bounds
rect_memory = self._rectarray.ndarray()
if len(rect_memory) == 0:
self._br = QRectF()
return self._br
x_min = rect_memory[:, 0].min()
y_min = rect_memory[:, 1].min()
x_max = (rect_memory[:, 0] + rect_memory[:, 2]).max()
y_max = (rect_memory[:, 1] + rect_memory[:, 3]).max()
# expand for arrow path
arrow_br = self._arrow_path.boundingRect()
x_min = min(x_min, arrow_br.left())
y_min = min(y_min, arrow_br.top())
x_max = max(x_max, arrow_br.right())
y_max = max(y_max, arrow_br.bottom())
self._br = QRectF(
x_min,
y_min,
x_max - x_min,
y_max - y_min,
)
return self._br
def paint(
self,
p: QtGui.QPainter,
opt: QtWidgets.QStyleOptionGraphicsItem,
w: QtWidgets.QWidget,
) -> None:
'''
Batch render all rects and arrows in minimal paint calls.
'''
# draw all rects in single batch call (data coordinates)
p.setPen(self._rect_pen)
p.setBrush(self._rect_brush)
drawargs = self._rectarray.drawargs()
p.drawRects(*drawargs)
# draw arrows in scene/pixel coordinates so they maintain
# size regardless of zoom level
orig_tr = p.transform()
p.resetTransform()
# rebuild arrow path in scene coordinates
arrow_path_scene = QtGui.QPainterPath()
# arrow geometry matching pg.ArrowItem defaults
# headLen=10, headWidth=2.222
# headWidth is the half-width (center to edge distance)
head_len = self._arrow_size
head_width = head_len * 0.2222 # 2.222 at size=10
for spec in self._gap_specs:
if 'arrow_x' not in spec:
continue
arrow_x = spec['arrow_x']
arrow_y = spec['arrow_y']
pointing = spec['pointing']
# transform data coords to scene coords
scene_pt = orig_tr.map(QPointF(arrow_x, arrow_y))
sx = scene_pt.x()
sy = scene_pt.y()
# create arrow polygon in scene/pixel coords
# matching pg.ArrowItem geometry but rotated for up/down
if pointing == 'down':
# tip points downward (negative y direction)
arrow_poly = QtGui.QPolygonF([
QPointF(sx, sy), # tip
QPointF(
sx - head_width,
sy - head_len,
), # left base
QPointF(
sx + head_width,
sy - head_len,
), # right base
])
else: # up
# tip points upward (positive y direction)
arrow_poly = QtGui.QPolygonF([
QPointF(sx, sy), # tip
QPointF(
sx - head_width,
sy + head_len,
), # left base
QPointF(
sx + head_width,
sy + head_len,
), # right base
])
arrow_path_scene.addPolygon(arrow_poly)
arrow_path_scene.closeSubpath()
p.setPen(self._arrow_pen)
p.setBrush(self._arrow_brush)
p.drawPath(arrow_path_scene)
# restore original transform
p.setTransform(orig_tr)
def reposition(
self,
array: np.ndarray|None = None,
fqme: str|None = None,
timeframe: float|None = None,
) -> None:
'''
Reposition all annotations based on timestamps.
Used when viz is updated (eg during backfill) and abs-index
range changes - we need to lookup new indices from timestamps.
'''
# skip reposition if timeframe doesn't match
# (e.g., 1s gaps being repositioned with 60s array)
if (
timeframe is not None
and
self._timeframe is not None
and
timeframe != self._timeframe
):
log.debug(
f'Skipping reposition for {self._fqme} gaps:\n'
f' gap timeframe: {self._timeframe}s\n'
f' array timeframe: {timeframe}s\n'
)
return
if array is None:
array = self._array
if array is None:
log.warning(
'GapAnnotations.reposition() called but no array '
'provided'
)
return
# collect all unique timestamps we need to lookup
timestamps: set[float] = set()
for spec in self._gap_specs:
if spec.get('start_time') is not None:
timestamps.add(spec['start_time'])
if spec.get('end_time') is not None:
timestamps.add(spec['end_time'])
if spec.get('time') is not None:
timestamps.add(spec['time'])
# vectorized timestamp -> row lookup using binary search
time_to_row: dict[float, dict] = {}
if timestamps:
import numpy as np
time_arr = array['time']
ts_array = np.array(list(timestamps))
search_indices = np.searchsorted(
time_arr,
ts_array,
)
# vectorized bounds check and exact match verification
valid_mask = (
(search_indices < len(array))
& (time_arr[search_indices] == ts_array)
)
valid_indices = search_indices[valid_mask]
valid_timestamps = ts_array[valid_mask]
matched_rows = array[valid_indices]
time_to_row = {
float(ts): {
'index': float(row['index']),
'open': float(row['open']),
'close': float(row['close']),
}
for ts, row in zip(
valid_timestamps,
matched_rows,
)
}
# rebuild rect array from gap specs with new indices
rect_memory = self._rectarray.ndarray()
for (
i,
spec,
) in enumerate(self._gap_specs):
start_time = spec.get('start_time')
end_time = spec.get('end_time')
if (
start_time is None
or end_time is None
):
continue
start_row = time_to_row.get(start_time)
end_row = time_to_row.get(end_time)
if (
start_row is None
or end_row is None
):
log.warning(
f'Timestamp lookup failed for gap[{i}] during '
f'reposition:\n'
f' fqme: {fqme}\n'
f' timeframe: {timeframe}s\n'
f' start_time: {start_time}\n'
f' end_time: {end_time}\n'
f' array time range: '
f'{array["time"][0]} -> {array["time"][-1]}\n'
)
continue
start_idx = start_row['index']
end_idx = end_row['index']
start_close = start_row['close']
end_open = end_row['open']
from_idx: float = 0.16 - 0.06
start_x = start_idx + 1 - from_idx
end_x = end_idx + from_idx
# update rect in array
rect_memory[i, 0] = start_x
rect_memory[i, 1] = min(start_close, end_open)
rect_memory[i, 2] = end_x - start_x
rect_memory[i, 3] = abs(end_open - start_close)
# rebuild arrow path with new indices
self._arrow_path.clear()
for spec in self._gap_specs:
time_val = spec.get('time')
if time_val is None:
continue
arrow_row = time_to_row.get(time_val)
if arrow_row is None:
continue
arrow_x = arrow_row['index']
arrow_y = arrow_row['close']
pointing = spec['pointing']
# create arrow polygon
if pointing == 'down':
arrow_poly = QtGui.QPolygonF([
QPointF(arrow_x, arrow_y),
QPointF(
arrow_x - self._arrow_size/2,
arrow_y - self._arrow_size,
),
QPointF(
arrow_x + self._arrow_size/2,
arrow_y - self._arrow_size,
),
])
else: # up
arrow_poly = QtGui.QPolygonF([
QPointF(arrow_x, arrow_y),
QPointF(
arrow_x - self._arrow_size/2,
arrow_y + self._arrow_size,
),
QPointF(
arrow_x + self._arrow_size/2,
arrow_y + self._arrow_size,
),
])
self._arrow_path.addPolygon(arrow_poly)
self._arrow_path.closeSubpath()
# invalidate bounding rect cache
self._br = None
self.prepareGeometryChange()
self.update()

View File

@ -168,7 +168,7 @@ class ArrowEditor(Struct):
''' '''
uid: str = arrow._uid uid: str = arrow._uid
arrows: list[pg.ArrowItem] = self._arrows[uid] arrows: list[pg.ArrowItem] = self._arrows[uid]
log.info( log.debug(
f'Removing arrow from views\n' f'Removing arrow from views\n'
f'uid: {uid!r}\n' f'uid: {uid!r}\n'
f'{arrow!r}\n' f'{arrow!r}\n'
@ -286,7 +286,9 @@ class LineEditor(Struct):
for line in lines: for line in lines:
line.show_labels() line.show_labels()
line.hide_markers() line.hide_markers()
log.debug(f'Level active for level: {line.value()}') log.debug(
f'Line active @ level: {line.value()!r}'
)
# TODO: other flashy things to indicate the order is active # TODO: other flashy things to indicate the order is active
return lines return lines
@ -329,7 +331,11 @@ class LineEditor(Struct):
if line in hovered: if line in hovered:
hovered.remove(line) hovered.remove(line)
log.debug(f'deleting {line} with oid: {uuid}') log.debug(
f'Deleting level-line\n'
f'line: {line!r}\n'
f'oid: {uuid!r}\n'
)
line.delete() line.delete()
# make sure the xhair doesn't get left off # make sure the xhair doesn't get left off
@ -337,7 +343,11 @@ class LineEditor(Struct):
cursor.show_xhair() cursor.show_xhair()
else: else:
log.warning(f'Could not find line for {line}') log.warning(
f'Could not find line for removal ??\n'
f'\n'
f'{line!r}\n'
)
return lines return lines
@ -569,11 +579,11 @@ class SelectRect(QtWidgets.QGraphicsRectItem):
if update_label: if update_label:
self.init_label(view_rect) self.init_label(view_rect)
print( log.debug(
'SelectRect modify:\n' f'SelectRect modify,\n'
f'QRectF: {view_rect}\n' f'QRectF: {view_rect}\n'
f'start_pos: {start_pos}\n' f'start_pos: {start_pos!r}\n'
f'end_pos: {end_pos}\n' f'end_pos: {end_pos!r}\n'
) )
self.show() self.show()
@ -640,8 +650,11 @@ class SelectRect(QtWidgets.QGraphicsRectItem):
dmn=dmn, dmn=dmn,
)) ))
# print(f'x2, y2: {(x2, y2)}') # tracing
# print(f'xmn, ymn: {(xmn, ymx)}') # log.info(
# f'x2, y2: {(x2, y2)}\n'
# f'xmn, ymn: {(xmn, ymx)}\n'
# )
label_anchor = Point( label_anchor = Point(
xmx + 2, xmx + 2,

View File

@ -38,7 +38,6 @@ from piker.ui.qt import (
QtGui, QtGui,
QGraphicsPathItem, QGraphicsPathItem,
QStyleOptionGraphicsItem, QStyleOptionGraphicsItem,
QGraphicsItem,
QGraphicsScene, QGraphicsScene,
QWidget, QWidget,
QPointF, QPointF,

View File

@ -22,6 +22,7 @@ a chart from some other actor.
from __future__ import annotations from __future__ import annotations
from contextlib import ( from contextlib import (
asynccontextmanager as acm, asynccontextmanager as acm,
contextmanager as cm,
AsyncExitStack, AsyncExitStack,
) )
from functools import partial from functools import partial
@ -46,6 +47,7 @@ from piker.log import get_logger
from piker.types import Struct from piker.types import Struct
from piker.service import find_service from piker.service import find_service
from piker.brokers import SymbolNotFound from piker.brokers import SymbolNotFound
from piker.toolz import Profiler
from piker.ui.qt import ( from piker.ui.qt import (
QGraphicsItem, QGraphicsItem,
) )
@ -98,6 +100,8 @@ def rm_annot(
annot: ArrowEditor|SelectRect|pg.TextItem annot: ArrowEditor|SelectRect|pg.TextItem
) -> bool: ) -> bool:
global _editors global _editors
from piker.ui._annotate import GapAnnotations
match annot: match annot:
case pg.ArrowItem(): case pg.ArrowItem():
editor = _editors[annot._uid] editor = _editors[annot._uid]
@ -122,9 +126,35 @@ def rm_annot(
scene.removeItem(annot) scene.removeItem(annot)
return True return True
case GapAnnotations():
scene = annot.scene()
if scene:
scene.removeItem(annot)
return True
return False return False
@cm
def no_qt_updates(*items):
'''
Disable Qt widget/item updates during context to batch
render operations and only trigger single repaint on exit.
Accepts both QWidgets and QGraphicsItems.
'''
for item in items:
if hasattr(item, 'setUpdatesEnabled'):
item.setUpdatesEnabled(False)
try:
yield
finally:
for item in items:
if hasattr(item, 'setUpdatesEnabled'):
item.setUpdatesEnabled(True)
async def serve_rc_annots( async def serve_rc_annots(
ipc_key: str, ipc_key: str,
annot_req_stream: MsgStream, annot_req_stream: MsgStream,
@ -429,6 +459,333 @@ async def serve_rc_annots(
aids.add(aid) aids.add(aid)
await annot_req_stream.send(aid) await annot_req_stream.send(aid)
case {
'cmd': 'batch',
'fqme': fqme,
'timeframe': timeframe,
'rects': list(rect_specs),
'arrows': list(arrow_specs),
'texts': list(text_specs),
'show_individual_arrows': bool(show_individual_arrows),
}:
# batch submission handler - process multiple
# annotations in single IPC round-trip
ds: DisplayState = _dss[fqme]
try:
chart: ChartPlotWidget = {
60: ds.hist_chart,
1: ds.chart,
}[timeframe]
except KeyError:
msg: str = (
f'No chart for timeframe={timeframe}s, '
f'skipping batch annotation'
)
log.error(msg)
await annot_req_stream.send({'error': msg})
continue
cv: ChartView = chart.cv
viz: Viz = chart.get_viz(fqme)
shm = viz.shm
arr = shm.array
result: dict[str, list[int]] = {
'rects': [],
'arrows': [],
'texts': [],
}
profiler = Profiler(
msg=(
f'Batch annotate {len(rect_specs)} gaps '
f'on {fqme}@{timeframe}s'
),
disabled=False,
delayed=False,
)
aids_set: set[int] = ctxs[ipc_key][1]
# build unified gap_specs for GapAnnotations class
from piker.ui._annotate import GapAnnotations
gap_specs: list[dict] = []
n_gaps: int = max(
len(rect_specs),
len(arrow_specs),
)
profiler('setup batch annot creation')
# collect all unique timestamps for vectorized lookup
timestamps: list[float] = []
for rect_spec in rect_specs:
if start_time := rect_spec.get('start_time'):
timestamps.append(start_time)
if end_time := rect_spec.get('end_time'):
timestamps.append(end_time)
for arrow_spec in arrow_specs:
if time_val := arrow_spec.get('time'):
timestamps.append(time_val)
profiler('collect `timestamps: list` complet!')
# build timestamp -> row mapping using binary search
# O(m log n) instead of O(n*m) with np.isin
time_to_row: dict[float, dict] = {}
if timestamps:
import numpy as np
time_arr = arr['time']
ts_array = np.array(timestamps)
# binary search for each timestamp in sorted time array
search_indices = np.searchsorted(
time_arr,
ts_array,
)
profiler('`np.searchsorted()` complete!')
# vectorized bounds check and exact match verification
valid_mask = (
(search_indices < len(arr))
& (time_arr[search_indices] == ts_array)
)
# get all valid indices and timestamps
valid_indices = search_indices[valid_mask]
valid_timestamps = ts_array[valid_mask]
# use fancy indexing to get all rows at once
matched_rows = arr[valid_indices]
# extract fields to plain arrays BEFORE dict building
indices_arr = matched_rows['index'].astype(float)
opens_arr = matched_rows['open'].astype(float)
closes_arr = matched_rows['close'].astype(float)
profiler('extracted field arrays')
# build dict from plain arrays (much faster)
time_to_row: dict[float, dict] = {
float(ts): {
'index': idx,
'open': opn,
'close': cls,
}
for (
ts,
idx,
opn,
cls,
) in zip(
valid_timestamps,
indices_arr,
opens_arr,
closes_arr,
)
}
profiler('`time_to_row` creation complete!')
profiler(f'built timestamp lookup for {len(timestamps)} times')
# build gap_specs from rect+arrow specs
for i in range(n_gaps):
gap_spec: dict = {}
# get rect spec for this gap
if i < len(rect_specs):
rect_spec: dict = rect_specs[i].copy()
start_time = rect_spec.get('start_time')
end_time = rect_spec.get('end_time')
if (
start_time is not None
and end_time is not None
):
# lookup from pre-built mapping
start_row = time_to_row.get(start_time)
end_row = time_to_row.get(end_time)
if (
start_row is None
or end_row is None
):
log.warning(
f'Timestamp lookup failed for '
f'gap[{i}], skipping'
)
continue
start_idx = start_row['index']
end_idx = end_row['index']
start_close = start_row['close']
end_open = end_row['open']
from_idx: float = 0.16 - 0.06
gap_spec['start_pos'] = (
start_idx + 1 - from_idx,
start_close,
)
gap_spec['end_pos'] = (
end_idx + from_idx,
end_open,
)
gap_spec['start_time'] = start_time
gap_spec['end_time'] = end_time
gap_spec['color'] = rect_spec.get(
'color',
'dad_blue',
)
# get arrow spec for this gap
if i < len(arrow_specs):
arrow_spec: dict = arrow_specs[i].copy()
x: float = float(arrow_spec.get('x', 0))
y: float = float(arrow_spec.get('y', 0))
time_val: float|None = arrow_spec.get('time')
# timestamp-based index lookup (only for x, NOT y!)
# y is already set to the PREVIOUS bar's close
if time_val is not None:
arrow_row = time_to_row.get(time_val)
if arrow_row is not None:
x = arrow_row['index']
# NOTE: do NOT update y! it's the
# previous bar's close, not current
else:
log.warning(
f'Arrow timestamp {time_val} not '
f'found for gap[{i}], using x={x}'
)
gap_spec['arrow_x'] = x
gap_spec['arrow_y'] = y
gap_spec['time'] = time_val
gap_spec['pointing'] = arrow_spec.get(
'pointing',
'down',
)
gap_spec['alpha'] = arrow_spec.get('alpha', 169)
gap_specs.append(gap_spec)
profiler(f'built {len(gap_specs)} gap_specs')
# create single GapAnnotations item for all gaps
if gap_specs:
gaps_item = GapAnnotations(
gap_specs=gap_specs,
array=arr,
color=gap_specs[0].get('color', 'dad_blue'),
alpha=gap_specs[0].get('alpha', 169),
arrow_size=10.0,
fqme=fqme,
timeframe=timeframe,
)
chart.plotItem.addItem(gaps_item)
# register single item for repositioning
aid: int = id(gaps_item)
annots[aid] = gaps_item
aids_set.add(aid)
result['rects'].append(aid)
profiler(
f'created GapAnnotations item for {len(gap_specs)} '
f'gaps'
)
# A/B comparison: optionally create individual arrows
# alongside batch for visual comparison
if show_individual_arrows:
godw = chart.linked.godwidget
arrows: ArrowEditor = ArrowEditor(godw=godw)
for i, spec in enumerate(gap_specs):
if 'arrow_x' not in spec:
continue
aid_str: str = str(uuid4())
arrow: pg.ArrowItem = arrows.add(
plot=chart.plotItem,
uid=aid_str,
x=spec['arrow_x'],
y=spec['arrow_y'],
pointing=spec['pointing'],
color='bracket', # different color
alpha=spec.get('alpha', 169),
headLen=10.0,
headWidth=2.222,
pxMode=True,
)
arrow._abs_x = spec['arrow_x']
arrow._abs_y = spec['arrow_y']
annots[aid_str] = arrow
_editors[aid_str] = arrows
aids_set.add(aid_str)
result['arrows'].append(aid_str)
profiler(
f'created {len(gap_specs)} individual arrows '
f'for comparison'
)
# handle text items separately (less common, keep
# individual items)
n_texts: int = 0
for text_spec in text_specs:
kwargs: dict = text_spec.copy()
text: str = kwargs.pop('text')
x: float = float(kwargs.pop('x'))
y: float = float(kwargs.pop('y'))
time_val: float|None = kwargs.pop('time', None)
# timestamp-based index lookup
if time_val is not None:
matches = arr[arr['time'] == time_val]
if len(matches) > 0:
x = float(matches[0]['index'])
y = float(matches[0]['close'])
color = kwargs.pop('color', 'dad_blue')
anchor = kwargs.pop('anchor', (0, 1))
font_size = kwargs.pop('font_size', None)
text_item: pg.TextItem = pg.TextItem(
text,
color=hcolor(color),
anchor=anchor,
)
if font_size is None:
from ._style import get_fonts
font, font_small = get_fonts()
font_size = font_small.px_size - 1
qfont: QFont = text_item.textItem.font()
qfont.setPixelSize(font_size)
text_item.setFont(qfont)
text_item.setPos(float(x), float(y))
chart.plotItem.addItem(text_item)
text_item._abs_x = float(x)
text_item._abs_y = float(y)
aid: str = str(uuid4())
annots[aid] = text_item
aids_set.add(aid)
result['texts'].append(aid)
n_texts += 1
profiler(
f'created text annotations: {n_texts} texts'
)
profiler.finish()
await annot_req_stream.send(result)
case { case {
'cmd': 'remove', 'cmd': 'remove',
'aid': int(aid)|str(aid), 'aid': int(aid)|str(aid),
@ -471,10 +828,26 @@ async def serve_rc_annots(
# XXX: reposition all annotations to ensure they # XXX: reposition all annotations to ensure they
# stay aligned with viz data after reset (eg during # stay aligned with viz data after reset (eg during
# backfill when abs-index range changes) # backfill when abs-index range changes)
chart: ChartPlotWidget = {
60: ds.hist_chart,
1: ds.chart,
}[timeframe]
viz: Viz = chart.get_viz(fqme)
arr = viz.shm.array
n_repositioned: int = 0 n_repositioned: int = 0
for aid, annot in annots.items(): for aid, annot in annots.items():
# GapAnnotations batch items have .reposition()
if hasattr(annot, 'reposition'):
annot.reposition(
array=arr,
fqme=fqme,
timeframe=timeframe,
)
n_repositioned += 1
# arrows and text items use abs x,y coords # arrows and text items use abs x,y coords
if ( elif (
hasattr(annot, '_abs_x') hasattr(annot, '_abs_x')
and and
hasattr(annot, '_abs_y') hasattr(annot, '_abs_y')
@ -539,12 +912,21 @@ async def remote_annotate(
finally: finally:
# ensure all annots for this connection are deleted # ensure all annots for this connection are deleted
# on any final teardown # on any final teardown
profiler = Profiler(
msg=f'Annotation teardown for ctx {ctx.cid}',
disabled=False,
ms_threshold=0.0,
)
(_ctx, aids) = _ctxs[ctx.cid] (_ctx, aids) = _ctxs[ctx.cid]
assert _ctx is ctx assert _ctx is ctx
profiler(f'got {len(aids)} aids to remove')
for aid in aids: for aid in aids:
annot: QGraphicsItem = _annots[aid] annot: QGraphicsItem = _annots[aid]
assert rm_annot(annot) assert rm_annot(annot)
profiler(f'removed all {len(aids)} annotations')
class AnnotCtl(Struct): class AnnotCtl(Struct):
''' '''
@ -746,6 +1128,64 @@ class AnnotCtl(Struct):
) )
return aid return aid
async def add_batch(
self,
fqme: str,
timeframe: float,
rects: list[dict]|None = None,
arrows: list[dict]|None = None,
texts: list[dict]|None = None,
show_individual_arrows: bool = False,
from_acm: bool = False,
) -> dict[str, list[int]]:
'''
Batch submit multiple annotations in single IPC msg for
much faster remote annotation vs. per-annot round-trips.
Returns dict of annotation IDs:
{
'rects': [aid1, aid2, ...],
'arrows': [aid3, aid4, ...],
'texts': [aid5, aid6, ...],
}
'''
ipc: MsgStream = self._get_ipc(fqme)
with trio.fail_after(10):
await ipc.send({
'fqme': fqme,
'cmd': 'batch',
'timeframe': timeframe,
'rects': rects or [],
'arrows': arrows or [],
'texts': texts or [],
'show_individual_arrows': show_individual_arrows,
})
result: dict = await ipc.receive()
match result:
case {'error': str(msg)}:
log.error(msg)
return {
'rects': [],
'arrows': [],
'texts': [],
}
# register all AIDs with their IPC streams
for aid_list in result.values():
for aid in aid_list:
self._ipcs[aid] = ipc
if not from_acm:
self._annot_stack.push_async_callback(
partial(
self.remove,
aid,
)
)
return result
async def add_text( async def add_text(
self, self,
fqme: str, fqme: str,
@ -881,3 +1321,14 @@ async def open_annot_ctl(
_annot_stack=annots_stack, _annot_stack=annots_stack,
) )
yield client yield client
# client exited, measure teardown time
teardown_profiler = Profiler(
msg='Client AnnotCtl teardown',
disabled=False,
ms_threshold=0.0,
)
teardown_profiler('exiting annots_stack')
teardown_profiler('annots_stack exited')
teardown_profiler('exiting gather_contexts')

1263
poetry.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -52,7 +52,6 @@ dependencies = [
"bidict >=0.23.1", "bidict >=0.23.1",
"colorama >=0.4.6, <0.5.0", "colorama >=0.4.6, <0.5.0",
"colorlog >=6.7.0, <7.0.0", "colorlog >=6.7.0, <7.0.0",
"ib-insync >=0.9.86, <0.10.0",
"numpy>=2.0", "numpy>=2.0",
"polars >=0.20.6", "polars >=0.20.6",
"polars-fuzzy-match>=0.1.5", "polars-fuzzy-match>=0.1.5",
@ -76,6 +75,8 @@ dependencies = [
"numba>=0.61.0", "numba>=0.61.0",
"pyvnc", "pyvnc",
"exchange-calendars>=4.13.1", "exchange-calendars>=4.13.1",
"ib-async>=2.1.0",
"aeventkit>=2.1.0", # XXX, imports as eventkit?
] ]
# ------ dependencies ------ # ------ dependencies ------
# NOTE, by default we ship only a "headless" deps set bc # NOTE, by default we ship only a "headless" deps set bc
@ -105,7 +106,7 @@ default-groups = [
[dependency-groups] [dependency-groups]
uis = [ uis = [
"pyqtgraph", "pyqtgraph >= 0.14.0",
"qdarkstyle >=3.0.2, <4.0.0", "qdarkstyle >=3.0.2, <4.0.0",
"pyqt6 >=6.7.0, <7.0.0", "pyqt6 >=6.7.0, <7.0.0",
@ -192,9 +193,12 @@ include = ["piker"]
[tool.uv.sources] [tool.uv.sources]
pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" } tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" }
pyvnc = { git = "https://github.com/regulad/pyvnc.git" } pyvnc = { git = "https://github.com/regulad/pyvnc.git" }
# pyqtgraph = { git = "https://github.com/pyqtgraph/pyqtgraph.git", branch = 'master' }
# pyqtgraph = { path = '../pyqtgraph', editable = true }
# ?TODO, resync our fork?
# pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
# to get fancy next-cmd/suggestion feats prior to 0.22.2 B) # to get fancy next-cmd/suggestion feats prior to 0.22.2 B)
# https://github.com/xonsh/xonsh/pull/6037 # https://github.com/xonsh/xonsh/pull/6037

53
uv.lock
View File

@ -10,6 +10,18 @@ resolution-markers = [
"python_full_version < '3.14' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version < '3.14' and sys_platform != 'emscripten' and sys_platform != 'win32'",
] ]
[[package]]
name = "aeventkit"
version = "2.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "numpy" },
]
sdist = { url = "https://files.pythonhosted.org/packages/5c/8c/c08db1a1910f8d04ec6a524de522edd0bac181bdf94dbb01183f7685cd77/aeventkit-2.1.0.tar.gz", hash = "sha256:4e7d81bb0a67227121da50a23e19e5bbf13eded541a9f4857eeb6b7b857b738a", size = 24703, upload-time = "2025-06-22T15:54:03.961Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/8d/8c/2a4b912b1afa201b25bdd0f5bccf96d5a8b5dccb6131316a8dd2d9cabcc1/aeventkit-2.1.0-py3-none-any.whl", hash = "sha256:962d43f79e731ac43527f2d0defeed118e6dbaa85f1487f5667540ebb8f00729", size = 26678, upload-time = "2025-06-22T15:54:02.141Z" },
]
[[package]] [[package]]
name = "aiodns" name = "aiodns"
version = "3.6.0" version = "3.6.0"
@ -396,18 +408,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/56/01/6f77d042b83260ef9ed73ea9647dfa0ef8414eba0a3fc57a509a088ad39b/elasticsearch-8.19.2-py3-none-any.whl", hash = "sha256:c16ba20c4c76cf6952e836dae7f4e724e00ba7bf31b94b79472b873683accdd4", size = 949706, upload-time = "2025-10-28T16:36:41.003Z" }, { url = "https://files.pythonhosted.org/packages/56/01/6f77d042b83260ef9ed73ea9647dfa0ef8414eba0a3fc57a509a088ad39b/elasticsearch-8.19.2-py3-none-any.whl", hash = "sha256:c16ba20c4c76cf6952e836dae7f4e724e00ba7bf31b94b79472b873683accdd4", size = 949706, upload-time = "2025-10-28T16:36:41.003Z" },
] ]
[[package]]
name = "eventkit"
version = "1.0.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "numpy" },
]
sdist = { url = "https://files.pythonhosted.org/packages/16/1e/0fac4e45d71ace143a2673ec642701c3cd16f833a0e77a57fa6a40472696/eventkit-1.0.3.tar.gz", hash = "sha256:99497f6f3c638a50ff7616f2f8cd887b18bbff3765dc1bd8681554db1467c933", size = 28320, upload-time = "2023-12-11T11:41:35.339Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/93/d9/7497d650b69b420e1a913329a843e16c715dac883750679240ef00a921e2/eventkit-1.0.3-py3-none-any.whl", hash = "sha256:0e199527a89aff9d195b9671ad45d2cc9f79ecda0900de8ecfb4c864d67ad6a2", size = 31837, upload-time = "2023-12-11T11:41:33.358Z" },
]
[[package]] [[package]]
name = "exceptiongroup" name = "exceptiongroup"
version = "1.3.1" version = "1.3.1"
@ -630,16 +630,17 @@ wheels = [
] ]
[[package]] [[package]]
name = "ib-insync" name = "ib-async"
version = "0.9.86" version = "2.1.0"
source = { registry = "https://pypi.org/simple" } source = { registry = "https://pypi.org/simple" }
dependencies = [ dependencies = [
{ name = "eventkit" }, { name = "aeventkit" },
{ name = "nest-asyncio" }, { name = "nest-asyncio" },
{ name = "tzdata" },
] ]
sdist = { url = "https://files.pythonhosted.org/packages/55/bb/733d5c81c8c2f54e90898afc7ff3a99f4d53619e6917c848833f9cc1ab56/ib_insync-0.9.86.tar.gz", hash = "sha256:73af602ca2463f260999970c5bd937b1c4325e383686eff301743a4de08d381e", size = 69859, upload-time = "2023-07-02T12:43:31.968Z" } sdist = { url = "https://files.pythonhosted.org/packages/30/4d/dfc1da8224c3ffcdcd668da7283c4e5f14239a07f83ea66af99700296fc3/ib_async-2.1.0.tar.gz", hash = "sha256:6a03a87d6c06acacb0217a5bea60a8a168ecd5b5a7e86e1c73678d5b48cbc796", size = 87678, upload-time = "2025-12-08T01:42:32.004Z" }
wheels = [ wheels = [
{ url = "https://files.pythonhosted.org/packages/8f/f3/28ea87be30570f4d6b8fd24380d12fa74e59467ee003755e76aeb29082b8/ib_insync-0.9.86-py3-none-any.whl", hash = "sha256:a61fbe56ff405d93d211dad8238d7300de76dd6399eafc04c320470edec9a4a4", size = 72980, upload-time = "2023-07-02T12:43:29.928Z" }, { url = "https://files.pythonhosted.org/packages/80/e7/8f33801788c66f15e9250957ff7f53a8000843f79af1a3ed7a96def0e96b/ib_async-2.1.0-py3-none-any.whl", hash = "sha256:f6d8b991bdbd6dd38e700c61b3dced06ebe0f14be4e5263e2ef10ba10b88d434", size = 88876, upload-time = "2025-12-08T01:42:30.883Z" },
] ]
[[package]] [[package]]
@ -1099,6 +1100,7 @@ name = "piker"
version = "0.1.0a0.dev0" version = "0.1.0a0.dev0"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "aeventkit" },
{ name = "async-generator" }, { name = "async-generator" },
{ name = "attrs" }, { name = "attrs" },
{ name = "bidict" }, { name = "bidict" },
@ -1107,7 +1109,7 @@ dependencies = [
{ name = "cryptofeed" }, { name = "cryptofeed" },
{ name = "exchange-calendars" }, { name = "exchange-calendars" },
{ name = "httpx" }, { name = "httpx" },
{ name = "ib-insync" }, { name = "ib-async" },
{ name = "msgspec" }, { name = "msgspec" },
{ name = "numba" }, { name = "numba" },
{ name = "numpy" }, { name = "numpy" },
@ -1175,6 +1177,7 @@ uis = [
[package.metadata] [package.metadata]
requires-dist = [ requires-dist = [
{ name = "aeventkit", specifier = ">=2.1.0" },
{ name = "async-generator", specifier = ">=1.10,<2.0.0" }, { name = "async-generator", specifier = ">=1.10,<2.0.0" },
{ name = "attrs", specifier = ">=23.1.0,<24.0.0" }, { name = "attrs", specifier = ">=23.1.0,<24.0.0" },
{ name = "bidict", specifier = ">=0.23.1" }, { name = "bidict", specifier = ">=0.23.1" },
@ -1183,7 +1186,7 @@ requires-dist = [
{ name = "cryptofeed", specifier = ">=2.4.0,<3.0.0" }, { name = "cryptofeed", specifier = ">=2.4.0,<3.0.0" },
{ name = "exchange-calendars", specifier = ">=4.13.1" }, { name = "exchange-calendars", specifier = ">=4.13.1" },
{ name = "httpx", specifier = ">=0.27.0,<0.28.0" }, { name = "httpx", specifier = ">=0.27.0,<0.28.0" },
{ name = "ib-insync", specifier = ">=0.9.86,<0.10.0" }, { name = "ib-async", specifier = ">=2.1.0" },
{ name = "msgspec", specifier = ">=0.19.0,<0.20" }, { name = "msgspec", specifier = ">=0.19.0,<0.20" },
{ name = "numba", specifier = ">=0.61.0" }, { name = "numba", specifier = ">=0.61.0" },
{ name = "numpy", specifier = ">=2.0" }, { name = "numpy", specifier = ">=2.0" },
@ -1218,7 +1221,7 @@ dev = [
{ name = "prompt-toolkit", specifier = "==3.0.40" }, { name = "prompt-toolkit", specifier = "==3.0.40" },
{ name = "pyperclip", specifier = ">=1.9.0" }, { name = "pyperclip", specifier = ">=1.9.0" },
{ name = "pyqt6", specifier = ">=6.7.0,<7.0.0" }, { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
{ name = "pyqtgraph", git = "https://github.com/pikers/pyqtgraph.git" }, { name = "pyqtgraph", specifier = ">=0.14.0" },
{ name = "pytest" }, { name = "pytest" },
{ name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" }, { name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" },
{ name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" }, { name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" },
@ -1236,7 +1239,7 @@ repl = [
testing = [{ name = "pytest" }] testing = [{ name = "pytest" }]
uis = [ uis = [
{ name = "pyqt6", specifier = ">=6.7.0,<7.0.0" }, { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
{ name = "pyqtgraph", git = "https://github.com/pikers/pyqtgraph.git" }, { name = "pyqtgraph", specifier = ">=0.14.0" },
{ name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" }, { name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" },
{ name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" }, { name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" },
] ]
@ -1603,11 +1606,15 @@ wheels = [
[[package]] [[package]]
name = "pyqtgraph" name = "pyqtgraph"
version = "0.12.3" version = "0.14.0"
source = { git = "https://github.com/pikers/pyqtgraph.git#373f9561ea8ec4fef9b4e8bdcdd4bbf372dd6512" } source = { registry = "https://pypi.org/simple" }
dependencies = [ dependencies = [
{ name = "colorama" },
{ name = "numpy" }, { name = "numpy" },
] ]
wheels = [
{ url = "https://files.pythonhosted.org/packages/32/36/4c242f81fdcbfa4fb62a5645f6af79191f4097a0577bd5460c24f19cc4ef/pyqtgraph-0.14.0-py3-none-any.whl", hash = "sha256:7abb7c3e17362add64f8711b474dffac5e7b0e9245abdf992e9a44119b7aa4f5", size = 1924755, upload-time = "2025-11-16T19:43:22.251Z" },
]
[[package]] [[package]]
name = "pyreadline3" name = "pyreadline3"